Go Condition Variable Nedir?

Condition variable kavramının ne olduğu adından ötürü akılda bazı şeyler canlandırabilir. Türkçesi “Koşul Değişkeni” olan özellik ile yazının devamında ufak bir kuyruk yapısı yazacağım.

Redis, RabbitMQ gibi araçlar kullandıysanız eğer bu araçlar sizin bazı işlerinizi sıraya alıp, kuyruklayarak gerçekleştirmenize imkan sağlarlar.

Bu konuda en bilinen örnek ise e-posta gönderimidir.

Varsayalım ki elinizde 10.000 adet kullanıcı var ve hepsine bireysel olarak yani ayrı ayrı e-posta göndereceksiniz. 10.000 adet e-postayı tek seferde bir for döngüsü içerisinde gönderebilir misiniz? Aslında gönderebilirsiniz yani teknik olarak mümkün fakat bu esnada işlemciniz diğer beklemeye alabilir veya posta gönderdiğiniz sunucunun rate limit’ini doldurabilirsiniz. İşte kuyruk yapıları bu tür durumlara çözüm sağlıyor.

Şimdi asıl konudan sapmadan Condition Variable ile ufak bir kuyruk yapısı hazırlayalım.

Kullanımı şu şekilde

c := &sync.Cond{L: &sync.Mutex{}}

Peki burada Mutex diye bir şey kullandık bu nedir?

Mutex, birden fazla thread/goroutine tarafından okuma, yazma yapılabilen bir veriye sadece bir thread/goroutine’inin erişim sağlaması için hazırlanmış, aynı anda iki adet thread/goroutine’in erişmesini önleyen, sıraya sokan yapıdır. Mutex sadece Go’da değil bir çok sistemde Mutual Exclusion adı ile mevcut bir yapıdır.

Condition Variable içerisinde bizim örneğimizde kullanacağız 2 adet fonksiyon barındırıyor. Bunlar Wait() ve Signal()

Wait() fonksiyonu Lock yani kilitli durumdaki mutex’i Unlocked yani açık hale getirir ve diğer goroutine’lerin veri okuma/yazma işlemi yapmalarına imkan tanır ve bulunduğu goroutine’i bekleme durumuna geçirir.

Signal() fonksiyonu ise Wait() fonksiyonunun beklemeye aldığı goroutine’i uyandırır ve devam etmesini sağlar.

Örnek ile daha iyi anlaşılabilir;

package main
var jobs []int
func main() {
  c := &sync.Cond{L: &sync.Mutex{}}
  go func() {
  for {
      c.L.Lock()
      fmt.Println("Loop started")
      if len(jobs) == 0 {
          fmt.Println("Loop waiting")
          c.Wait()
      }
      fmt.Println("Loop continue")
      jobs = jobs[:len(jobs)-1]
      c.L.Unlock()
     }
  }()
}

Yukaridaki fonksiyonu bize ne anlatıyor?

Öncelikle sonsuz bir döngümüz var biz durdurana kadar devam edecek.

Hemen ardından c.L.Lock() ile mutex kilitli hale geldi ve biz açık hale getirene kadar başka bir goroutine erişemeyecek, eğer jobs adındaki slice’ımızda beklemede olan hiç bir iş yok ise yani len(jobs) == 0 ise c.Wait() ile bu goroutine’i beklemeye alıyoruz ve for döngümüz de beklemeye alınıyor. Boş yere döngümüz devam etmiyor.

jobs adlı slice’a yeni bir iş eklendiğinde yani len(jobs)>0 durumunda olduğunda ise for döngümüz devam edecek. Fakat burada sorun şu c.Wait() ile bu goroutine beklemeye alındı. Bunu uyandırmamız ve işleme devam etmesini söylememiz lazım. Bunu ise c.Signal() fonksiyonu ile yapacağız.

c.L.Lock()
jobs = append(jobs, 1)
c.L.Unlock()
c.Signal()

Burada yine eşzamanlı okuma/yazma işlemini önlemek adına mutex lock-unlock işlemi uyguladık ve jobs slice’ının içerisine yeni bir iş ekledik. Hemen ardından c.Signal() ile beklemede olan goroutine’i uyandırıyoruz ve işleme devam etmesini sağlıyoruz.

Anlatmaya çalıştığım örneğin tam hali aşağıdaki gibidir. Aşağıdaki örnekte elimizde 1.000 adet iş var ve bunları 500ms aralıklarla gerçekleştiriyoruz.

package main
import (
"fmt"
"sync"
"time"
)
var jobs []int
func main() {
c := &sync.Cond{L: &sync.Mutex{}}
go func() {
for {
c.L.Lock()
fmt.Println("Loop started")
if len(jobs) == 0 {
fmt.Println("Loop waiting")
c.Wait()
}
fmt.Println("Loop continue")
jobs = jobs[:len(jobs)-1]
fmt.Println("One job done")
c.L.Unlock()
}
}()
time.Sleep(time.Second * 2)
for i := 1; i < 1000; i++ {
time.Sleep(time.Millisecond * 500)
c.L.Lock()
jobs = append(jobs, 1)
c.L.Unlock()
c.Signal()
}
var s string
fmt.Scanln(&s)
}