D.ershane - Veri Paylaşarak Eş Zamanlı Programlama

Veri Paylaşarak Eş Zamanlı Programlama

Bir önceki bölümdeki yöntemler iş parçacıklarının mesajlaşarak bilgi alış verişinde bulunmalarını sağlıyordu. Daha önce de söylediğim gibi, programın doğru çalışması açısında güvenli olan yöntem odur.

Diğer yöntem, iş parçacıklarının aynı verilere doğrudan erişmelerine dayanır. İş parçacıkları aynı veriyi doğrudan okuyabilirler ve değiştirebilirler. Örneğin sahip, işçiyi bool bir değişkenin adresi ile başlatabilir ve işçi de sonlanıp sonlanmayacağını doğrudan o değişkenin değerini okuyarak anlayabilir. Başka bir örnek olarak, sahip bir kaç tane iş parçacığını hesaplarının sonuçlarını ekleyecekleri bir değişkenin adresi ile başlatabilir ve işçiler de o değişkenin değerini doğrudan arttırabilirler.

Veri paylaşımının güvenli olmamasının bir nedeni, iş parçacıklarının veriyi okurken ve değiştirirken birbirlerinden habersizce yarış halinde olmalarıdır (race condition). İşletim sisteminin iş parçacıklarını ne zaman duraksatacağı ve ne zaman tekrar başlatacağı konusunda hiçbir tahminde bulunulamayacağından programın davranışı bu yüzden şaşırtıcı derecede karmaşık olabilir. Hatta, veri paylaşımına dayanan eş zamanlı programların doğruluklarının kanıtlanamayacakları kanıtlanabilir.

Bu başlık altında kullanacağım örnekleri fazlaca basit ve anlamsız bulabilirsiniz. Buna rağmen, veri paylaşımının burada göreceğimiz sakıncaları gerçek programlarda da bulunur.

Paylaşım otomatik değildir

Çoğu dilin tersine, D'de iş parçacıkları verileri açıkça paylaşamazlar. Örneğin, sonlanmasını bildirmek için işçiye bool türündeki bir değişkenin adresini göndermeye çalışan aşağıdaki kod D'de derlenemez:

import std.concurrency;

void işçi(bool * devam_mı)
{
    while (*devam_mı) {
        // ...
    }
}

void main()
{
    bool devam_mı = true;
    spawn(&işçi, &devam_mı);      // ← derleme HATASI

    // ...

    // Daha sonra işçi'nin sonlanmasını sağlamak için
    devam_mı = false;

    // ...
}

std.concurrency modülündeki bir static assert, bir iş parçacığının değişebilen (mutable) verisine başka iş parçacığı tarafından erişilmesini engeller:

src/phobos/std/concurrency.d(329): Error: static assert
"Aliases to mutable thread-local data not allowed."

main() içindeki devam_mı değişebilen bir veri olduğundan ona erişim sağlayan adresi hiçbir iş parçacığına geçirilemez.

Öte yandan, değişmez verilerin adreslerini iş parçacıklarına geçirmekte bir sakınca yoktur çünkü o veriler hiç değişmediklerinden verinin değeri konusunda hata olamaz:

import std.stdio;
import std.concurrency;
import core.thread;

void işçi(immutable(int) * veri)
{
    writeln("veri: ", *veri);
}

void main()
{
    immutable int bilgi = 42;
    spawn(&işçi, &bilgi);         // ← derlenir

    thread_joinAll();
}

Yukarıdaki program bu sefer derlenir ve beklenen çıktıyı üretir:

veri: 42

bilgi'nin yaşamı main() ile sınırlı olduğundan, ona erişmekte olan iş parçacığı sonlanmadan main()'den çıkılmamalıdır. Programın sonunda çağrılan core.thread.thread_joinAll(), işçilerin tamamlanmalarını bekler. main()'den çıkılması böylece engellendiği için bilgi değişkeni işçi() işlediği sürece geçerli kalacaktır.

Veri paylaşmak için `shared`

Gerektiğinde, değişebilen veriler de paylaşılabilirler. Bu durumda programın davranışının doğruluğunu sağlamak programcının sorumluluğundadır.

Ortaklaşa erişilecek olan verilerin shared belirteciyle tanımlanmaları yeterlidir. Aşağıdaki programdaki iş parçacıkları iki değişkenin adreslerini alıyorlar ve o değişkenlerin değerlerini değiş tokuş ediyorlar:

import std.stdio;
import std.concurrency;
import core.thread;

void değişTokuşçu(shared(int) * birinci, shared(int) * ikinci)
{
    foreach (i; 0 .. 10_000) {
        int geçici = *ikinci;
        *ikinci = *birinci;
        *birinci = geçici;
    }
}

void main()
{
    shared int i = 1;
    shared int j = 2;

    writefln("önce : %s ve %s", i, j);

    foreach (adet; 0 .. 10) {
        spawn(&değişTokuşçu, &i, &j);
    }

    // Bütün işlemlerin bitmesini bekliyoruz
    thread_joinAll();

    writefln("sonra: %s ve %s", i, j);
}

Yukarıdaki program artık derlenir ama yanlış çalışır. Programda 10 tane iş parçacığı başlatılıyor. Hepsi de main() içindeki i ve j isimli aynı değişkenlere eriştiklerinden farkında olmadan birbirlerinin işlerini bozuyorlar.

Yukarıdaki programdaki toplam değiş tokuş adedi 10 çarpı 10 bindir. Bu değer bir çift sayı olduğundan, i'nin ve j'nin değerlerinin program sonunda yine başlangıçtaki gibi 1 ve 2 olmalarını bekleriz:

önce : 1 ve 2
sonra: 1 ve 2      ← beklenen sonuç

Program farklı zamanlarda ve ortamlarda gerçekten de o sonucu üretebilir ama aşağıdaki yanlış sonuçların olasılığı daha yüksektir:

önce : 1 ve 2
sonra: 1 ve 1      ← yanlış sonuç

önce : 1 ve 2
sonra: 2 ve 2      ← yanlış sonuç

Başka zamanlarda sonuç "2 ve 1" de çıkabilir.

Bunun nedenini A ve B olarak isimlendireceğimiz yalnızca iki iş parçacığının işleyişiyle bile açıklayabiliriz. İşletim sistemi iş parçacıklarını belirsiz zamanlarda duraksatıp tekrar başlattığı için bu iki iş parçacığı, verileri birbirlerinden habersiz olarak aşağıdaki biçimde değiştirebilirler.

i'nin 1 ve j'nin 2 oldukları duruma bakalım. Aynı değişTokuşçu() işlevini işlettikleri halde A ve B iş parçacıklarının yerel geçici değişkenleri farklıdır. Ayırt edebilmek için onları aşağıda geçiciA ve geçiciB olarak belirtiyorum. Her iki iş parçacığının işlettiğı aynı 3 satırlık kodun zaman ilerledikçe nasıl işletildiklerini yukarıdan aşağıya doğru gösteriyorum: 1 numaralı işlem ilk işlem, 6 numaralı işlem de son işlem. Her işlemde i ve j'den hangisinin değiştiğini de sarıyla işaretliyorum:

Zaman            İş parçacığı A                       İş parçacığı B

  1:   int geçici = *ikinci; (geçiciA==2)
  2:   *ikinci = *birinci;   (i==1, j==1)

                (A duraksatılmış ve B başlatılmış olsun)

  3:                                        int geçici = *ikinci; (geçiciB==1)
  4:                                        *ikinci = *birinci;   (i==1, j==1)

                (B duraksatılmış ve A tekrar başlatılmış olsun)

  5:   *birinci = geçici;    (i==2, j==1)

                (A duraksatılmış ve B tekrar başlatılmış olsun)

  6:                                        *birinci = geçici;    (i==1, j==1)

Görüldüğü gibi, hem i hem de j 1 değerini almışlardır. Artık programın sonuna kadar başka değer almaları mümkün değildir.

Yukarıdaki işlem sıraları bu programdaki hatayı açıklamaya yeten yalnızca bir durumdur. Onun yerine 10 iş parçacığının etkileşimlerinden oluşan çok sayıda başka karmaşık durum da düşünülebilir.

Veri korumak için `synchronized`

Yukarıdaki hatalı durum, aynı verinin birden fazla iş parçacığı tarafından okunması ve yazılması nedeniyle oluşmaktadır. Bu tür hataları önlemenin yolu, belirli bir anda yalnızca tek iş parçacığı tarafından işletilmesi gereken kod bloğunu synchronized olarak işaretlemektir. Yapılacak tek değişiklik programın artık doğru sonuç üretmesi için yeterlidir:

    foreach (i; 0 .. 10_000) {
        synchronized {
            int geçici = *ikinci;
            *ikinci = *birinci;
            *birinci = geçici;
        }
    }

Çıktısı:

önce : 1 ve 2
sonra: 1 ve 2      ← doğru sonuç

synchronized, isimsiz bir kilit oluşturur ve bu kilidi belirli bir anda yalnızca tek iş parçacığına verir. Yukarıdaki kod bloğu da bu sayede belirli bir anda tek iş parçacığı tarafından işletilir ve i ve j'nin değerleri her seferinde doğru olarak değiş tokuş edilmiş olur. Değişkenler programın çalışması sırasında ya "1 ve 2" ya da "2 ve 1" durumundadırlar.

Kullanacağı kilit veya kilitler synchronized'a açıkça da verilebilir. Bu, belirli bir anda birden fazla bloktan yalnızca birisinin işlemesini sağlar.

Bunun bir örneğini görmek için aşağıdaki programa bakalım. Bu programda paylaşılan veriyi değiştiren iki kod bloğu bulunuyor. Bu blokları int türündeki aynı değişkenin adresi ile çağıracağız. Birisi bu değişkenin değerini arttıracak, diğeri ise azaltacak:

void arttırıcı(shared(int) * değer)
{
    foreach (i; 0 .. 1000) {
        ++(*değer);
    }
}

void azaltıcı(shared(int) * değer)
{
    foreach (i; 0 .. 1000) {
        --(*değer);
    }
}

Aynı veriyi değiştirdikleri için bu iki bloğun da synchronized olarak işaretlenmeleri düşünülebilir, ancak yeterli olmaz. Bu bloklar farklı olduklarından her birisi farklı bir kilit ile korunacaktır ve değişkenin tek iş parçacığı tarafından değiştirilmesi yine sağlanamayacaktır:

import std.stdio;
import std.concurrency;
import core.thread;

void arttırıcı(shared(int) * değer)
{
    foreach (i; 0 .. 1000) {
        synchronized {  // ← bu kilit aşağıdakinden farklıdır
            ++(*değer);
        }
    }
}

void azaltıcı(shared(int) * değer)
{
    foreach (i; 0 .. 1000) {
        synchronized {  // ← bu kilit yukarıdakinden farklıdır
            --(*değer);
        }
    }
}

void main()
{
    shared int ortak = 0;

    foreach (i; 0 .. 100) {
        spawn(&arttırıcı, &ortak);
        spawn(&azaltıcı, &ortak);
    }

    thread_joinAll();
    writeln("son değeri: ", ortak);
}

Eşit sayıda arttırıcı ve azaltıcı iş parçacığı başlatılmış olduğundan ortak isimli değişkenin son değerinin sıfır olmasını bekleriz ama büyük olasılıkla sıfırdan farklı çıkar:

son değeri: -3325  ← sıfır değil

Farklı blokların aynı kilidi (veya kilitleri) paylaşmaları için kilidin (veya kilitlerin) synchronized'a parantez içinde bildirilmesi gerekir:

    synchronized (kilit_nesnesi, başka_kilit_nesnesi, ...)

D'de özel bir kilit nesnesi yoktur, herhangi bir sınıf türünün herhangi bir nesnesi kilit olarak kullanılabilir. Yukarıdaki programdaki iş parçacıklarının aynı kilidi kullanmaları için main() içinde bir nesne oluşturulabilir ve iş parçacıklarına parametre olarak o gönderilebilir. Programın değişen yerlerini işaretliyorum:

import std.stdio;
import std.concurrency;
import core.thread;

class Kilit
{}

void arttırıcı(shared(int) * değer, shared Kilit kilit)
{
    foreach (i; 0 .. 1000) {
        synchronized (kilit) {
            ++(*değer);
        }
    }
}

void azaltıcı(shared(int) * değer, shared Kilit kilit)
{
    foreach (i; 0 .. 1000) {
        synchronized (kilit) {
            --(*değer);
        }
    }
}

void main()
{
    shared Kilit kilit = new shared(Kilit)();
    shared int ortak = 0;

    foreach (i; 0 .. 100) {
        spawn(&arttırıcı, &ortak, kilit);
        spawn(&azaltıcı, &ortak, kilit);
    }

    thread_joinAll();
    writeln("son değeri: ", ortak);
}

Bütün iş parçacıkları main() içinde tanımlanmış olan aynı kilidi kullandıklarından belirli bir anda bu iki synchronized bloğundan yalnızca bir tanesi işletilir ve sonuç beklendiği gibi sıfır çıkar:

son değeri: 0      ← doğru sonuç

Bu dersi yazdığım sırada kullandığım dmd 2.058, D'nin veri paylaşımına dayanan eş zamanlı programlama olanaklarının hepsini desteklemiyor. Bu olanaklardan bazılarını fazla ayrıntıya girmeden göstereceğim.

Sınıf türleri synchronized olarak işaretlenebilirler. Bunun anlamı, o türün bütün üye işlevlerinin o türün belirli bir nesnesi ile ifade edilen aynı kilidi kullanacaklarıdır:

synchronized class Sınıf
{
    void foo()
    {
        // ...
    }

    void bar()
    {
        // ...
    }
}

synchronized olarak işaretlenen türlerin bütün üye işlevleri nesnenin kendisini kilit olarak kullanırlar. Yukarıdaki sınıfın eşdeğeri aşağıdaki sınıftır:

class Sınıf
{
    void foo()
    {
        synchronized (this) {
            // ...
        }
    }

    void bar()
    {
        synchronized (this) {
            // ...
        }
    }
}

Birden fazla nesnenin kilitlenmesi gerektiğinde bütün nesneler aynı synchronized deyimine yazılmalıdırlar. Aksi taktirde farklı iş parçacıkları farklı nesnelerin kilitlerini ele geçirmiş olabileceklerinden, sonsuza kadar birbirlerini bekleyerek takılıp kalabilirler.

Bunun tanınmış bir örneği, bir banka hesabından diğerine para aktaran işlevdir. Böyle bir işlemin hatasız gerçekleşmesi için her iki banka hesabının da kilitlenmesinin gerekeceği açıktır. Bu durumda yukarıda gördüğümüz synchronized kullanımını aşağıdaki gibi uygulamak hatalı olur:

void paraAktar(shared BankaHesabı kimden,
               shared BankaHesabı kime)
{
    synchronized (kimden) {           // ← HATALI
        synchronized (kime) {
            // ...
        }
    }
}

Yanlışlığın nedenini şöyle basit bir durumla açıklayabiliriz: Bir iş parçacığının A hesabından B hesabına para aktardığını, başka bir iş parçacığının da B hesabından A hesabına para aktardığını düşünelim. İşletim sisteminin iş parçacıklarını belirsiz zamanlarda duraksatması sonucunda; kimden olarak A hesabını işlemekte olan iş parçacığı A nesnesini, kimden olarak B nesnesini işlemekte olan iş parçacığı da B nesnesini kilitlemiş olabilir. Bu durumda her ikisi de diğerinin elinde tuttuğu nesneyi kilitlemeyi bekleyeceklerinden sonsuza kadar takılıp kalacaklardır.

Bu sorunun çözümü synchronized deyiminde birden fazla nesne belirtmektir:

void paraAktar(shared BankaHesabı kimden,
               shared BankaHesabı kime)
{
    synchronized (kimden, kime) {     // ← doğru
        // ...
    }
}

Derleyici ya nesnelerin ikisinin birden kilitleneceğini ya da hiçbirisinin kilitlenmeyeceğini garanti eder.

Özet

İş parçacıklarının birbirlerine bağlı olmadıkları durumlarda iki önceki bölümün konusu olan std.parallelism modülünün sunduğu koşut programlamayı yeğleyin. Ancak iş parçacıkları birbirlerine bağlı olduklarında std.concurrency'nin sunduğu eş zamanlı programlamayı düşünün.
Eş zamanlı programlama gerçekten gerektiğinde bir önceki bölümün konusu olan mesajlaşmayı yeğleyin çünkü veri paylaşımı çeşitli program hatalarına açıktır.
Yalnızca immutable ve shared olan veriler paylaşılabilirler.
synchronized belirli bir kod bloğunun belirli bir anda tek iş parçacığı tarafından işletilmesini sağlar.
Bir sınıf türü synchronized olarak tanımlandığında, belirli bir nesnesi üzerinde belirli bir anda üye işlevlerinden yalnızca birisi işletilir.

[ ↢ Geri ] [ İleri ↣ ]

Veri Paylaşarak Eş Zamanlı Programlama

Paylaşım otomatik değildir

Veri paylaşmak için shared

Veri korumak için synchronized

Özet

Veri paylaşmak için `shared`

Veri korumak için `synchronized`