C++ pthread bloqueo de interbloqueo de cola (creo)

Question

Estoy teniendo un problema con pthreads donde creo que me estoy quedando estancado. Creé una cola de bloqueo que pensé que estaba funcionando, pero después de hacer más pruebas descubrí que si intento cancelar varios hilos que están bloqueando en la secuencia de bloqueo, parece que tengo un punto muerto.

La cola de bloqueo es muy simple y se ve así:

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; 
} 

Para la prueba, he creado 4 hilos que tiran en esta cola de bloqueo. Agregué algunas declaraciones de impresión a la cola de bloqueo y cada hilo está llegando al método pthread_cond_wait(). Sin embargo, cuando intento llamar a pthread_cancel() y pthread_join() en cada subproceso, el programa simplemente cuelga.

También he probado esto con solo un hilo y funciona perfectamente.

De acuerdo con la documentación, pthread_cond_wait() es un punto de cancelación, por lo que llamar cancelar en esos hilos debería hacer que dejen de ejecutarse (y esto solo funciona con 1 hilo). Sin embargo, pthread_mutex_lock no es un punto de cancelación. ¿Podría estar sucediendo algo en la línea de cuando se llama pthread_cancel(), el hilo cancelado adquiere el mutex antes de terminar y no lo desbloquea, y luego, cuando se cancela el siguiente hilo, no puede adquirir el mutex y los bloqueos? O hay algo más que estoy haciendo mal.

Cualquier consejo sería encantador. Gracias :)

Answer 1

pthread_cancel() es mejor evitarlo.

Puede desbloquear todos los hilos bloqueados en Blocking_Queue :: pull() lanzando una excepción desde allí.

Un punto débil en la cola es que T t = _queue.front(); invoca el constructor de copia de T que puede arrojar una excepción, lo que hace que la cola mutezca bloqueada para siempre. Mejor uso bloqueos de ámbito C++.

Aquí es un ejemplo de la terminación de subprocesos graciosa:

$ 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

No estoy exactamente familiarizado con pthread_cancel() - Prefiero la terminación cooperativa.

¿No sería un pthread_cancel() dejar su mutex bloqueado? Supongo que debe limpiar con un controlador de cancelación.

Answer 3

He tenido una experiencia similar con pthread_cond_wait()/pthread_cancel(). Tuve problemas con un bloqueo que todavía se retenía después de que el hilo volviera por alguna razón, y era imposible desbloquearlo, ya que tienes que desbloquear el mismo hilo que bloqueaste. Observé estos errores al hacer pthread_mutex_destroy() ya que tenía un solo productor, una sola situación de consumidor, por lo que no se produjo el interbloqueo.

pthread_cond_wait() se supone que bloquea el mutex al regresar, y esto podría haber sucedido, pero el desbloqueo final no se realizó ya que cancelamos el hilo por la fuerza. Por seguridad, generalmente trato de evitar el uso de pthread_cancel() ya que algunas plataformas ni siquiera lo soportan. Puedes usar un bool volátil o átomos atómicos y comprobar si el hilo debe cerrarse. De esa forma, los mutexes se manejarán limpiamente también.