C++ 多线程 - 具有条件变量的通道答案

【问题标题】：C++ Multithreading - Channels with Condition VariablesC++ 多线程 - 具有条件变量的通道
【发布时间】：2018-09-10 12:13:44
【问题描述】：

我正在尝试在我的多线程程序中包含条件变量和通道，并制作了一个基本程序来尝试了解它们的工作原理。在这个程序中，一个线程将数字 0 到 9 添加到通道缓冲区，另一个线程将显示每个数字并从缓冲区中弹出。

目前，程序运行，但没有显示任何内容。我怀疑线程正在等待资源，因此进入了死锁，但我不确定如何解决这个问题。

Source.cpp（调用线程）：

#include "channel.h"
#include <iostream>
channel channel1;

void function1() {
    for (int i = 0; i < 10; i++) {
        channel1.write(to_string(i));
    }
}

void function2() {
    string val;
    for (int i = 0; i < 10; i++) {
        val = channel1.read();
        cout << val << "\n";
    }
}

    void main() {
        thread t1(function1);
        thread t2(function2);
        t1.join();
    t2.join();
    return;
}

channel.h（写入/读取缓冲区的方法）：

#pragma once
#include <mutex>
#include <list>
#include <string>
using namespace std;

typedef unique_lock<mutex> mutex_lock;
class channel {
public:
    list<string> buffer;
    mutex buffer_mutex; // controls access to buffer
    condition_variable cv;

    void write(string data) {
        mutex_lock lock(buffer_mutex);
        buffer.push_back(data);
        cv.notify_all();
    }

    string read() {
        string item = "";
        while (item == "") {
            mutex_lock lock(buffer_mutex);
            cv.wait(lock);
            string item = buffer.front();
            buffer.pop_front();
            return item;
        }
    }
};

非常感谢任何帮助:)

【问题讨论】：

考虑当第一个线程设法将所有内容都塞入队列时会发生什么，并且甚至在另一个线程到达read() 之前就非常迅速地向条件变量发出信号。第一个线程向条件变量发出信号，但没有任何等待，但没关系。现在，另一个线程最终进入read()，然后启动wait()ing 以通知条件变量。现在，您希望该信号来自哪里？在我们的太阳爆炸之前，没有任何东西会发出那个条件变量的信号。想想吧。
与往常一样，头文件中全局范围内的 using 指令是一种攻击行为。让你知道。

标签： c++ multithreading channel condition-variable

【解决方案1】：

看到这段代码，我引入了一个bool data_avail 来明确意图，一个时间延迟以便编写器不会一直锁定，并且while item != "" 被删除，因为它被认为是不必要的。

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

using namespace std;

typedef unique_lock<mutex> mutex_lock;

class channel {
public:
    list<string> buffer;
    mutex buffer_mutex; // controls access to buffer
    condition_variable cv;
    bool data_avail = false;

    void write(string data) {
        mutex_lock lock(buffer_mutex);
        buffer.push_back(data);
        data_avail = true;
        cv.notify_all();
    }

    string read() {
        string item ;
        mutex_lock lock(buffer_mutex);
        cv.wait(lock,[&](){ return data_avail;});
        string item = buffer.front();
        buffer.pop_front();
        data_avail = false;
        return item;
    }
};

channel channel1;

void function1() {
    for (int i = 0; i < 10; i++) {
        channel1.write(to_string(i));
        this_thread::sleep_for(chrono::milliseconds(100));
    }
}

void function2() {
    string val;
    for (int i = 0; i < 10; i++) {
        val = channel1.read();
        cout << val << "\n";

    }
}

int main() {
    thread t1(function1);
    thread t2(function2);
    t1.join();
    t2.join();
    return 0;
}

输出：

【讨论】：

【解决方案2】：

这是一个无缓冲通用通道实现。它直接来自@seccpur 的回答。

#include <iostream>
#include <thread>
#include <condition_variable>

using namespace std;

typedef unique_lock<mutex> mutex_lock;

template<class T>
class channel {
    T buffer;
    mutex buffer_mutex; // controls access to buffer
    condition_variable read_cond;
    condition_variable write_cond;
    bool data_avail = false;

public:
    void write(T data) {
        mutex_lock lock(buffer_mutex);
        write_cond.wait(lock, [&]() { return !data_avail; });
        buffer = data;
        data_avail = true;
        read_cond.notify_all();
    }

    T read() {
        mutex_lock lock(buffer_mutex);
        read_cond.wait(lock, [&]() { return data_avail; });
        T item = buffer;
        data_avail = false;
        write_cond.notify_all();
        return item;
    }
};

int main() {
    const int LENGTH = 30;

    channel<int> channel1;

    auto function1 = [&](){
        for (int i = 0; i < LENGTH; ++i) {
            channel1.write(i);
        }
    };

    auto function2 = [&](){
        for (int i = 0; i < LENGTH; ++i) {
            std::cout << 2 * channel1.read() << std::endl;
        }
    };

    thread t1(function1);
    thread t2(function2);
    t1.join();
    t2.join();
    return 0;
}

【讨论】：