c++ - 具有pthread和锁且无提升的单读者多作者

Question

考虑下一段代码。

#include <iostream>
#include <vector>
#include <map>

using namespace std;

map<pthread_t,vector<int>> map_vec;
vector<pair<pthread_t ,int>> how_much_and_where;

pthread_cond_t CV = PTHREAD_COND_INITIALIZER;
pthread_mutex_t mutex = PTHREAD_MUTEX_INITIALIZER;

void* writer(void* args)
{
    while(*some condition*)
    {
        int howMuchPush = (rand() % 5) + 1;
        for (int i = 0; i < howMuchPush; ++i)
        {
            // WRITE
            map_vec[pthread_self()].push_back(rand() % 10);
        }

        how_much_and_where.push_back(make_pair(pthread_self(), howMuchPush));
        // Wake up the reader - there's something to read.
        pthread_cond_signal(&CV);
    }

    cout << "writer thread: " <<  pthread_self()  << endl;
    return nullptr;
}

void* reader(void* args) {

    pair<pthread_t, int> to_do;

    pthread_cond_wait(&CV, &mutex);
    while(*what condition??*)
    {
        to_do = how_much_and_where.front();
        how_much_and_where.erase(how_much_and_where.begin());

        // READ
        cout << to_do.first << " wrote " << endl;
        for (int i = 0; i < to_do.second; i++)
        {
            cout << map_vec[to_do.first][i] << endl;
        }

        // Done reading. Go to sleep.
        pthread_cond_wait(&CV, &mutex);
    }

    return nullptr;
}

//----------------------------------------------------------------------------//


int main()
{
    pthread_t threads[4];

    // Writers
    pthread_create(&threads[0], nullptr, writer, nullptr);
    pthread_create(&threads[1], nullptr, writer, nullptr);
    pthread_create(&threads[2], nullptr, writer, nullptr);
    // reader
    pthread_create(&threads[4], nullptr, reader, nullptr);


    pthread_join(threads[0], nullptr);
    pthread_join(threads[1], nullptr);
    pthread_join(threads[2], nullptr);
    pthread_join(threads[3], nullptr);

    return 0;
}

背景

每个作者都有自己的容器来向其写入数据。
并假设有一个读者知道写者何时完成写数据块，以及该数据块的大小是多少(读者有一个容器，写者可以将成对的数据写到该容器中)。

问题

显然，我应该锁定共享源-map_vec和how_much_and_where。但在这种情况下，我不知道-
在此资源上定位锁定的有效方法(例如，在for循环中的每个map_vec之前锁定push_back？还是在for循环之前锁定它-但是不将其推送到队列是浪费且漫长的操作，这可能会导致读者要等待太多吗？)/
安全定位锁的方法，以防止死锁。

我不知道应该在什么正确的条件下
while循环-我认为也许只要how_much_and_where是
不是空的，而是很明显的一种情况，在这种情况下，读者可能会在作家添加一对之前就清空how_much_and_where。

假设某个作家在读者忙于阅读某些东西时发送了一个信号
数据。据我了解，此信号将被忽略，并且
可能永远无法处理作家所插入的一对(#of信号
并为读者处理<#of对\任务。我怎样才能
防止这种情况？

Answer 1

为简化起见，我们应该将通用/可重复使用的生产者-消费者队列(或我通常称之为“阻塞队列”)的实现与实际生产者和消费者(非通用)的实现分离开来。/可重用-它们特定于您的程序)。从设计的角度来看，这将使代码更加清晰和易于管理。

1.实现通用(可重用)阻塞队列

首先，您应该实现一个“阻塞队列”，该队列可以管理多个生产者和一个消费者。该阻塞队列将包含处理多线程/同步的代码，并且使用者线程可以使用它来接收来自多个生产者线程的项目。可以使用许多不同的方式(不仅是互斥锁+ cond组合)来实现这种阻塞队列，这取决于您是否拥有1个或多个消费者和1个或多个生产者(有时可以引入不同种类的[特定于平台的]优化)当您只有1个消费者或1个生产者时)。如果需要，最简单的使用互斥锁+条件对的队列实现可自动处理多个生产者和多个消费者。

队列只有一个内部容器(可以是非线程安全的std::queue，vector或list)，用于容纳项以及相关联的互斥体+偶数对，可以保护该容器免于同时访问多个线程。队列必须提供两个操作:

produce(item):将一项放入队列并立即返回。伪代码如下所示:

锁定互斥锁

将新项目添加到内部容器

通过cond 发出

信号

解锁互斥锁

返回

wait_and_get():如果队列中至少有一项，则它将删除最旧的一项并立即返回，否则它会等待有人通过produce(item)操作将一项放入队列。

锁定互斥锁

如果容器为空:

等待cond(pthread_cond_wait)

删除最旧的项目

解锁互斥锁

返回删除的最旧项目

2.使用阻塞队列实现程序

现在您有了一个可重用的阻塞队列，我们​​可以实现生产者和使用者以及控制事物的主线程。

生产者

他们只是将一堆物品扔进队列(通过调用阻塞队列的 produce(item))，然后退出。如果项目的生产不占用大量计算或不需要等待大量的IO操作，那么在示例程序中，这将很快完成。为了模拟线程在其中进行繁重工作的真实世界场景，您可以执行以下操作:在每个生产者线程上，您仅将X个(说5个)数量的项目放入队列，但是在每个项目之间您等待了随机的秒数在1到3秒之间。请注意，一段时间后，生产者线程会在完成工作时自行退出。

消费者

使用者有一个无限循环，在该循环中，它始终使用 wait_and_get()从队列中获取下一个项目，并以某种方式对其进行处理。如果它是一个特殊的项目，它表示处理结束，那么它将跳出无限循环，而不是处理该项目。伪代码:

无限循环:

从队列中获取下一个项目(wait_and_get())

如果这是表示处理结束的特殊项目，则跳出循环...

否则让我们处理这个项目

主线程

以任何顺序启动所有线程，包括生产者和使用者。

等待所有生产者线程完成(用pthread_join()编码)。

请记住，生产者会在没有外部刺激的情况下结束并自行退出。当您加入所有生产者后，这意味着每个生产者都已退出，因此没有人会再次调用队列的produce(item)操作。但是，队列中可能仍然有未处理的项目，并且消费者仍可以对这些项目进行处理。

将最后一个特殊的“处理结束”项目放入使用者的队列。

当消费者处理完生产者生产的最后一个商品时，它仍将使用wait_and_get()向队列询问下一个商品-由于等待下一个永远不会到达的商品，这可能导致死锁。为了在主线程上提供帮助，我们将最后一个特殊项目放在队列中，该队列向使用者发出信号，表明处理结束。请记住，我们的消费者实现包含对此特殊项目的检查，以找出何时完成处理。重要的是，只有在生产者完成之后(加入他们之后)，才必须将这个特殊项放到主线程上的队列中!

如果您有多个使用者，则将多个特殊的“处理结束”项目放入队列(每个使用者1个)比使队列更聪明以仅使用一个“处理结束”项目来处理多个使用者要容易得多。由于主线程负责协调整个过程(线程创建，线程连接等)，因此它确切知道使用者的数量，因此很容易将相同数量的“处理结束”项目放入队列。

等待使用者线程通过加入而终止。

将处理结束的特殊项目放入队列后，我们等待消费者线程处理剩余的项目(由生产者生产)以及我们的最后一个特殊项目(由主“协调者”线程生产)，该特殊项目要求消费者结束。我们在使用者线程中通过pthread_join()在主线程上等待。

附加条款:

在我的线程系统实现中，阻塞队列的项目通常是指针-指向必须执行/处理的“作业”对象的指针。 (您可以将阻塞队列实现为模板类，在这种情况下，阻塞队列的用户可以指定项目的类型)。在我的情况下，很容易将特殊的“处理结束”项目放到消费者的队列中:为此，我通常使用一个简单的NULL作业指针。在您的情况下，您将必须找出可以在队列中使用的哪种特殊值，以表示对消费者的处理结束。

生产者可能有自己的队列和一堆其他数据结构，他们可以玩这些数据结构来“生产项目”，但是消费者并不关心这些数据结构。消费者只关心通过其自己的阻塞队列收到的单个物品。如果生产者想要从消费者那里得到一些东西，那么它必须通过队列将一个项目(“工作”)发送给消费者。阻塞队列实例属于使用者线程-它在任意线程和使用者线程之间提供单向通信 channel 。甚至使用者线程本身也可以将项目放入自己的队列中(在某些情况下，这很有用)。

pthread_cond_wait文档说，此功能可以在没有实际信号的情况下唤醒(尽管我从未见过此功能的虚假唤醒引起的单个错误)。为此，应将代码中的if container is empty then pthread_cond_wait部分替换为while the container is empty pthread_cond_wait，但同样，这种虚假的唤醒可能只是一个懒散的怪物，仅在某些具有线程原语的特定linux实现的体系结构中才存在，因此您的代码可能会在没有桌面线程的台式机上工作关心这个问题。