C5 pthread blokowanie kolejka impasu (myślę)

Question

Mam problem z pthreads, gdzie myślę, że dostaję impasu. Stworzyłem kolejkę blokującą, która moim zdaniem działała, ale po wykonaniu kilku testów odkryłem, że jeśli spróbuję anulować wiele wątków blokujących blokowanie, wydaje się, że dostaję zakleszczenia.

Kolejka blokowanie jest bardzo prosty i wygląda następująco:

template <class T> class Blocking_Queue 
{ 
public: 
    Blocking_Queue() 
    { 
     pthread_mutex_init(&_lock, NULL); 
     pthread_cond_init(&_cond, NULL); 
    } 

    ~Blocking_Queue() 
    { 
     pthread_mutex_destroy(&_lock); 
     pthread_cond_destroy(&_cond); 
    } 

    void put(T t) 
    { 
     pthread_mutex_lock(&_lock); 
     _queue.push(t); 
     pthread_cond_signal(&_cond); 
     pthread_mutex_unlock(&_lock); 
    } 

    T pull() 
    { 
     pthread_mutex_lock(&_lock); 
     while(_queue.empty()) 
     { 
      pthread_cond_wait(&_cond, &_lock); 
     } 

     T t = _queue.front(); 
     _queue.pop(); 

     pthread_mutex_unlock(&_lock); 

     return t; 
    } 

priavte: 
    std::queue<T> _queue; 
    pthread_cond_t _cond; 
    pthread_mutex_t _lock; 
} 

Do testowania Stworzyłem 4 wątki, które ciągnąć na tej kolejki blokującej. Dodałem kilka instrukcji drukowania do kolejki blokującej, a każdy wątek przechodzi do metody pthread_cond_wait(). Jednak gdy próbuję wywołać pthread_cancel() i pthread_join() w każdym wątku program po prostu się zawiesi.

Testowałem to również z jednym tylko wątkiem i działa idealnie.

Zgodnie z dokumentacją, pthread_cond_wait() jest punktem anulowania, więc wywołanie cancel na tych wątkach powinno spowodować, że przestaną wykonywać (i to działa tylko z jednym wątkiem). Jednak pthread_mutex_lock nie jest punktem anulowania. Czy coś może się zdarzyć w czasie, gdy zostanie wywołana funkcja pthread_cancel(), anulowany wątek pobiera muteks przed zakończeniem i nie odblokowuje go, a następnie, gdy następny wątek zostanie anulowany, nie może uzyskać dostępu do muteksu i zakleszczeń? Czy jest coś jeszcze, co robię źle.

Każda rada byłaby piękna. Dzięki :)

Answer 1

pthread_cancel() najlepiej unikać.

Możesz odblokować wszystkie wątki zablokowane w Blocking_Queue :: pull() przez zgłoszenie wyjątku.

Jednym ze słabych punktów w kolejce jest to, że T t = _queue.front(); wywołuje konstruktora kopii T, który może rzucić wyjątek, powodując, że muteks w kolejce zostanie zablokowany na zawsze. Lepiej użyj blokad o zasięgu C++.

Oto przykład płynnego zakończenia wątku:

$ cat test.cc 
#include <boost/thread/mutex.hpp> 
#include <boost/thread/thread.hpp> 
#include <boost/thread/condition_variable.hpp> 
#include <exception> 
#include <list> 
#include <stdio.h> 

struct BlockingQueueTerminate 
    : std::exception 
{}; 

template<class T> 
class BlockingQueue 
{ 
private: 
    boost::mutex mtx_; 
    boost::condition_variable cnd_; 
    std::list<T> q_; 
    unsigned blocked_; 
    bool stop_; 

public: 
    BlockingQueue() 
     : blocked_() 
     , stop_() 
    {} 

    ~BlockingQueue() 
    { 
     this->stop(true); 
    } 

    void stop(bool wait) 
    { 
     // tell threads blocked on BlockingQueue::pull() to leave 
     boost::mutex::scoped_lock lock(mtx_); 
     stop_ = true; 
     cnd_.notify_all(); 

     if(wait) // wait till all threads blocked on the queue leave BlockingQueue::pull() 
      while(blocked_) 
       cnd_.wait(lock); 
    } 

    void put(T t) 
    { 
     boost::mutex::scoped_lock lock(mtx_); 
     q_.push_back(t); 
     cnd_.notify_one(); 
    } 

    T pull() 
    { 
     boost::mutex::scoped_lock lock(mtx_); 

     ++blocked_; 
     while(!stop_ && q_.empty()) 
      cnd_.wait(lock); 
     --blocked_; 

     if(stop_) { 
      cnd_.notify_all(); // tell stop() this thread has left 
      throw BlockingQueueTerminate(); 
     } 

     T front = q_.front(); 
     q_.pop_front(); 
     return front; 
    } 
}; 

void sleep_ms(unsigned ms) 
{ 
    // i am using old boost 
    boost::thread::sleep(boost::get_system_time() + boost::posix_time::milliseconds(ms)); 
    // with latest one you can do this 
    //boost::thread::sleep(boost::posix_time::milliseconds(10)); 
} 

void thread(int n, BlockingQueue<int>* q) 
try 
{ 
    for(;;) { 
     int m = q->pull(); 
     printf("thread %u: pulled %d\n", n, m); 
     sleep_ms(10); 
    } 
} 
catch(BlockingQueueTerminate&) 
{ 
    printf("thread %u: finished\n", n); 
} 

int main() 
{ 
    BlockingQueue<int> q; 

    // create two threads 
    boost::thread_group tg; 
    tg.create_thread(boost::bind(thread, 1, &q)); 
    tg.create_thread(boost::bind(thread, 2, &q)); 
    for(int i = 1; i < 10; ++i) 
     q.put(i); 
    sleep_ms(100); // let the threads do something 
    q.stop(false); // tell the threads to stop 
    tg.join_all(); // wait till they stop 
} 

$ g++ -pthread -Wall -Wextra -o test -lboost_thread-mt test.cc 

$ ./test 
thread 2: pulled 1 
thread 1: pulled 2 
thread 1: pulled 3 
thread 2: pulled 4 
thread 1: pulled 5 
thread 2: pulled 6 
thread 1: pulled 7 
thread 2: pulled 8 
thread 1: pulled 9 
thread 2: finished 
thread 1: finished 

Answer 2

Nie jestem dokładnie obeznany z pthread_cancel() - Wolę wspólne zakończenie.

Czy pthread_cancel() nie zostawi zablokowanego Mutexa? Przypuszczam, że musisz wyczyścić za pomocą programu obsługi anulowania.

Answer 3

Miałem podobne doświadczenie z pthread_cond_wait()/pthread_cancel(). Miałem problemy z zablokowaniem blokady po tym, jak wątek został zwrócony z jakiegoś powodu, i nie można było go odblokować, ponieważ musisz odblokować w tym samym wątku, który zablokowałeś. Zauważyłem te błędy podczas wykonywania pthread_mutex_destroy(), ponieważ miałem jednego producenta, sytuację jednego konsumenta, więc nie doszło do impasu.

pthread_cond_wait() powinien blokować muteks podczas powrotu, a to mogło się stać, ale ostateczne odblokowanie nie zostało przerwane, ponieważ zdecydowanie anulowaliśmy wątek. Ze względów bezpieczeństwa staram się ogólnie unikać pthread_cancel(), ponieważ niektóre platformy tego nie obsługują. Możesz użyć lotnego boolu lub atomu i sprawdzić, czy wątek powinien zostać zamknięty. W ten sposób muteksy będą również obsługiwane w sposób czysty.