使用 pthread 并发推送(）到共享队列

Concurrently push()ing to a shared queue with pthread?

本文关键字：共享 队列 pthread 并发 使用      更新时间：2023-10-16

我正在练习pthread。

在我的原始程序中，它push共享队列一个名为 request 的类的实例，但我首先要确保我将某些内容推送到共享队列。

这是一个非常简单的代码，但它只是抛出了很多我无法找出原因的错误。

我想这可能是语法，但无论我尝试什么都没有用。

你明白为什么它不起作用吗？

以下是我一直在尝试的代码。

extern "C" {
    #include<pthread.h>
    #include<unistd.h>
}
#include<queue>
#include<iostream>
#include<string>
using namespace std;
class request {
 public:
    string req;
    request(string s) : req(s) {}
};
int n;
queue<request> q;
pthread_mutex_t mut = PTHREAD_MUTEX_INITIALIZER;
void * putToQueue(string);
int main ( void ) {
    pthread_t t1, t2;
    request* ff = new request("First");
    request* trd = new request("Third");
    int result1 = pthread_create(&t1, NULL, &putToQueue, reinterpret_cast<void*>(&ff));
    if (result1 != 0) cout << "error 1" << endl;
    int result2 = pthread_create(&t2, NULL, &putToQueue, reinterpret_cast<void*>(&trd));
    if (result2 != 0) cout << "error 2" << endl;
    pthread_join(t1, NULL);
    pthread_join(t2, NULL);
    for(int i=0; i<q.size(); ++i) {
        cout << q.front().req << " is in queue" << endl;
        q.pop();
        --n;
    }
    return 0;
}
void * putToQueue(void* elem) {
    pthread_mutex_lock(&mut);
    q.push(reinterpret_cast<request>(elem));
    ++n;
    cout << n << " items are in the queue." << endl;
    pthread_mutex_unlock(&mut);
    return 0;
}

下面的代码注释了必须更改的所有内容。我会详细描述为什么他们必须改变，但我希望代码不言自明。它仍然不是防弹的。有很多事情可以做得不同或更好（失败new的异常处理等），但至少它可以编译、运行并且不会泄漏内存。

#include <queue>
#include <iostream>
#include <string>
#include <pthread.h>
#include <unistd.h>
using namespace std;
// MINOR: param should be a const-ref
class request {
public:
    string req;
    request(const string& s) : req(s) {}
};
int n;
queue<request> q;
pthread_mutex_t mut = PTHREAD_MUTEX_INITIALIZER;

// FIXED: made protoype a proper pthread-proc signature
void * putToQueue(void*);
int main ( void )
{
    pthread_t t1, t2;
    // FIXED: made thread param the actual dynamic allocation address
    int result1 = pthread_create(&t1, NULL, &putToQueue, new request("First"));
    if (result1 != 0) cout << "error 1" << endl;
    // FIXED: made thread param the actual dynamic allocation address
    int result2 = pthread_create(&t2, NULL, &putToQueue, new request("Third"));
    if (result2 != 0) cout << "error 2" << endl;
    pthread_join(t1, NULL);
    pthread_join(t2, NULL);
    // FIXED: was skipping elements because the queue size was shrinking
    //  with each pop in the while-body.
    while (!q.empty())
    {
        cout << q.front().req << " WAS in queue" << endl;
        q.pop();
    }
    return 0;
}
// FIXED: pretty much a near-total-rewrite
void* putToQueue(void* elem)
{
    request *req = static_cast<request*>(elem);
    if (pthread_mutex_lock(&mut) == 0)
    {
        q.push(*req);
        cout << ++n << " items are in the queue." << endl;
        pthread_mutex_unlock(&mut);
    }
    delete req; // FIXED: squelched memory leak
    return 0;
}
输出

（您的输出可能会有所不同）

1 items are in the queue.
2 items are in the queue.
Third WAS in queue
First WAS in queue

如评论中所述，我建议跳过直接使用 pthreads，改用 C++11 线程原语。我将从一个简单的受保护队列类开始：

template <class T, template<class, class> class Container=std::deque>
class p_q {
    typedef typename Container<T, std::allocator<T>> container;
    typedef typename container::iterator iterator;
    container data;
    std::mutex m;
public:
    void push(T a) {
        std::lock_guard<std::mutex> l(m);
        data.emplace_back(a);
    }
    iterator begin() { return data.begin(); }
    iterator end() { return data.end(); }
    // omitting front() and pop() for now, because they're not used in this code
};
使用它，代码

的主流几乎与单线程代码一样简单和干净，如下所示：

int main() {
    p_q<std::string> q;
    auto pusher = [&q](std::string const& a) { q.push(a); };
    std::thread t1{ pusher, "First" };
    std::thread t2{ pusher, "Second" };
    t1.join();
    t2.join();
    for (auto s : q)
        std::cout << s << "n";
}

就目前而言，这是一个多生产者、单一消费者的队列。此外，这取决于生产者在消费发生时不再运行的事实。在这种情况下确实如此，但不会/不会总是如此。如果不是这种情况，您将需要一个（稍微）更复杂的队列，该队列在从队列中读取/弹出时执行锁定，而不仅仅是在写入队列时。

相关文章：