Wielowątkowe przesyłanie wiadomości w języku C++

Question

Mam za zadanie zmodyfikować synchroniczny program C, aby mógł działać równolegle. Celem jest zapewnienie maksymalnej mobilności, ponieważ jest to program open source, z którego korzysta wiele osób. Z tego powodu pomyślałem, że najlepiej będzie zawinąć program w warstwie C++, aby móc korzystać z przenośnych bibliotek wspomagających. Zrobiłem to już i wszystko wydaje się działać zgodnie z oczekiwaniami.

Problem, który mam, polega na wyborze najlepszego sposobu przekazywania wiadomości między wątkami. Na szczęście architektura programu jest wieloproducentem i pojedynczym konsumentem. Co więcej, kolejność wiadomości nie jest ważna. Czytałem, że koleje jednoprodukcyjne/jedno-konsumenckie (SPSC) skorzystałyby z tej architektury. Ci doświadczeni w programowaniu wielowątkowym mają jakąkolwiek radę? Jestem całkiem nowy w tych rzeczach. Również wszelkie przykłady kodu wykorzystujące doładowanie w celu wdrożenia SPSC będą bardzo mile widziane.

Answer 1

Poniżej znajduje się technika, której użyłem w mojej bibliotece współpracy wielozadaniowej/wielowątkowej (MACE) http://bytemaster.github.com/mace/. Ma tę zaletę, że jest wolny od blokady, z wyjątkiem sytuacji, gdy kolejka jest pusta.

struct task { 
    boost::function<void()> func; 
    task* next; 
}; 


boost::mutex      task_ready_mutex; 
boost::condition_variable  task_ready; 
boost::atomic<task*>    task_in_queue; 

// this can be called from any thread 
void thread::post_task(task* t) { 
    // atomically post the task to the queue. 
    task* stale_head = task_in_queue.load(boost::memory_order_relaxed); 
    do { t->next = stale_head; 
    } while(!task_in_queue.compare_exchange_weak(stale_head, t, boost::memory_order_release)); 

    // Because only one thread can post the 'first task', only that thread will attempt 
    // to aquire the lock and therefore there should be no contention on this lock except 
    // when *this thread is about to block on a wait condition. 
    if(!stale_head) { 
     boost::unique_lock<boost::mutex> lock(task_ready_mutex); 
     task_ready.notify_one(); 
    } 
} 

// this is the consumer thread. 
void process_tasks() { 
    while(!done) { 
    // this will atomically pop everything that has been posted so far. 
    pending = task_in_queue.exchange(0,boost::memory_order_consume); 
    // pending is a linked list in 'reverse post order', so process them 
    // from tail to head if you want to maintain order. 

    if(!pending) { // lock scope 
     boost::unique_lock<boost::mutex> lock(task_ready_mutex);     
     // check one last time while holding the lock before blocking. 
     if(!task_in_queue) task_ready.wait(lock); 
    } 
} 

Answer 2

Jeśli istnieje tylko jeden konsument, ale wielu producentów, to użyłbym tablicy lub struktury danych podobnej do tablicy z czasem dostępu O (1), gdzie każde gniazdo tablicy reprezentuje kolejkę jednego producenta-konsumenta. Wielką zaletą kolejki pojedynczego producenta-konsumenta jest fakt, że można go zablokować bez żadnych wyraźnych mechanizmów synchronizacji, co czyni go bardzo szybką strukturą danych w środowisku wielowątkowym. Zobacz my answer here, aby uzyskać pełną wersję kolejki jednego producenta-konsumenta.

Answer 3

Istnieje wiele przykładów kolejek producent-konsument w sieci, bezpiecznych dla wielu producentów/konsumentów. @bytemaster opublikował taki, który używa odsyłacza wewnątrz każdej wiadomości, aby wyeliminować pamięć w samej klasie kolejki - to dobre podejście, używam go sam na osadzonych zadaniach.

Tam, gdzie klasa kolejkowa musi zapewniać pamięć, zwykle uruchamiam "kolejkę puli" o rozmiarze N, załadowaną instancjami klasy N * wiadomości podczas uruchamiania. Wątki wymagające komunikacji muszą wysłać wiadomość * z puli, załadować ją i umieścić w kolejce. Kiedy w końcu "zużyta" wiadomość * zostanie zepchnięta z powrotem do puli. Ogranicza to liczbę wiadomości, więc wszystkie kolejki muszą mieć tylko długość N - bez zmiany rozmiaru, bez nowych(), bez usuwania(), łatwe wykrywanie wycieków.