Haftalık C++ 10- std::thread (III)

Merhaba arkadaşlar, uzun bir aradan sonra haftalık C++ yazılarımıza devam ediyoruz. std::thread kütüphanesine ilişkin daha önce başlamış olduğumuz serinin üçüncü yazısı ile sizler ile birlikteyim. Eğer diğer yazılarımı henüz okumadı iseniz, aşağıdaki bağlantılardan muhakkak okumanız öneriyorum, özellikle birinci yazıyı:

Haftalık C++ 7- std::thread (I)

Haftalık C++ 8- std::thread (II)

Haftalık C++ 10- std::thread (III)

İçerik

1 Giriş:
2 std::mutex:
3 std::condition_variable:
4 Sonuç:
5 Kaynaklar:

Giriş:

Gelelim bu yazının konusuna. Bundan önceki yazılarımda, temel std::thread kullanımı ve yardımcı yapılara değinmiştim. Bu yazımda ise, multithreaded yazılım geliştirmede çok önemli bir yere sahip olan senkronizasyon yapılarından sizlere bahsedeceğim. Aslında bakarsanız, bir önceki yazımda anlattığım std::atomics de senkronizasyon yapılarından birisidir ve her ne kadar bütün durumları karşılamasa da, burada anlatacağım bazı problemler için kullanılabilir.

Senkronizasyon yapıları, özellikle bir den fazla thread’in aynı bellek uzayındaki ortak kaynaklara (özellikle bellekte her iki thread’in erişimi olduğu ve yerel olarak alınmamış değişkenler) erişerek değiştirmek istediği durumlarda, ortaya çıkabilecek problemleri önlemek için kullanılırlar. Bunun yanında thread’ler arası haberleşme ve koordinasyon için de kullanılırlar. Bunu uygulamak, tabi söylemekten biraz daha zor, çünkü bu yapıların yanlış kullanılması, bulunması ve ayıklanması zor hatalara yol açabileceği gibi, performans anlamında da uygulamalarınıza ket vurabilir. Bu sebeple de özellikle kritik verilerin paylaşımı konusunu, oldukça dikkatli değerlendirmenizde fayda var. Her ne kadar bu yazımda değinmesem de (ileride belki ayrı bir yazıda bu konuya değinebiliriz), multithreaded programlama için kullanılabilecek bir çok tasarım kalıpları ve yaklaşımları mevcut, uygulamalarınızda mümkün olduğunca bunları kullanmanız, bu tarz problemler ile karşılaşma riskini azaltacaktır.

Şimdi dilerseniz, bir önceki paragrafta ortak kaynaklara erişim konusunu basit bir örnekle inceleyelim:

void saveItem(int item)
{
....
    int freeSlotIndex{0};

...    

    freeSlotIndex = foundFreeIndex;
    storage[freeSlotIndex] = item;
    foundFreeIndex++;

....

}

void saveItem(int item)

{

....

int freeSlotIndex{0};

...

freeSlotIndex = foundFreeIndex;

storage[freeSlotIndex] = item;

foundFreeIndex++;

....

}

Yukarıdaki metot basitçe verilen bir sayıyı, basit bir dizide bulduğu ilk yerde saklıyor. Normal şartlarda, hiçbir sıkıntı görünmeyen bu metodun paralel çalıştırılması durumunda karşılaşabileceğimiz örnek bir duruma göz atalım. Aşağıda iki farklı thread’in bu metodu aynı anda çağırdığı durumda, oluşabilecek örnek bir koşum sırası vermeye çalıştım (bu elbette değişebilir).

Thread 1 Koşumu:

Thread 2 Koşumu:

1) freeSlotIndex = foundFreeIndex;

4) storage[freeSlotIndex] = item;

5) foundFreeIndex++;

2) freeSlotIndex = foundFreeIndex;

3) storage[freeSlotIndex] = item;

6) foundFreeIndex++;

Göreceğiniz üzere hangi satırın hangi sıra ile çağrıldığı duruma göre ilgili dizinin boş yerine yazılan değerlerden birisi diğerini ezebiliyor. Bu probleme multihreaded programlama dünyasında “race condition” (Yarış Koşulu) denilmektedir. Ortak kaynaklara erişilmesi durumunda, bu tarz problemler oluşabilmektedir. Tabi bu problem her zaman oluşacak diye bir durum da yok (inanın bana oluşması ve bunları yakalamak sizler için çok daha hayırlı :). Uygulamada bu tarz problemlere yol açabilecek kısımlara da “critical section” (Kritik Bölüm) denilir. Özellikle, bu yarış koşullarını bulmak her zaman bu kadar kolay olamayabiliyor, çünkü her zaman oluşmuyorlar. Bu problemleri önlemek için, kritik bölümleri koruma altına alıyoruz. Burada özellikle dikkat etmenizi istediğim konu: kaynaklara olan erişimde, eğer her parti sadece okuma yapıyorsa, herhangi bir sıkıntı olmaz ve emniyetli bir şekilde hepsi okuma yapabilir. Ama en az bir taraf, bu veriler üzerine yazma işlemi yapacak ise, işte çarşı o zaman karışıyor. Bunu da not alalım bir kenara.

Yukarıdaki örnekte oluşabilecek olan bir probleme göz attık, peki kritik bölümleri koruma altına almazsak ne tarz problemler ile karşılaşabiliriz? Hemen bakalım:

Senkronize Olmayan Veri Erişimi (“Unsynchronized Data Access”): Aslında yukarıda verdiğimiz örnek bu gruba girmekte. Birden fazla thread paralel bir şekilde, ortak bir veriye okuma ve yazma yapıyor ise hangisinin önce yazdığı problemi ortaya çıkabilir.
Yarı Yazılmış Veri (“Half-written Data”): Benzer şekilde bir thread veriye yazıyorken, diğer thread onu tam da yazma işleminin ortasında okuyabilir. Elinde ne eski ne de yeni veri olabilir 🙂 Bunu da çok basit bir örnek ile anlatmaya çalışayım: elimizde aşağıdaki gibi bir kod bildirimi olduğunu düşünelim:

long long x = 0;

Bir thread bu veriyi aşağıdaki gibi değiştiriyor:

x = -1;

Diğeri de aşağıdaki gibi okuyor:

std::cout << x;

İşte tam bahsettiğimiz probleme örnek olabilecek bir durum.

Sıraları Değiştirilmiş Kod Bildirimleri (“Reordered Statements”): Ayrı ayrı threadlerdeki kod bildirimleri, performans veya benzeri sebeplerle değiştirilmiş olabilirler. Bunlar her ne kadar tek başlarına sıkıntı olmasalar da (sıralı koşturma), paralel koşumlarda, beklenen davranış gözlemlenmeyecektir.

Bu konulara ilişkin daha detaylı bilgiler için (özellikle bu tarz problemler ve yaklaşımlar için), ilk yazımda bahsettiğim aşağıdaki kitaplara bir göz atabilirsiniz:

Şimdi bu yapılardan ilki olan std::mutex’lere bakalım.

std::mutex:

Mutex’ler, “mutual exclusion” olarak da bilinir, bir kaynağa olan eş zamanlı erişim ihtiyacını, o kaynağa özel erişim sunarak sağlayan yapılara denir. Burada taraflar ilgili kaynağa erişmek için std::mutex üzerinden kilitleme işini yapar ve kaynağa erişir, bu sırada diğer threadlerin bu kaynağa erişimi engellenir. Ta ki, ilgili taraf bu kilidi kaldırana kadar. İlk olarak Edsger W. Dijsktra tarafından tanımlanmıştır. Basitçe aslında paylaşılan kaynağa, tek erişimi sağlayan en temel senkronizasyon yapısıdır.

std::mutex sınıfını kullanabilmek için ‘<mutex>’ başlık dosyasını eklemeniz gerekmekte. Bu kütüphane altında mutex anlamında kullanabileceğiniz önemli sınıfları ve hangi C++ ile kullanıma sunulduklarını, aşağıdaki kısa açıklamalar ile özetlemeye çalıştım. Bunlara ilişkin referans bilgilerine bu adresten ulaşabilirsiniz. Sonrasında bunlarına kullanımına hep birlikte örnek kodlar üzerinden bakacağız. Daha detaylı örnek ve kullanımlar için ise, kaynakların adreslerini en son bölümde vereceğim.

Temel mutex Sınıfları – I
std::mutex (C++11)	Temel karşılıklı dışlama (mutual exclusion)/güvenli erişim kabiliyeti sunar
std::timed_mutex (C++11)	Karşılıklı dışlama (mutual exclusion)/güvenli erişim kabiliyetini süreli bir şekilde sunar
recursive_mutex (C++11)	Karşılıklı dışlama (mutual exclusion)/güvenli erişim kabiliyetini, aynı thread için yineli bir şekilde gerçekleştirilmesine olanak sağlar
recursive_timed_mutex (C++11)	Karşılıklı dışlama (mutual exclusion)/güvenli erişim kabiliyetini, aynı thread için yineli ve verilen zaman için gerçekleştirilecek şekilde yapılabilmesine olanak sağlar

lock_guard

(C++11)

RAII (Resource Acquisition Is Initialization) mekanizmasına uygun bir şekilde mutex kullanılarak verilen kapsam için karşılıklı dışlama kabiliyeti sunar.Aslında arka planda yapılan şey, yapıcı içerisinde ilgili mutex’i kilitlemek ve kapsam dışında çıkan scoped_lock nesnesi ile çağrılacak yıkıcı içerisinde de bu kilidi geri açmak.

unique_lock

(C++11)

Karşılıklı dışlamayı içerisinde barındıran ve sahipliğini taşıma yolu ile aktarabileceğiniz mekanizmayı sunan sınıftır

Kullanıma Sunulan Bazı Bağımsız Fonksiyonlar

try_lock (C++11)

Metoda geçirilen kilitlenebilir nesnelerin, hepsini kitlemeye çalışır. Kilitleyemediği ilk nesnenin indeksini döner. Başarılı durumda -1 dönülür.

lock (C++11)

Metoda geçirilen kitlenebilir nesneleri kilitler, eğer kilitleyemez ise mevcut thread’i bloklar

İlk yazımda da bahsettiğim gibi, std::thread kütüphanesi aslında ilk olarak C++ 11 ile sunulmaya başlandı, yukarıdaki bahsettiğim sınıflar da öle. C++ 14 ve 17 ile de aşağıdaki gibi bir takım eklemeler oldu:

Temel Mutex Sınıfları II

shared_mutex

(C++17)

Temel karşılıklı dışlama (mutual exclusion) kabiliyetinin bir çok nesne tarafından paylaşılabilmesine olanak sağlar. Özellikle aynı anda birden fazla okuma ve yaz yapılabileceği durumlarda, okumalar esnasında kullanıcıların herhangi bir bloklama olmadan veriye erişmelerine olanak sağlar,

shared_timed_mutex

(C++14)

Bir önceki sınıfın, belirli bir süre için bu kilitlerin tutulabilmesine olanak sağlayan sınıfı

scoped_lock

(C++17)

RAII (Resource Acquisition Is Initialization) mekanizmasına uygun bir şekilde, bir ya da birden fazla mutex kullanılarak verilen kapsam için karşılıklı dışlama kabiliyeti sunar. Lock_guard’tan farkı, birden fazla mutex’i alabilmesidir. Bu kullanıma ilişkin bir örnek aşağıdaki adreste verilmekte:

https://stackoverflow.com/questions/17113619/whats-the-best-way-to-lock-multiple-stdmutexes/17113678

shared_lock

(C++14)

Verilen paylaşımlı mutex için, taşınabilir sahiplik ve bir önceki sınıftakine benzer kabiliyetler sunar

Evet, elimizdeki cephanenin neler olduğunu öğrendiğimize göre, şimdi bunları nasıl kullanabileceğimize bir göz atalım.

İlk örneğimiz, multi-threaded yazılım geliştirirken, eminim hepiniz, konsola bir şeyler bastırmak istemişsinizdir ve ilk tecrübe ettiğiniz deneyim de karmakarışık çıktılar olacaktır muhtemelen. Gelin bunu, yukarıdaki temel yapıları kullanarak basitçe nasıl çözebileceğimize bakalım. Tahmin edebileceğiniz üzere, bu durumdaki paylaşılan kaynak standart çıktı akışı (“standard output stream“).

#include <thread>
#include <mutex>
#include <iostream>
#include <string>
#include <chrono>

// Ortak veri paylaşımı (ki burada standard output) için muteximiz
std::mutex gPrintMutex;

void threadSafePrintVersion1(const std::string& input)
{
    gPrintMutex.lock();
    for(auto c : input)
    {    
        std::cout.put(c);
	    std::this_thread::sleep_for(std::chrono::milliseconds(1));
    }
	gPrintMutex.unlock();
}

void threadSafePrintVersion2(const std::string& input)
{
    std::lock_guard<std::mutex> localLock(gPrintMutex);
    for(auto c : input)
    {    
        std::cout.put(c);
	    std::this_thread::sleep_for(std::chrono::milliseconds(1));
    }
}

int main()
{
	std::thread firstThread(threadSafePrintVersion1, "Hello world from thread 1!\n");
	std::thread secondThread(threadSafePrintVersion2, "Hello world from thread 2!\n");
	std::cout << "Hello world from main thread!\n";
	
	firstThread.join();
	secondThread.join();

	return 0;
}

#include <thread>

#include <mutex>

#include <iostream>

#include <string>

#include <chrono>

// Ortak veri paylaşımı (ki burada standard output) için muteximiz

std::mutex gPrintMutex;

void threadSafePrintVersion1(const std::string& input)

{

gPrintMutex.lock();

for(auto c : input)

{

std::cout.put(c);

std::this_thread::sleep_for(std::chrono::milliseconds(1));

}

gPrintMutex.unlock();

}

void threadSafePrintVersion2(const std::string& input)

{

std::lock_guard<std::mutex> localLock(gPrintMutex);

for(auto c : input)

{

std::cout.put(c);

std::this_thread::sleep_for(std::chrono::milliseconds(1));

}

int main()

{

std::thread firstThread(threadSafePrintVersion1, "Hello world from thread 1!\n");

std::thread secondThread(threadSafePrintVersion2, "Hello world from thread 2!\n");

std::cout << "Hello world from main thread!\n";

firstThread.join();

secondThread.join();

return 0;

}

Yukarıda verilen kodta her iki metot da aynı işi yapmakta. Fakat std::scoped_lock daha okunaklı (tabi bana göre 🙂 hem de direk std::mutex kullanımı durumunda ortaya çıkabilecek unlock() unutmak ya da gereksiz lock() çağrılması benzeri problemleri bertaraf etmekte. Burada tabi sıkıntılı durumu ortaya koymak için koda bir takım eklemeler (bekleme ve put API’lerinin kullanımı) yaptım 🙂 Normalde basit uygulamalarda std::cout bunu sıkıntı yaşamadan görüntüleyebilir. Ama yukarıdaki mutex’ler ile ilgili satırları yorumladığınızda neler olduğunu görebilirsiniz.

Şimdi de, ilk durum kadar olmasa da, std::recursive_lock kullanmanızı gerektirecek durumlara bir göz atalım. Bu ihtiyaç, genellikle her bir metodu içerisinde mutex kullanılan ve bu metotların birbirlerini çağırmaları gerektiği durumlarda ortaya çıkar. Hemen bir örneğe bakalım:

struct Calculator
{
    std::mutex mMutex;
    int mLastResult;

    Calculator() : i(0) {}

    void multiply(int x)
    {
        std::lock_guard<std::mutex> lock{mMutex};
        mLastResult *= x;
    }

    void divide(int x)
    {
        std::lock_guard<std::mutex> lock{mMutex};
        mLastResult /= x;
    }

	// Yukarıdaki her iki metodu da cagirmamiz gerekecek!
	void both(int x, int y)
	{
	    std::lock_guard<std::mutex> lock{mMutex};
	    multiply(x);
	    divide(y);
	}
};

int main()
{
    Complex complex;
    complex.both(32, 23);

    return 0;
}

struct Calculator

{

std::mutex mMutex;

int mLastResult;

Calculator() : i(0) {}

void multiply(int x)

{

std::lock_guard<std::mutex> lock{mMutex};

mLastResult *= x;

}

void divide(int x)

{

std::lock_guard<std::mutex> lock{mMutex};

mLastResult /= x;

}

// Yukarıdaki her iki metodu da cagirmamiz gerekecek!

void both(int x, int y)

{

std::lock_guard<std::mutex> lock{mMutex};

multiply(x);

divide(y);

}

};

int main()

{

Complex complex;

complex.both(32, 23);

return 0;

}

Bu örnekte gördüğünüz gibi tek tek multiply() ya da divide() metotlarını çağırdığınız zaman herhangi bir problem ile karşılaşmayacaksınız. Fakat both() metodunu çağırdığınızda uygulama bloklanacak, sizce neden?

both() metodu, mMutex kilidini başta kilitliyor, daha sonra multiply()’ı çağırdığımızda ise aynı kilidi kilitlemeye çalıştığında, “deadlock” dediğimiz duruma sebebiyet vermiş oluyoruz. Aynı thread, bir mutex’i iki kere kilitleyemez. Şimdi std::mutex kullanımlarının hepsini, std::recursive_mutex ile değiştirelim ve öyle çalıştıralım, bu durumda herhangi bir problem olmadığını göreceksiniz.

Yukarıdaki kullanımlarda, eğer ilgili mutex zaten kilitli ise ilgili thread bloklanır. Bunun yerine ilgili mutex’i kilitleyip/kilitleyemediğinizi sorgulamak isteyebilirsiniz. Bu durumda try_lock() ve türevlerini kullanabilirsiniz. Hemen bir örnek kullanımına (referans sayfasından) bakalım:

// try_lock
#include <chrono>
#include <mutex>
#include <thread>
#include <iostream> // std::cout
 
std::chrono::milliseconds interval(100);
 
std::mutex mutex;
int job_shared = 0; // both threads can modify 'job_shared',
    // mutex will protect this variable
 
int job_exclusive = 0; // only one thread can modify 'job_exclusive'
    // no protection needed
 
// this thread can modify both 'job_shared' and 'job_exclusive'
void job_1() 
{
    std::this_thread::sleep_for(interval); // let 'job_2' take a lock
 
    while (true) 
    {
        // try to lock mutex to modify 'job_shared'
        if (mutex.try_lock()) 
        {
            std::cout << "job shared (" << job_shared << ")\n";
            mutex.unlock();
            return;
        } 
        else 
        {
            // can't get lock to modify 'job_shared'
            // but there is some other work to do
            ++job_exclusive;
            std::cout << "job exclusive (" << job_exclusive << ")\n";
            std::this_thread::sleep_for(interval);
        }
    }
}
 
// this thread can modify only 'job_shared'
void job_2() 
{
    mutex.lock();
    std::this_thread::sleep_for(5 * interval);
    ++job_shared;
    mutex.unlock();
}
 
int main() 
{
    std::thread thread_1(job_1);
    std::thread thread_2(job_2);
 
    thread_1.join();
    thread_2.join();
}

// try_lock

#include <chrono>

#include <mutex>

#include <thread>

#include <iostream> // std::cout

std::chrono::milliseconds interval(100);

std::mutex mutex;

int job_shared = 0; // both threads can modify 'job_shared',

// mutex will protect this variable

int job_exclusive = 0; // only one thread can modify 'job_exclusive'

// no protection needed

// this thread can modify both 'job_shared' and 'job_exclusive'

void job_1()

{

std::this_thread::sleep_for(interval); // let 'job_2' take a lock

while (true)

{

// try to lock mutex to modify 'job_shared'

if (mutex.try_lock())

{

std::cout << "job shared (" << job_shared << ")\n";

mutex.unlock();

return;

}

else

{

// can't get lock to modify 'job_shared'

// but there is some other work to do

++job_exclusive;

std::cout << "job exclusive (" << job_exclusive << ")\n";

std::this_thread::sleep_for(interval);

}

// this thread can modify only 'job_shared'

void job_2()

{

mutex.lock();

std::this_thread::sleep_for(5 * interval);

++job_shared;

mutex.unlock();

}

int main()

{

std::thread thread_1(job_1);

std::thread thread_2(job_2);

thread_1.join();

thread_2.join();

}

Şimdi bir diğer kullanıma daha göz atalım ve sonrasında mutex sayfasını kapatalım. Bu kullanım, std::timed_mutex’ler ile ilgili. Bu sınıf standart std::mutex’in API’leri yanında try_lock_for ve try_lock_until API’lerini sunar. try_lock_for() API’si, verilen mutex’i kilitlemeye çalışır, eğer bu mutex kilitli ise, sonsuza kadar beklemek yerine, verilen süre kadar bu mutex’in kilidinin açılmasını bekler. Eğer alabilirse, true döner, alamaz ise false döner. try_lock_until’de buna benzer bir kullanım sunar, tek farkı süre yerine direk zaman verilmesidir. Yine referans sayfasında verilen örneğe bakalım:

#include <iostream>
#include <mutex>
#include <thread>
#include <vector>
#include <sstream>
 
std::mutex cout_mutex; // control access to std::cout
std::timed_mutex mutex;
 
void job(int id) 
{
    using Ms = std::chrono::milliseconds;
    std::ostringstream stream;
 
    for (int i = 0; i < 3; ++i) {
        if (mutex.try_lock_for(Ms(100))) {
            stream << "success ";
            std::this_thread::sleep_for(Ms(100));
            mutex.unlock();
        } else {
            stream << "failed ";
        }
        std::this_thread::sleep_for(Ms(100));
    }
 
    std::lock_guard<std::mutex> lock(cout_mutex);
    std::cout << "[" << id << "] " << stream.str() << "\n";
}
 
int main() 
{
    std::vector<std::thread> threads;
    for (int i = 0; i < 4; ++i) {
        threads.emplace_back(job, i);
    }
 
    for (auto& i: threads) {
        i.join();
    }
}

#include <iostream>

#include <mutex>

#include <thread>

#include <vector>

#include <sstream>

std::mutex cout_mutex; // control access to std::cout

std::timed_mutex mutex;

void job(int id)

{

using Ms = std::chrono::milliseconds;

std::ostringstream stream;

for (int i = 0; i < 3; ++i) {

if (mutex.try_lock_for(Ms(100))) {

stream << "success ";

std::this_thread::sleep_for(Ms(100));

mutex.unlock();

} else {

stream << "failed ";

}

std::this_thread::sleep_for(Ms(100));

}

std::lock_guard<std::mutex> lock(cout_mutex);

std::cout << "[" << id << "] " << stream.str() << "\n";

}

int main()

{

std::vector<std::thread> threads;

for (int i = 0; i < 4; ++i) {

threads.emplace_back(job, i);

}

for (auto& i: threads) {

i.join();

}

Bazen bir metodun, farklı thread’lerden sadece bir kere çağrılmasını isteyebilirsiniz. Bu kütüphane bu tarz durumlar için de std::call_once fonksiyonunu sunuyor. Buna ilişkin de bir örneğe bakalım:

std::once_flag flag;

void do_something()
{
    std::call_once(flag, [](){std::cout << "Called once" << '\n';});
    std::cout << "Called each time" << '\n';
}

int main(){
    std::thread t1(do_something);
    std::thread t2(do_something);
    std::thread t3(do_something);
    std::thread t4(do_something);

    t1.join();
    t2.join();
    t3.join();
    t4.join();

    return 0;
}

std::once_flag flag;

void do_something()

{

std::call_once(flag, [](){std::cout << "Called once" << '\n';});

std::cout << "Called each time" << '\n';

}

int main(){

std::thread t1(do_something);

std::thread t2(do_something);

std::thread t3(do_something);

std::thread t4(do_something);

t1.join();

t2.join();

t3.join();

t4.join();

return 0;

}

Yukarıda verdiğim örnek kod içerisindeki, doSomenthing() metodunun ilk kısmı, kullanılan ilgili std::once_flag sayesinde sadece bir kere çağrılıyor olacak.

Bu kullanım ile birlikte mutex’lere ilişkin baya bir konuyu işlemiş olduk. Daha bir şey kalmadı mı? Elbette bir çok konu daha var (daha detaylı bilgi için kaynaklara bir göz atabilirsiniz, ya da her zaman google’a danışabilirsiniz 🙂 ama bence bunlar size başlangıç için yeterli olacaktır. Bir sonraki başlıkta bir diğer önemli yapı olan std::condition_variable’lara bakacağız.

std::condition_variable:

Thread kütüphanesi ile sunulan bir diğer önemli yapı da std::condition_variable’dır. Özellikle multithreaded yazılım geliştirdiğiniz durumlarda, bir thread’in diğerini beklemesi, onun yaptığı iş ya da bir koşul sağlandığı zaman çalışmaya devam etmesi gibi durumlar ile karşılaşmışsınızdır. İşte tam da bu gibi durumlar için std::conditional_variable yapıları kullanılabilir. Burada tabi, bir thread’ten veri almak için std::future mekanizmasının kullanılabileceğini söyleyebilirsiniz, doğrudur, fakat std::future’ların tek amacı farklı bir thread’ten veri döndürülmesi ya da hatalı durumun bildirilmesi olduğu için bazı durumlarda daha güçlü bir mekanizmaya ihtiyaç duyabiliriz.

Daha önce pek bahsetmedik ama bir thread işini yaparken diğerine ilişkin durumu aslında bir döngü içerisinde bekleyerek de kontrol edebiliriz. Bu yaklaşım “busy wait” ya da “polling” de deniliyor. Hemen bir örnek üzerinden bu duruma bir göz atalım:

#include <thread>
#include <mutex>
#include <chrono>

// Durum 1: Iyimser bir kontrol durumu
bool isReady {false};
std::mutex isReadyMutex;

// Burada ilgili isReady değişkeni true olana kadar döngü içerisinde bekliyoruz
// ...
// unique_lock burada kilitlemeye baslar
std::unique_lock<std::mutex> loopLock{isReadyMutex};

while(!isReady)
{
	loopLock.unlock();
	// CPU kullanimini bir sonraki thread'e gecirmek icin haber vermek olarak dusunebiliriz 
	std::this_thread::yield();
	// 100 ms sonra tekrar kontrol edelim
	std::this_thread::sleep_for(std::chrono::milliseconds(100));	
	loopLock.lock();
}

#include <thread>

#include <mutex>

#include <chrono>

// Durum 1: Iyimser bir kontrol durumu

bool isReady {false};

std::mutex isReadyMutex;

// Burada ilgili isReady değişkeni true olana kadar döngü içerisinde bekliyoruz

// ...

// unique_lock burada kilitlemeye baslar

std::unique_lock<std::mutex> loopLock{isReadyMutex};

while(!isReady)

{

loopLock.unlock();

// CPU kullanimini bir sonraki thread'e gecirmek icin haber vermek olarak dusunebiliriz

std::this_thread::yield();

// 100 ms sonra tekrar kontrol edelim

std::this_thread::sleep_for(std::chrono::milliseconds(100));

loopLock.lock();

}

Burada gördüğünüz üzere sürekli bir kontrol var ve bu her ne kadar küçük bir iş gibi dursa da, küçük bekleme süreleri sağlamak zor olabilir, daha uzunları için ise gecikmeler yaşanabilir. İşte bunu yerine std::condition_variable kullanabilirsiniz. Peki nedir bu yapılar, hemen referans dokümana bakalım, yine en güzel tanımı kendisi gayet öz bir şekilde veriyor:

Bir çok thread’in birbirleri ile iletişim kurmasına olanak sağlayan bir senkronizasyon yapısıdır. Bu yapı bir ya da daha fazla thread’in, bir ya da fazla thread’i beklemesini (istenirse verilen bir süre kadar), daha sonra ise işine devam etmesini sağlar.

Her bir koşullu değişkeni muhakkak bir mutex ile ilişkilendirilerek kullanılır. Koşullu değişkenlerini kullanmak için ‘<mutex>’ ve ‘<condition_variable>’ başlık dosyalarını eklemeniz gerekmekte. Yukarıda verdiğim örneğe benzer durumlar için:

Bir ya da birden fazla thread herhangi bir koşulun sağlandığını haber vermek için, haber verilecek olan thread sayısına göre:
- notify_one() ya da notify_all() API’lerinden birisini çağırır
Belirtilen koşulu beklemek için ya da ondan haberdar olmak için de ilgili thread içerisinde:
- wait() API’si kullanılır.

Şimdi ilk verdiğimiz örneği bu API’leri kullanarak yazalım:

#include <mutex>
#include <condition_variable>

bool isReady {false};
std::mutex isReadyMutex;
std::condition_variable isReadyCondVariab;

// Bekleyen thread isi bitince asagidaki API'leri cagirmali
isReadyCondVariab.notify_one();

// Ilgili isReady kosulu icin bekleyen thread de asagidaki API'yi cagirir
std::unique_lock<std::mutex> lockCV{isReadyMutex};
isReadyCondVariab.wait(lockCV);

#include <mutex>

#include <condition_variable>

bool isReady {false};

std::mutex isReadyMutex;

std::condition_variable isReadyCondVariab;

// Bekleyen thread isi bitince asagidaki API'leri cagirmali

isReadyCondVariab.notify_one();

// Ilgili isReady kosulu icin bekleyen thread de asagidaki API'yi cagirir

std::unique_lock<std::mutex> lockCV{isReadyMutex};

isReadyCondVariab.wait(lockCV);

Burada önemli bir durumdan daha bahsetmekte fayda var. Özellikle bu koşullu değişkenleri zaman zaman ilgili thread bunları tetiklemese de, öyleymiş gibi bloklamayı sonlandırabilir. Bu durumlarda ilgili koşulların tekrar kontrol edilmesinde fayda var. http://www.modernescpp.com/index.php/c-core-guidelines-be-aware-of-the-traps-of-condition-variables sayfasında bu durum örnekler ile güzel bir şekilde açıklanmış, oraya bir göz atabilirsiniz.

Şimdi çoklu thread’ tarafından kullanılabilecek bir kuyruk sınıfında koşullu değişkeni ve diğer senkronizasyon yapılarının kullanımına bir göz atalım.

template <typename T>
class SharedQueue
{
public:
        // Yıkıcı
	~SharedQueue(void)
	{
		Invalidate();
	}

	// Kuyruktaki ilk elemani, dönebilir ise out üzerinden döner. Eger dönemez ise false, öbür türlü true döner
	bool tryPop(T& out)
	{
		std::lock_guard<uMutex> lock{mMutex};
		if (mQueue.empty() || !mValid)
			return false;
		
		out = std::move(mQueue.front());
		mQueue.pop();
		return true;
	}

	// Bir önceki metottan farkli olarak eğer kuyrukta bir eleman yok ise gelene kadar bekler
	// Basarili olarak out doldurulabilirse true, obur turlu false donulur
	bool waitPop(T& out)
	{
		std::unique_lock<uMutex> lock{mMutex};
		mCondition.wait(lock, [this]()
		{
			// Yukarıda bahsettiğim beklenmedik uyanmalara karsi, buradaki gibi onlem alabilirsiniz.
			// Yani queue hala bos ve gecerli oldugu muddetce beklemeye devam
			return !mQueue.empty() || !mValid;
		});
		
		if (!mValid)
		{
			return false;
		}
		out = std::move(mQueue.front());
		mQueue.pop();
		return true;
	}

	// Kuyruga yeni bir eleman ekleyelim
	void push(T value)
	{
		std::lock_guard<uMutex> lock{mMutex};
		mQueue.push(std::move(value));
		mCondition.notify_one();
	}

	// Kuyruk bos mu kontrolu
	bool empty(void) const
	{
		std::lock_guard<uMutex> lock{mMutex};
		return mQueue.empty();
	}

	// Kuyrugu bosalt
	void clear(void)
	{
		std::lock_guard<uMutex> lock{mMutex};
		while (!mQueue.empty())
		{
			mQueue.pop();
		}
		mCondition.notify_all();
	}

	// Kuyrugu tasarladigimiz yaklasimdan oturu. Kuyrugu yok etmeden once veya uygulamadan cikmadan once
	// herhangi bir bekleyen thread var ise burada onlar haberdar ediliyor. Bu cagridan sonra artik
	// kuyruk kullanılmaz hale gelir.
	void invalidate(void)
	{
		std::lock_guard<uMutex> lock{ mMutex };
		mValid = false;
		mCondition.notify_all();
	}

	// Mevcut kuyruk kullanılabilir mi, degil mi?
	bool isValid(void) const
	{
		std::lock_guard<uMutex> lock{ mMutex };
		return mValid;
	}

private:
	std::atomic_bool        mValid{ true };
	uMutex                  mMutex;
	std::queue<T>           mQueue;
	std::condition_variable mCondition;
};

template <typename T>

class SharedQueue

{

public:

// Yıkıcı

~SharedQueue(void)

{

Invalidate();

}

// Kuyruktaki ilk elemani, dönebilir ise out üzerinden döner. Eger dönemez ise false, öbür türlü true döner

bool tryPop(T& out)

{

std::lock_guard<uMutex> lock{mMutex};

if (mQueue.empty() || !mValid)

return false;

out = std::move(mQueue.front());

mQueue.pop();

return true;

}

// Bir önceki metottan farkli olarak eğer kuyrukta bir eleman yok ise gelene kadar bekler

// Basarili olarak out doldurulabilirse true, obur turlu false donulur

bool waitPop(T& out)

{

std::unique_lock<uMutex> lock{mMutex};

mCondition.wait(lock, [this]()

{

// Yukarıda bahsettiğim beklenmedik uyanmalara karsi, buradaki gibi onlem alabilirsiniz.

// Yani queue hala bos ve gecerli oldugu muddetce beklemeye devam

return !mQueue.empty() || !mValid;

});

if (!mValid)

{

return false;

}

out = std::move(mQueue.front());

mQueue.pop();

return true;

}

// Kuyruga yeni bir eleman ekleyelim

void push(T value)

{

std::lock_guard<uMutex> lock{mMutex};

mQueue.push(std::move(value));

mCondition.notify_one();

}

// Kuyruk bos mu kontrolu

bool empty(void) const

{

std::lock_guard<uMutex> lock{mMutex};

return mQueue.empty();

}

// Kuyrugu bosalt

void clear(void)

{

std::lock_guard<uMutex> lock{mMutex};

while (!mQueue.empty())

{

mQueue.pop();

}

mCondition.notify_all();

}

// Kuyrugu tasarladigimiz yaklasimdan oturu. Kuyrugu yok etmeden once veya uygulamadan cikmadan once

// herhangi bir bekleyen thread var ise burada onlar haberdar ediliyor. Bu cagridan sonra artik

// kuyruk kullanılmaz hale gelir.

void invalidate(void)

{

std::lock_guard<uMutex> lock{ mMutex };

mValid = false;

mCondition.notify_all();

}

// Mevcut kuyruk kullanılabilir mi, degil mi?

bool isValid(void) const

{

std::lock_guard<uMutex> lock{ mMutex };

return mValid;

}

private:

std::atomic_bool mValid{ true };

uMutex mMutex;

std::queue<T> mQueue;

std::condition_variable mCondition;

};

Bu örnek ile aslında çoğu senkronizasyon yapısının birlikte nasıl kullanılabileceğini de görmüş olduk.

Burada bahsettiğim koşullu değişkenleri bir kaç API’si daha var ama onların da kullanımı aynı mantık, sadece bazı ek kullanım kolaylıkları (verilen zaman ya da zamana kadar bekle gibi) sunuluyor. Onlara da bakmak için https://en.cppreference.com/w/cpp/thread/condition_variable sayfasına bir göz atabilirsiniz.

Sonuç:

Evet arkadaşlar, üç yazılık thread maceramız nihayet son buluyor. Bu üç yazı ile birlikte C++ thread kütüphanesine ilişkin en önemli yapıları sizlere aktarmaya çalıştım, umarım faydalı olmuştur. Bundan sonraki adım, tabiki bunları kullanmak 🙂 Her ne kadar kütüphaneye ilişkin anlatacaklarım bitse de, multithreaded programlamaya ilişkin bir yazım daha olacak, sonra yeni yazılara yelken açabiliriz.

Ben yazılımperver, bol ve eğlenceli kodlamalar 🙂

Paylaşmak güzeldir.

Giriş:

std::mutex:

std::condition_variable:

Sonuç:

Kaynaklar:

1 Comment → Haftalık C++ 10- std::thread (III)

Bir cevap yazın Cevabı iptal et