如何使用pthreads以智能方式共享變量到線程？

Question

我想用智能方式使用pthreads在C++中同步線程。

我有一個全局變量：

int Resources = 0; 

我有兩個線程函數：

void *incResources(void *arg) 
{ 
    while(1) 
    { 
    pthread_mutex_lock (&resourcesMutex); 
    Resources += 2; 
    pthread_mutex_unlock (&resourcesMutex); 
    } 

pthread_exit((void*) 0); 
} 

void *consumeResources(void *arg) 
{ 
    while(1) 
    { 
    pthread_mutex_lock (&resourcesMutex); 
    Resources--; 
    pthread_mutex_unlock (&resourcesMutex); 
    } 
    pthread_exit((void*) 0); 
} 

而且在主要功能我intialize 2個消耗線程和一個遞增線程：

pthread_mutex_init(&resourcesMutex, NULL); 

pthread_create(&callThd[0], &attr, incResources, (void *)i); 
pthread_create(&callThd[1], &attr, consumeResources, (void *)i); 
pthread_create(&callThd[2], &attr, consumeResources, (void *)i); 

我覺得這樣效率不高，可以做得更好。你能提供一些想法嗎？我試圖用等，但我沒有得到它：/ 謝謝！

Answer 1

我看好C++方法，我強烈建議閱讀C++ Concurrency in Action: by Anthony Williams，並留下pthread以便在您可以使用期貨和類似的高級別事物。如果你必須去手動線程擺弄，你也可以找到很好的例子。

您的問題陳述對於明智的建議太模糊 - 良好線程的基本思想是根本沒有共享狀態，而像您的握手情況很可能使用一些同步隊列來實現該目的。

Answer 2

更明智的做法是使用std::mutex和std::thread（或Boost等價物），因此您不需要手動解鎖互斥鎖。

條件變量將允許用戶阻止（不浪費CPU週期），直到有可用的工作對他們來說：

struct Resource 
{ 
    int value; 
    std::mutex mx; 
    std::condition_variable cv; 
}; 

void incResources(Resource& res) 
{ 
    while(1) 
    { 
    { 
     std::lock_guard<std::mutex> l(res.mx); 
     res.value += 2; 
    } 
    res.cv.notify_all(); 
    } 
} 

void consumeResources(Resource& res) 
{ 
    while(1) 
    { 
    std::unique_lock<std::mutex> l(res.mx); 
    while (res.value == 0) 
     res.cv.wait(l); 
    res.value--; 
    } 
} 

，並在主線程：

Resources res; 
res.value = 0; 
std::thread t1(incResources, std::ref(res)); 
std::thread t2(consumeResources, std::ref(res)); 
std::thread t3(consumeResources, std::ref(res)); 
// ... 
t1.join(); 
t2.join(); 
t3.join(); 

Answer 3

我想，如果您使用的是C++，沒有理由爲什麼要優先使用pthreads來代替C++ 11 std::thread和STL同步類。

如果您不能使用C++ 11標準，則應該將pthreads本機接口包裝爲合理的C++類表示（請參閱，例如boost::thread或STTCL Posix Thread實現）。

Answer 4

它看起來像你試圖實現一個生產者和消費者，+ =線程創建工作（數字被減少）和消費者帶走他們。

而不是讓消費者在這樣一個平凡的旋轉循環中，看看條件變量。

std::queue<Job*> queue; 
pthread_mutex mutex; 
pthread_cond cond; 

void AddJob(Job* job) { 
    pthread_mutex_lock(&mutex); 
    queue.push_back(job); 
    pthread_cond_signal(&cond); 
    pthread_mutex_unlock(&mutex); 
} 

void* QueueWorker(void* /*threadInfo*/) { 
    Job* job = NULL; 
    for (;;) { 
     pthread_mutex_lock(&mutex); 
     while (queue.empty()) { 
      // unlock the mutex until the cond is signal()d or broadcast() to. 
      // if this call succeeds, we will have the mutex locked again on the other side. 
      pthread_cond_wait(&cond, &mutex); 
     } 
     // take the first task and then release the lock. 
     job = queue.pop(); 
     pthread_mutex_unlock(&mutex); 

     if (job != NULL) 
      job->Execute(); 
    } 

    return NULL; 
} 

這可擴展到多個消費者。另外，儘管熟悉pthreads實現可能很有用，但您應該查看一個可用的線程包裝器。 C++ 11引入了std :: thread和std :: mutex，許多人都以boost爲代價，但是我個人發現OpenSceneGraph團隊的「OpenThreads」庫是最簡單和最優雅的工作之一。

編輯：這是一個完整的工作實現，儘管有一些人爲的機制來結束運行。

#include <queue> 
#include <pthread.h> 
#include <stdio.h> 
#include <stdlib.h> 
#include <unistd.h> 

static int jobNo = 0; 
class Job { 
public: 
    Job() : m_i(++jobNo) { printf("Created job %d.\n", m_i); } 
    int m_i; 
    void Execute() { printf("Job %d executing.\n", m_i); usleep(500 * 1000); } 
}; 

std::queue<Job*> queue; 
pthread_mutex_t mutex = PTHREAD_MUTEX_INITIALIZER; 
pthread_cond_t cond = PTHREAD_COND_INITIALIZER; 

void AddJob(Job* job) { 
    pthread_mutex_lock(&mutex); 
    queue.push(job); 
    pthread_cond_signal(&cond); 
    pthread_mutex_unlock(&mutex); 
} 

void* QueueWorker(void* /*threadInfo*/) { 
    Job* job = NULL; 
    for (;;) { 
     pthread_mutex_lock(&mutex); 
     while (queue.empty()) { 
      // unlock the mutex until the cond is signal()d or broadcast() to. 
      // if this call succeeds, we will have the mutex locked again on the other side. 
      pthread_cond_wait(&cond, &mutex); 
     } 
     // take the first task and then release the lock. 
     job = queue.front(); 
     queue.pop(); 
     pthread_mutex_unlock(&mutex); 

     if (job == NULL) { 
      // in this demonstration, NULL ends the run, so forward to any other threads. 
      AddJob(NULL); 
      break; 
     } 
     job->Execute(); 
     delete job; 
    } 
    return NULL; 
} 

int main(int argc, const char* argv[]) { 
    pthread_t worker1, worker2; 
    pthread_create(&worker1, NULL, &QueueWorker, NULL); 
    pthread_create(&worker2, NULL, &QueueWorker, NULL); 

    srand(time(NULL)); 

    // queue 5 jobs with delays. 
    for (size_t i = 0; i < 5; ++i) { 
     long delay = (rand() % 800) * 1000; 
     printf("Producer sleeping %fs\n", (float)delay/(1000*1000)); 
     usleep(delay); 
     Job* job = new Job(); 
     AddJob(job); 
    } 
    // 5 more without delays. 
    for (size_t i = 0; i < 5; ++i) { 
     AddJob(new Job); 
    } 
    // null to end the run. 
    AddJob(NULL); 

    printf("Done with jobs.\n"); 
    pthread_join(worker1, NULL); 
    pthread_join(worker2, NULL); 

    return 0; 
}