Tarkibga o'tish

Worker pool

Worker pool — ko‘p vazifani cheklangan sondagi ishchi goroutinelar yordamida bajarish usuli. Vazifalar navbatga qo‘yiladi, bo‘shagan worker esa navbatdagi vazifani oladi.

Masalan, 1000 ta faylni qayta ishlash kerak bo‘lsin. Har bir fayl uchun alohida goroutine yaratish mumkin, ammo bu bir paytda juda ko‘p fayl ochilishiga, xotira sarfi oshishiga yoki tashqi xizmat limitlari buzilishiga olib kelishi mumkin. Worker pool bir vaqtda nechta vazifa bajarilishini chegaralaydi.

Taksi xizmatini tasavvur qiling: mijozlar vazifalar navbati, haydovchilar esa workerlar. Mijozlar soni haydovchilar sonidan ko‘p bo‘lishi mumkin. Bo‘shagan haydovchi navbatdagi mijozga xizmat qiladi.

Tuzilishning asosiy qismlari

Oddiy worker pool to‘rtta qismdan iborat:

  • producer vazifalarni yaratadi va jobs kanaliga yuboradi;
  • bir nechta worker jobs kanalidan vazifa oladi;
  • workerlar natijani results kanaliga yuboradi;
  • coordinator workerlar tugashini kutadi va results kanalini yopadi.

jobs kanalini boshqa vazifa yuborilmasligini biladigan producer yopadi. results kanalini esa barcha worker tugaganini biladigan coordinator yopishi kerak. Bir nechta workerning o‘zi resultsni yopsa, boshqa worker yopilgan kanalga yuborib, panic keltirib chiqarishi mumkin.

Birinchi ishlaydigan misol

package main

import (
    "fmt"
    "sync"
)

type Result struct {
    JobID  int
    Worker int
    Value  int
}

func worker(id int, jobs <-chan int, results chan<- Result, wg *sync.WaitGroup) {
    defer wg.Done()

    for job := range jobs {
        results <- Result{
            JobID:  job,
            Worker: id,
            Value:  job * 2,
        }
    }
}

func main() {
    const jobCount = 5
    const workerCount = 3

    jobs := make(chan int, jobCount)
    results := make(chan Result, jobCount)

    var wg sync.WaitGroup
    wg.Add(workerCount)
    for id := 1; id <= workerCount; id++ {
        go worker(id, jobs, results, &wg)
    }

    go func() {
        for job := 1; job <= jobCount; job++ {
            jobs <- job
        }
        close(jobs)
    }()

    go func() {
        wg.Wait()
        close(results)
    }()

    for result := range results {
        fmt.Printf("ish=%d worker=%d natija=%d\n", result.JobID, result.Worker, result.Value)
    }
}

Natijaning mumkin bo‘lgan ko‘rinishi:

ish=1 worker=1 natija=2
ish=2 worker=2 natija=4
ish=3 worker=3 natija=6
ish=4 worker=1 natija=8
ish=5 worker=2 natija=10

Workerlar bir vaqtda ishlagani uchun qaysi ishni qaysi worker olishi va natijalar tartibi har ishga tushirishda o‘zgarishi mumkin. Muhim kafolat shuki, har bir yuborilgan ish bir marta olinadi va natijasi kanalga yuboriladi.

jobs <-chan int worker kanalni faqat o‘qishini bildiradi. results chan<- Result esa kanalga faqat yozishga ruxsat beradi. Yo‘nalishli kanallar kompilyatorga noto‘g‘ri amalni ushlashga yordam beradi.

for job := range jobs kanal yopilguncha qiymatlarni oladi. Producer close(jobs) qilgach, bufferda qolgan ishlar avval olinadi, keyin worker sikldan chiqadi. Har bir worker defer wg.Done() orqali tugaganini bildiradi.

Coordinator wg.Wait() bilan barcha workerlarni kutadi va shundan keyingina resultsni yopadi. main esa for result := range results yordamida kanal yopilguncha natijalarni qabul qiladi. Shu tartib deadlock va yopilgan kanalga yuborish xatolarining oldini oladi.

Buffer va qarshi bosim

Yuqoridagi jobs va results kanallari barcha qiymatlarni sig‘dira oladigan bufferga ega. Kichik misolda bu kodni tushunishni yengillashtiradi, ammo real loyihada navbatni cheksiz kattalashtirish kerak emas.

Buffer to‘lsa, producer yuborishda bloklanadi. Bu qarshi bosim (backpressure) deyiladi: workerlar ulgura olmayotganda yangi ishlar nazoratsiz yig‘ilmaydi. Buffersiz kanalda producer har bir ishni worker qabul qilguncha kutadi. Bufferli kanalda esa buffer to‘lguncha oldinga o‘ta oladi.

Buffer hajmi worker soni bilan bir xil bo‘lishi shart emas. Uni vazifalarning kelish tezligi, bajarilish vaqti va ruxsat etilgan xotira sarfiga qarab tanlash kerak.

Xato va bekor qilishni boshqarish

Real vazifa xato bilan tugashi mumkin. Natija tarkibida ish identifikatori, qiymat va xatoni birga uzatish foydali:

type Result struct {
    JobID int
    Value int
    Err   error
}

Birinchi jiddiy xatoda qolgan ishlarni to‘xtatish kerak bo‘lsa, workerlar context.Contextni kuzatishi mumkin. Quyidagi misol manfiy son uchrasa poolni bekor qiladi:

package main

import (
    "context"
    "errors"
    "fmt"
    "sync"
)

type Job struct {
    ID    int
    Value int
}

type Result struct {
    JobID int
    Value int
    Err   error
}

func process(job Job) (int, error) {
    if job.Value < 0 {
        return 0, errors.New("manfiy qiymatga ruxsat berilmaydi")
    }

    return job.Value * job.Value, nil
}

func worker(
    ctx context.Context,
    jobs <-chan Job,
    results chan<- Result,
    wg *sync.WaitGroup,
) {
    defer wg.Done()

    for {
        select {
        case <-ctx.Done():
            return
        case job, ok := <-jobs:
            if !ok {
                return
            }

            value, err := process(job)
            result := Result{JobID: job.ID, Value: value, Err: err}

            select {
            case results <- result:
            case <-ctx.Done():
                return
            }
        }
    }
}

func main() {
    ctx, cancel := context.WithCancel(context.Background())
    defer cancel()

    jobs := make(chan Job)
    results := make(chan Result)

    var wg sync.WaitGroup
    const workerCount = 2
    wg.Add(workerCount)
    for id := 1; id <= workerCount; id++ {
        go worker(ctx, jobs, results, &wg)
    }

    go func() {
        defer close(jobs)
        input := []int{2, 3, -1, 4, 5}
        for id, value := range input {
            select {
            case jobs <- Job{ID: id + 1, Value: value}:
            case <-ctx.Done():
                return
            }
        }
    }()

    go func() {
        wg.Wait()
        close(results)
    }()

    for result := range results {
        if result.Err != nil {
            fmt.Printf("ish=%d xato: %v\n", result.JobID, result.Err)
            cancel()
            continue
        }
        fmt.Printf("ish=%d natija=%d\n", result.JobID, result.Value)
    }
}

Natija tartibi va xatodan oldin nechta ish tugashi o‘zgarishi mumkin. Natijada quyidagi xato satri bo‘ladi:

ish=3 xato: manfiy qiymatga ruxsat berilmaydi

Worker jobsni ham, ctx.Done()ni ham kutadi. Natija yuborishda ham bekor qilish tekshiriladi. Aks holda main natijalarni qabul qilishni to‘xtatsa, worker results <- result qatorida bloklanib qolishi mumkin.

Xato olinganda cancel() chaqiriladi. Producer va workerlar signalni kuzatib, imkon qadar tez tugaydi. Shu bilan birga bekor qilish darhol to‘xtash kafolati emas: selectda ish va bekor qilish bir paytda tayyor bo‘lsa, ulardan biri tanlanadi. Boshlangan process funksiyasi ham o‘z-o‘zidan to‘xtamaydi; uzoq vazifa bo‘lsa, uning o‘zi ham contextni tekshirishi kerak.

Natijalar tartibini saqlash

Worker pool bajarilish tartibini kafolatlamaydi. Ikkinchi ish birinchisidan tezroq tugasa, uning natijasi oldin keladi. Tartib muhim bo‘lsa, har bir ishga indeks bering va natijani shu indeks bo‘yicha joylashtiring:

ordered := make([]int, jobCount)
for result := range results {
    ordered[result.JobID-1] = result.Value
}

Bu usul barcha natijalarni xotirada saqlaydi. Natijalar juda ko‘p bo‘lsa, tartibni tiklash uchun alohida yig‘uvchi bosqich va cheklangan buffer kerak bo‘lishi mumkin. Tartib talab qilinmasa, natijani kelishi bilan qayta ishlash xotira va kechikishni kamaytiradi.

Worker sonini tanlash

“Worker qancha ko‘p bo‘lsa, dastur shuncha tez ishlaydi” degan qoida yo‘q.

  • CPUga bog‘liq (CPU-bound) ishda worker sonini mavjud parallelizmga yaqin tutish odatda boshlang‘ich nuqta bo‘ladi. Juda ko‘p worker kontekst almashishini oshiradi.
  • I/Oga bog‘liq (I/O-bound) ishda worker ko‘proq bo‘lishi mumkin, chunki ko‘p goroutine tarmoq yoki diskni kutadi.
  • Database va tashqi API bilan ishlaganda connection pool, rate limit va xizmatning ko‘tara oladigan yukini hisobga olish kerak.
  • Eng to‘g‘ri qiymat benchmark, profiling va ishlab chiqarish metrikalari bilan aniqlanadi.

Worker soni nol bo‘lsa, hech kim jobsni qabul qilmaydi va producer bloklanishi mumkin. Pool yaratadigan funksiya workerCount > 0 shartini tekshirishi kerak. Juda katta yoki cheksiz navbat esa xotirani band qiladi va eski vazifalarning kechikishini oshiradi.

Panic va resurslarni tozalash

defer wg.Done() worker qanday oddiy yo‘l bilan qaytishidan qat’i nazar, WaitGroup hisoblagichini kamaytiradi. Ammo panic boshqa masala: panic tiklanmasa, odatda butun dastur to‘xtaydi.

Worker chegarasida recover ishlatish ba’zi serverlarda zarur bo‘lishi mumkin, lekin panicni indamay yutish noto‘g‘ri. Uni xato natijasiga aylantirish, loglash va tizim holati buzilmaganini baholash kerak. Har bir vazifada ochilgan fayl, response body yoki boshqa resurs ham o‘z vaqtida yopilishi lozim.

Keng tarqalgan xatolar

  • jobs kanalini yopmaslik. for range jobs ishlatayotgan workerlar yangi vazifani abadiy kutadi.
  • Kanalni bir nechta workerdan yopish. Kanal faqat boshqa qiymat yuborilmasligini aniq biladigan tomon tomonidan yopiladi.
  • resultsni workerlar tugashidan oldin yopish. Keyingi yuborish send on closed channel panici bilan tugaydi.
  • Natijalarni hech kim qabul qilmasligi. results bufferi to‘lgach workerlar bloklanadi, Wait() esa tugamaydi.
  • Natijalar kirish tartibida keladi deb hisoblash. Concurrent bajarilishda tartibni alohida tiklash kerak.
  • Har bir ish uchun yana nazoratsiz goroutine yaratish. Bu pool qo‘ygan concurrency chegarasini amalda bekor qiladi.
  • WaitGroupga worker boshlanganidan keyin Add qilish. Add chaqiruvlarini goroutineni ishga tushirishdan oldin bajarish xavfsiz va tushunarli.
  • Nusxalangandan keyin WaitGroupdan foydalanish. sync.WaitGroup birinchi ishlatilgandan keyin nusxalanmasligi kerak; funksiyaga pointer uzatiladi.

Interviewda muhim nuqtalar

  • Worker pool concurrencyni, kanal bufferi esa navbatda turgan ishlar sonini boshqaradi. Ular bir xil tushuncha emas.
  • Kanalni yopish “qiymatlarni o‘chirish” emas. Bufferdagi qiymatlar qabul qilinadi, keyin ok == false bo‘ladi.
  • Natijalar tartibi kafolatlanmaydi.
  • WaitGroup workerlarning tugashini kutadi, lekin xato yoki bekor qilishni o‘zi uzatmaydi.
  • Bekor qilish producer, worker, natija yuborish va uzoq davom etadigan vazifaning o‘zida kuzatilishi kerak.

Xulosa

Worker pool bir vaqtda bajariladigan ishlar sonini chegaralab, CPU, xotira, database va tashqi xizmatlarga tushadigan yukni boshqaradi. Ishonchli tuzilishda kanal egasi, yopilish tartibi, natija oqimi, xato va bekor qilish qoidalari aniq bo‘ladi. Worker soni va buffer hajmi taxmin bilan emas, resurs limitlari hamda o‘lchovlar asosida tanlanadi.

📍 Toshkent, O‘zbekiston
🗓️
👤