thread - condition variable c++

Deteniendo C++ 11 std:: hilos que esperan en una std:: condition_variable

Question

Estoy tratando de entender los mecanismos básicos de multihilo en el nuevo estándar C ++ 11. El ejemplo más básico que se me ocurre es el siguiente:

• Un productor y un consumidor se implementan en hilos separados.
• El productor coloca una cierta cantidad de artículos dentro de una cola
• El consumidor toma elementos de la cola si hay algún presente

Este ejemplo también se usa en muchos libros escolares sobre subprocesos múltiples y todo sobre el proceso de comunicación funciona bien. Sin embargo, tengo un problema cuando se trata de detener el hilo del consumidor.

Quiero que el consumidor se ejecute hasta que reciba una señal de detención explícita (en la mayoría de los casos esto significa que espero que el productor termine para poder detener al consumidor antes de que finalice el programa). Desafortunadamente, los subprocesos de C ++ 11 carecen de un mecanismo de interrupción (que conozco por multiproceso en Java, por ejemplo) Por lo tanto, tengo que usar indicadores como isRunning para indicar que quiero que un hilo se detenga.

El principal problema ahora es: después de que haya detenido el subproceso de producción, la cola está vacía y el consumidor está esperando una condition_variable para obtener una señal cuando la cola se llene nuevamente. Por lo tanto, necesito reactivar el hilo llamando a notify_all() en la variable antes de salir.

He encontrado una solución de trabajo, pero parece algo desordenada. El código de ejemplo se lista a continuación (lo siento, pero de alguna manera no pude reducir el tamaño del código más lejos por un ejemplo mínimo "mínimo"):

La clase de cola:

class Queue{
public:
    Queue() : m_isProgramStopped{ false } { }

    void push(int i){
        std::unique_lock<std::mutex> lock(m_mtx);
        m_q.push(i);
        m_cond.notify_one();
    }

    int pop(){
        std::unique_lock<std::mutex> lock(m_mtx);
        m_cond.wait(lock, [&](){ return !m_q.empty() || m_isProgramStopped; });

        if (m_isProgramStopped){
            throw std::exception("Program stopped!");
        }

        int x = m_q.front();
        m_q.pop();

        std::cout << "Thread " << std::this_thread::get_id() << " popped " << x << "." << std::endl;
        return x;
    }

    void stop(){
        m_isProgramStopped = true;
        m_cond.notify_all();
    }

private:
    std::queue<int> m_q;
    std::mutex m_mtx;
    std::condition_variable m_cond;
    bool m_isProgramStopped;
};

El productor:

class Producer{
public:
    Producer(Queue & q) : m_q{ q }, m_counter{ 1 } { }

    void produce(){
        for (int i = 0; i < 5; i++){
            m_q.push(m_counter++);
            std::this_thread::sleep_for(std::chrono::milliseconds{ 500 });
        }
    }

    void execute(){
        m_t = std::thread(&Producer::produce, this);
    }

    void join(){
        m_t.join();
    }

private:
    Queue & m_q;
    std::thread m_t;

    unsigned int m_counter;
};

El consumidor:

class Consumer{
public:
    Consumer(Queue & q) : m_q{ q }, m_takeCounter{ 0 }, m_isRunning{ true }
    { }

    ~Consumer(){
        std::cout << "KILL CONSUMER! - TOOK: " << m_takeCounter << "." << std::endl;
    }

    void consume(){
        while (m_isRunning){
            try{
                m_q.pop();
                m_takeCounter++;
            }
            catch (std::exception e){
                std::cout << "Program was stopped while waiting." << std::endl;
            }
        }
    }

    void execute(){
        m_t = std::thread(&Consumer::consume, this);
    }

    void join(){
        m_t.join();
    }

    void stop(){
        m_isRunning = false;
    }

private:
    Queue & m_q;
    std::thread m_t;

    unsigned int m_takeCounter;
    bool m_isRunning;
};

Y finalmente el main() :

int main(void){
    Queue q;

    Consumer cons{ q };
    Producer prod{ q };

    cons.execute();
    prod.execute();

    prod.join();

    cons.stop();
    q.stop();

    cons.join();

    std::cout << "END" << std::endl;

    return EXIT_SUCCESS;
}

¿Es esta la forma correcta de terminar un hilo que está esperando una variable de condición o hay mejores métodos? Actualmente, la cola necesita saber si el programa se ha detenido (lo que, en mi opinión, destruye el acoplamiento suelto de los componentes) y necesito llamar a stop() en la cola explícitamente, lo que no parece correcto.

Además, la variable de condición que solo debe usarse como singal si la cola está vacía ahora significa otra condición, si el programa ha finalizado. Si no me equivoco, cada vez que un subproceso espera en una variable de condición para que ocurra algún evento, también debería verificar si el subproceso debe detenerse antes de continuar con su ejecución (lo que también parece ser incorrecto).

¿Tengo estos problemas porque mi diseño completo es defectuoso o me faltan algunos mecanismos que se pueden usar para salir de los subprocesos de forma limpia?

Answer 1

No, no hay nada malo con su diseño, y es el enfoque normal adoptado para este tipo de problema.

Es perfectamente válido para usted tener múltiples condiciones (por ejemplo, cualquier cosa en la cola o detención del programa) adjunta a una variable de condición. La clave es que los bits en la condición se comprueban cuando vuelve la wait .

En lugar de tener una bandera en la Queue para indicar que el programa se está deteniendo, debe pensar en la bandera como "puedo aceptar". Este es un mejor paradigma general y funciona mejor en un entorno de subprocesos múltiples.

Además, en lugar de que se haya lanzado una excepción si alguien la llama y se ha llamado a stop , podría reemplazar el método con bool try_pop(int &value) que devolverá true si se devolvió un valor, de lo contrario false . De esta manera, la persona que llama puede verificar si no se ha detenido la cola (agregue un bool is_stopped() const ). Aunque el manejo de excepciones funciona aquí, es un poco pesado y no es realmente un caso excepcional en un programa de subprocesos múltiples.

Answer 2

wait puede ser llamado con un tiempo de espera. Se devuelve el control al hilo y se puede comprobar la stop . Dependiendo de ese valor, puede wait más artículos para ser consumidos o terminar la ejecución. Una buena introducción al multiprocesamiento con c ++ es C ++ 11 Concurrencia .