آموزش کانکارنسی در گولنگ - کار با `sync.Cond` در Goroutine

در برنامه‌نویسی همزمان (Concurrent Programming)، همگام‌سازی (Synchronization) نقش کلیدی در جلوگیری از Race Conditions و اطمینان از عملکرد هماهنگ بین تردها یا گوروتین‌ها دارد. تصور کنید مسئله‌ای دارید که باید چند تولیدکننده (Producer) و مصرف‌کننده (Consumer) را که به یک منبع مشترک مثل یک بافر یا صف دسترسی دارند، هماهنگ کنید. این چالش کلاسیک در حوزه هم‌زمانی با نام "مسئله تولیدکننده-مصرف‌کننده" شناخته می‌شود.

در چنین شرایطی، همگام‌سازی ضروری است تا تولیدکننده‌ها داده‌ها را بازنویسی نکنند و مصرف‌کننده‌ها داده‌های نامعتبر یا قدیمی را نخوانند. این همگام‌سازی ضروری است، چون در صورت نبود آن، دسترسی هم‌زمان به داده‌های مشترک می‌تواند منجر به شرایط رقابتی، خرابی داده‌ها یا حتی کرش برنامه شود.

تولیدکننده‌ها باید منتظر بمانند اگر بافر پر باشد، و مصرف‌کننده‌ها نیز باید منتظر بمانند اگر بافر خالی باشد. گاهی ممکن است با بافری محدود و دارای اندازه ثابت مواجه شوید که باید دسترسی چند تولیدکننده و مصرف‌کننده به آن به‌درستی مدیریت شود.

sync.Cond چی هست؟

در زبان گو، ساختار sync.Cond یک مکانیزم سیگنال‌دهی است که به گوروتین‌ها اجازه می‌دهد تا زمانی منتظر بمانند که یک شرط خاص برقرار شود. این قابلیت به‌ویژه در هماهنگ‌سازی جریان‌های کاری پیچیده کاربرد دارد؛ جایی که برخی از گوروتین‌ها باید اجرای خود را متوقف کرده و منتظر بمانند تا گوروتین‌های دیگر عملیات مشخصی را انجام دهند.

مفاهیم اصلی پشت sync.Cond:

• مسدودسازی (Blocking): گوروتین‌ها می‌توانند منتظر دریافت یک سیگنال بمانند و تا زمان دریافت آن، اجرای خود را متوقف کنند.

• سیگنال‌دهی (Signaling): سایر گوروتین‌ها می‌توانند با ارسال سیگنال، گوروتین‌های منتظر را از برقرار شدن شرط مطلع کنند.

• بهره‌وری (Efficiency): با به خواب بردن گوروتین‌ها تا زمان دریافت سیگنال، از مصرف بی‌مورد منابع (Busy Waiting) جلوگیری می‌شود.

نحوه کار sync.Cond:

• مقدمه‌سازی (sync.Cond Initialization):
  برای استفاده از sync.Cond باید یک Locker تعریف کنید؛ معمولاً از sync.Mutex یا sync.RWMutex استفاده می‌شود. این Locker وظیفه محافظت از منابع اشتراکی را برعهده دارد.

• تابع Wait():
  وقتی یک گوروتین Wait() را فراخوانی می‌کند:

  • ابتدا قفل مربوطه را آزاد می‌کند تا گوروتین‌های دیگر بتوانند به منبع مشترک دسترسی داشته باشند.

  • سپس منتظر می‌ماند (مسدود می‌شود) تا سیگنال دریافت کند.

  • پس از دریافت سیگنال، مجدداً قفل را به‌دست می‌آورد و به اجرای خود ادامه می‌دهد.

• توابع Signal() و Broadcast():

  • Signal() فقط یک گوروتین منتظر را بیدار می‌کند تا ادامه دهد.

  • Broadcast() تمام گوروتین‌های منتظر را بیدار می‌کند.

مسئله: تولیدکننده–مصرف‌کننده با استفاده از Mutex و متغیر شرطی (Condition Variable)

تصور کنید یک بافر (یا صف) با اندازه‌ای ثابت دارید. چندین تولیدکننده وظیفه تولید آیتم و افزودن آن به بافر را دارند، در حالی‌ که چندین مصرف‌کننده این آیتم‌ها را از بافر حذف می‌کنند. چالش اصلی در این سناریو عبارت است از:

• اطمینان حاصل شود که تولیدکننده‌ها فقط زمانی آیتم اضافه کنند که در بافر فضا وجود داشته باشد.

• اطمینان حاصل شود که مصرف‌کننده‌ها فقط زمانی آیتم حذف کنند که بافر خالی نباشد.

• به تولیدکننده‌ها و مصرف‌کننده‌ها سیگنال داده شود تا زمانی که مجاز به افزودن یا حذف آیتم هستند، عملیات خود را انجام دهند.

در اینجا ساختار اولیه کد آورده شده است:

package main

import (
    "fmt"
    "sync"
    "time"
)

const bufferSize = 5

type Buffer struct {
    data []int
    mu   sync.Mutex
    cond *sync.Cond
}

func (b *Buffer) produce(item int) {
    // Producer logic to add item to the buffer
}

func (b *Buffer) consume() int {
    // Consumer logic to remove item from the buffer
    return 0
}

func main() {
    buffer := &Buffer{data: make([]int, 0, bufferSize)}
    buffer.cond = sync.NewCond(&buffer.mu)
    var wg sync.WaitGroup
    // Start producer goroutines
    for i := 1; i <= 3; i++ {
       wg.Add(1)
       go func(id int) {
          defer wg.Done()
          for j := 0; j < 5; j++ { // Each producer creates 5 items
             buffer.produce(id*10 + j) // Produce unique items based on id and j
             time.Sleep(100 * time.Millisecond)
          }
       }(i)
    }
    // Start consumer goroutines
    for i := 1; i <= 3; i++ {
       wg.Add(1)
       go func(id int) {
          defer wg.Done()
          for j := 0; j < 5; j++ { // Each consumer consumes 5 items
             item := buffer.consume()
             fmt.Printf("Consumer %d consumed item %d
", id, item)
             time.Sleep(150 * time.Millisecond)
          }
       }(i)
    }
    wg.Wait()
    fmt.Println("All producers and consumers finished.")
}

buffer.cond = sync.NewCond(&buffer.mu)

func (b *Buffer) produce(item int) {
    b.mu.Lock()
    defer b.mu.Unlock()

    // Wait if the buffer is full
    for len(b.data) == bufferSize {
       b.cond.Wait() // Release lock and wait until signaled
    }

    // Add item to the buffer
    b.data = append(b.data, item)
    fmt.Printf("Produced item %d
", item)

    // Signal a consumer that an item is available
    b.cond.Signal()
}

func (b *Buffer) consume() int {
    b.mu.Lock()
    defer b.mu.Unlock()

    // Wait if the buffer is empty
    for len(b.data) == 0 {
       b.cond.Wait() // Release lock and wait until signaled
    }

    // Remove item from the buffer
    item := b.data[0]
    b.data = b.data[1:]
    fmt.Printf("Consumed item %d
", item)

    // Signal a producer that space is available
    b.cond.Signal()

    return item
}