如何在沒有連接的情況下同步管理器/工作者pthreads？

Question

我熟悉多線程，並且我已經成功地在Java和Objective-C中開發了許多多線程程序。但我不能達到使用並行線程，而無需使用來自主線程加入C中的以下內容：

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

#define NUM_OF_THREADS 2 

struct thread_data { 
    int start; 
    int end; 
    int *arr; 
}; 

void print(int *ints, int n); 
void *processArray(void *args); 

int main(int argc, const char * argv[]) 
{ 
    int numOfInts = 10; 
    int *ints = malloc(numOfInts * sizeof(int)); 
    for (int i = 0; i < numOfInts; i++) { 
     ints[i] = i; 
    } 
    print(ints, numOfInts); // prints [0, 1, 2, 3, 4, 5, 6, 7, 8, 9] 

    pthread_t threads[NUM_OF_THREADS]; 
    struct thread_data thread_data[NUM_OF_THREADS]; 

    // these vars are used to calculate the index ranges for each thread 
    int remainingWork = numOfInts, amountOfWork; 
    int startRange, endRange = -1; 

    for (int i = 0; i < NUM_OF_THREADS; i++) { 

     amountOfWork = remainingWork/(NUM_OF_THREADS - i); 
     startRange = endRange + 1; 
     endRange = startRange + amountOfWork - 1; 

     thread_data[i].arr = ints; 
     thread_data[i].start = startRange; 
     thread_data[i].end = endRange; 

     pthread_create(&threads[i], NULL, processArray, (void *)&thread_data[i]); 

     remainingWork -= amountOfWork;  
    } 

    // 1. Signal to the threads to start working 


    // 2. Wait for them to finish 


    print(ints, numOfInts); // should print [1, 2, 3, 4, 5, 6, 7, 8, 9, 10] 

    free(ints); 
    return 0; 
} 

void *processArray(void *args) 
{ 
    struct thread_data *data = (struct thread_data *)args; 
    int *arr = data->arr; 
    int start = data->start; 
    int end = data->end; 

    // 1. Wait for a signal to start from the main thread 


    for (int i = start; i <= end; i++) { 
     arr[i] = arr[i] + 1; 
    } 

    // 2. Signal to the main thread that you're done 

    pthread_exit(NULL); 
} 

void print(int *ints, int n) 
{ 
    printf("["); 
    for (int i = 0; i < n; i++) { 
     printf("%d", ints[i]); 
     if (i+1 != n) 
      printf(", "); 
    } 
    printf("]\n"); 
} 

我想實現在上面的代碼如下：

在main（）：

1. 發信號給線程開始工作。
2. 等待後臺線程完成。

在processArray（）：

1. 等待一個信號，從主線程
2. 信號，即大功告成主線程開始

我不想要在主線程中使用連接，因爲在the real application中，主線程將創建一次線程，然後它將向後臺線程發信號通知多次，並且我不能讓主線程繼續，除非所有背景THR eads已完成工作。在processArray功能，我會把一個無限循環如下：

void *processArray(void *args) 
{ 
    struct thread_data *data = (struct thread_data *)args; 

    while (1) 
    { 
     // 1. Wait for a signal to start from the main thread 

     int *arr = data->arr; 
     int start = data->start; 
     int end = data->end;   

     // Process 
     for (int i = start; i <= end; i++) { 
      arr[i] = arr[i] + 1; 
     } 

     // 2. Signal to the main thread that you're done 

    } 

    pthread_exit(NULL); 
} 

請注意，我是新來的C和POSIX API，所以原諒我，如果我失去了一些東西明顯。但是我真的嘗試了很多東西，從使用互斥量和一系列信號量，以及兩者的混合，但都沒有成功。我認爲一個條件變量可能會有所幫助，但我無法理解它可以如何使用。

謝謝你的時間。

問題就迎刃而解了：

謝謝你們這麼多！我終於能夠安全地工作，不需要按照提示使用連接。雖然解決方案有點難看，但它可以完成工作，性能提升也是值得的（如下所示）。任何有興趣，這是我工作的實際應用，在主線程一直給連續工作到後臺線程的模擬：

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

#define NUM_OF_THREADS 5 

struct thread_data { 
    int id; 
    int start; 
    int end; 
    int *arr; 
}; 

pthread_mutex_t currentlyIdleMutex = PTHREAD_MUTEX_INITIALIZER; 
pthread_cond_t currentlyIdleCond = PTHREAD_COND_INITIALIZER; 
int currentlyIdle; 

pthread_mutex_t workReadyMutex = PTHREAD_MUTEX_INITIALIZER; 
pthread_cond_t workReadyCond = PTHREAD_COND_INITIALIZER; 
int workReady; 

pthread_cond_t currentlyWorkingCond = PTHREAD_COND_INITIALIZER; 
pthread_mutex_t currentlyWorkingMutex= PTHREAD_MUTEX_INITIALIZER; 
int currentlyWorking; 

pthread_mutex_t canFinishMutex = PTHREAD_MUTEX_INITIALIZER; 
pthread_cond_t canFinishCond = PTHREAD_COND_INITIALIZER; 
int canFinish; 

void print(int *ints, int n); 
void *processArray(void *args); 
int validateResult(int *ints, int num, int start); 

int main(int argc, const char * argv[]) 
{ 
    int numOfInts = 10; 
    int *ints = malloc(numOfInts * sizeof(int)); 
    for (int i = 0; i < numOfInts; i++) { 
     ints[i] = i; 
    } 
// print(ints, numOfInts); 

    pthread_t threads[NUM_OF_THREADS]; 
    struct thread_data thread_data[NUM_OF_THREADS]; 
    workReady = 0; 
    canFinish = 0; 
    currentlyIdle = 0; 
    currentlyWorking = 0; 

    // these vars are used to calculate the index ranges for each thread 
    int remainingWork = numOfInts, amountOfWork; 
    int startRange, endRange = -1; 
    // Create the threads and give each one its data struct. 
    for (int i = 0; i < NUM_OF_THREADS; i++) { 

     amountOfWork = remainingWork/(NUM_OF_THREADS - i); 
     startRange = endRange + 1; 
     endRange = startRange + amountOfWork - 1; 

     thread_data[i].id = i; 
     thread_data[i].arr = ints; 
     thread_data[i].start = startRange; 
     thread_data[i].end = endRange; 

     pthread_create(&threads[i], NULL, processArray, (void *)&thread_data[i]); 
     remainingWork -= amountOfWork; 
    } 

    int loops = 1111111; 
    int expectedStartingValue = ints[0] + loops; // used to validate the results 
    // The elements in ints[] should be incremented by 1 in each loop 
    while (loops-- != 0) { 

     // Make sure all of them are ready 
     pthread_mutex_lock(&currentlyIdleMutex); 
     while (currentlyIdle != NUM_OF_THREADS) { 
      pthread_cond_wait(&currentlyIdleCond, &currentlyIdleMutex); 
     } 
     pthread_mutex_unlock(&currentlyIdleMutex); 

     // All threads are now blocked; it's safe to not lock the mutex. 
     // Prevent them from finishing before authorized. 
     canFinish = 0; 
     // Reset the number of currentlyWorking threads 
     currentlyWorking = NUM_OF_THREADS; 

     // Signal to the threads to start 
     pthread_mutex_lock(&workReadyMutex); 
     workReady = 1; 
     pthread_cond_broadcast(&workReadyCond); 
     pthread_mutex_unlock(&workReadyMutex);  

     // Wait for them to finish 
     pthread_mutex_lock(&currentlyWorkingMutex); 
     while (currentlyWorking != 0) { 
      pthread_cond_wait(&currentlyWorkingCond, &currentlyWorkingMutex); 
     } 
     pthread_mutex_unlock(&currentlyWorkingMutex); 

     // The threads are now waiting for permission to finish 
     // Prevent them from starting again 
     workReady = 0; 
     currentlyIdle = 0; 

     // Allow them to finish 
     pthread_mutex_lock(&canFinishMutex); 
     canFinish = 1; 
     pthread_cond_broadcast(&canFinishCond); 
     pthread_mutex_unlock(&canFinishMutex); 
    } 

// print(ints, numOfInts); 

    if (validateResult(ints, numOfInts, expectedStartingValue)) { 
     printf("Result correct.\n"); 
    } 
    else { 
     printf("Result invalid.\n");  
    } 

    // clean up 
    for (int i = 0; i < NUM_OF_THREADS; i++) { 
     pthread_cancel(threads[i]); 
    } 
    free(ints); 

    return 0; 
} 

void *processArray(void *args) 
{ 
    struct thread_data *data = (struct thread_data *)args; 
    int *arr = data->arr; 
    int start = data->start; 
    int end = data->end; 

    while (1) { 

     // Set yourself as idle and signal to the main thread, when all threads are idle main will start 
     pthread_mutex_lock(&currentlyIdleMutex); 
     currentlyIdle++; 
     pthread_cond_signal(&currentlyIdleCond); 
     pthread_mutex_unlock(&currentlyIdleMutex); 

     // wait for work from main 
     pthread_mutex_lock(&workReadyMutex); 
     while (!workReady) { 
      pthread_cond_wait(&workReadyCond , &workReadyMutex); 
     } 
     pthread_mutex_unlock(&workReadyMutex); 

     // Do the work 
     for (int i = start; i <= end; i++) { 
      arr[i] = arr[i] + 1; 
     } 

     // mark yourself as finished and signal to main 
     pthread_mutex_lock(&currentlyWorkingMutex); 
     currentlyWorking--; 
     pthread_cond_signal(&currentlyWorkingCond); 
     pthread_mutex_unlock(&currentlyWorkingMutex); 

     // Wait for permission to finish 
     pthread_mutex_lock(&canFinishMutex); 
     while (!canFinish) { 
      pthread_cond_wait(&canFinishCond , &canFinishMutex); 
     } 
     pthread_mutex_unlock(&canFinishMutex); 
    } 

    pthread_exit(NULL); 
} 

int validateResult(int *ints, int n, int start) 
{ 
    int tmp = start; 
    for (int i = 0; i < n; i++, tmp++) { 
     if (ints[i] != tmp) { 
      return 0; 
     } 
    } 
    return 1; 
} 

void print(int *ints, int n) 
{ 
    printf("["); 
    for (int i = 0; i < n; i++) { 
     printf("%d", ints[i]); 
     if (i+1 != n) 
      printf(", "); 
    } 
    printf("]\n"); 
} 

我不知道但如果pthread_cancel就足夠了清理！至於屏障，如果它不限於@Jeremy所提到的某些操作系統，它會有很大的幫助。

基準：

我希望確保這些條件很多實際上並沒有放緩的算法，所以我設置這個基準測試的兩種解決方案比較：

#include <stdio.h> 
#include <stdlib.h> 
#include <pthread.h> 
#include <unistd.h> 
#include <sys/time.h> 
#include <sys/resource.h> 

#define NUM_OF_THREADS 5 
struct thread_data { 
    int start; 
    int end; 
    int *arr; 
}; 
pthread_mutex_t currentlyIdleMutex = PTHREAD_MUTEX_INITIALIZER; 
pthread_cond_t currentlyIdleCond = PTHREAD_COND_INITIALIZER; 
int currentlyIdle; 
pthread_mutex_t workReadyMutex = PTHREAD_MUTEX_INITIALIZER; 
pthread_cond_t workReadyCond = PTHREAD_COND_INITIALIZER; 
int workReady; 
pthread_cond_t currentlyWorkingCond = PTHREAD_COND_INITIALIZER; 
pthread_mutex_t currentlyWorkingMutex= PTHREAD_MUTEX_INITIALIZER; 
int currentlyWorking; 
pthread_mutex_t canFinishMutex = PTHREAD_MUTEX_INITIALIZER; 
pthread_cond_t canFinishCond = PTHREAD_COND_INITIALIZER; 
int canFinish; 

void *processArrayMutex(void *args); 
void *processArrayJoin(void *args); 
double doItWithMutex(pthread_t *threads, struct thread_data *data, int loops); 
double doItWithJoin(pthread_t *threads, struct thread_data *data, int loops); 

int main(int argc, const char * argv[]) 
{ 
    int numOfInts = 10; 
    int *join_ints = malloc(numOfInts * sizeof(int)); 
    int *mutex_ints = malloc(numOfInts * sizeof(int)); 
    for (int i = 0; i < numOfInts; i++) { 
     join_ints[i] = i; 
     mutex_ints[i] = i; 
    } 

    pthread_t join_threads[NUM_OF_THREADS]; 
    pthread_t mutex_threads[NUM_OF_THREADS]; 
    struct thread_data join_thread_data[NUM_OF_THREADS]; 
    struct thread_data mutex_thread_data[NUM_OF_THREADS]; 
    workReady = 0; 
    canFinish = 0; 
    currentlyIdle = 0; 
    currentlyWorking = 0; 

    int remainingWork = numOfInts, amountOfWork; 
    int startRange, endRange = -1; 
    for (int i = 0; i < NUM_OF_THREADS; i++) { 
     amountOfWork = remainingWork/(NUM_OF_THREADS - i); 
     startRange = endRange + 1; 
     endRange = startRange + amountOfWork - 1; 

     join_thread_data[i].arr = join_ints; 
     join_thread_data[i].start = startRange; 
     join_thread_data[i].end = endRange; 
     mutex_thread_data[i].arr = mutex_ints; 
     mutex_thread_data[i].start = startRange; 
     mutex_thread_data[i].end = endRange; 

     pthread_create(&mutex_threads[i], NULL, processArrayMutex, (void *)&mutex_thread_data[i]); 
     remainingWork -= amountOfWork; 
    } 

    int numOfBenchmarkTests = 100; 
    int numberOfLoopsPerTest= 1000; 

    double join_sum = 0.0, mutex_sum = 0.0; 
    for (int i = 0; i < numOfBenchmarkTests; i++) 
    { 
     double joinTime = doItWithJoin(join_threads, join_thread_data, numberOfLoopsPerTest); 
     double mutexTime= doItWithMutex(mutex_threads, mutex_thread_data, numberOfLoopsPerTest); 

     join_sum += joinTime; 
     mutex_sum+= mutexTime;  
    } 

    double join_avg = join_sum/numOfBenchmarkTests; 
    double mutex_avg= mutex_sum/numOfBenchmarkTests; 

    printf("Join average : %f\n", join_avg); 
    printf("Mutex average: %f\n", mutex_avg); 

    double diff = join_avg - mutex_avg; 
    if (diff > 0.0) 
     printf("Mutex is %.0f%% faster.\n", 100 * diff/join_avg); 
    else if (diff < 0.0) 
     printf("Join is %.0f%% faster.\n", 100 * diff/mutex_avg); 
    else 
     printf("Both have the same performance."); 

    free(join_ints); 
    free(mutex_ints); 

    return 0; 
} 

// From https://stackoverflow.com/a/2349941/408286 
double get_time() 
{ 
    struct timeval t; 
    struct timezone tzp; 
    gettimeofday(&t, &tzp); 
    return t.tv_sec + t.tv_usec*1e-6; 
} 

double doItWithMutex(pthread_t *threads, struct thread_data *data, int num_loops) 
{ 
    double start = get_time(); 

    int loops = num_loops; 
    while (loops-- != 0) { 
     // Make sure all of them are ready 
     pthread_mutex_lock(&currentlyIdleMutex); 
     while (currentlyIdle != NUM_OF_THREADS) { 
      pthread_cond_wait(&currentlyIdleCond, &currentlyIdleMutex); 
     } 
     pthread_mutex_unlock(&currentlyIdleMutex); 

     // All threads are now blocked; it's safe to not lock the mutex. 
     // Prevent them from finishing before authorized. 
     canFinish = 0; 
     // Reset the number of currentlyWorking threads 
     currentlyWorking = NUM_OF_THREADS; 

     // Signal to the threads to start 
     pthread_mutex_lock(&workReadyMutex); 
     workReady = 1; 
     pthread_cond_broadcast(&workReadyCond); 
     pthread_mutex_unlock(&workReadyMutex); 

     // Wait for them to finish 
     pthread_mutex_lock(&currentlyWorkingMutex); 
     while (currentlyWorking != 0) { 
      pthread_cond_wait(&currentlyWorkingCond, &currentlyWorkingMutex); 
     } 
     pthread_mutex_unlock(&currentlyWorkingMutex); 

     // The threads are now waiting for permission to finish 
     // Prevent them from starting again 
     workReady = 0; 
     currentlyIdle = 0; 

     // Allow them to finish 
     pthread_mutex_lock(&canFinishMutex); 
     canFinish = 1; 
     pthread_cond_broadcast(&canFinishCond); 
     pthread_mutex_unlock(&canFinishMutex); 
    } 

    return get_time() - start; 
} 

double doItWithJoin(pthread_t *threads, struct thread_data *data, int num_loops) 
{ 
    double start = get_time(); 

    int loops = num_loops; 
    while (loops-- != 0) { 
     // create them 
     for (int i = 0; i < NUM_OF_THREADS; i++) { 
      pthread_create(&threads[i], NULL, processArrayJoin, (void *)&data[i]); 
     } 
     // wait 
     for (int i = 0; i < NUM_OF_THREADS; i++) { 
      pthread_join(threads[i], NULL); 
     } 
    } 

    return get_time() - start; 
} 

void *processArrayMutex(void *args) 
{ 
    struct thread_data *data = (struct thread_data *)args; 
    int *arr = data->arr; 
    int start = data->start; 
    int end = data->end; 

    while (1) { 

     // Set yourself as idle and signal to the main thread, when all threads are idle main will start 
     pthread_mutex_lock(&currentlyIdleMutex); 
     currentlyIdle++; 
     pthread_cond_signal(&currentlyIdleCond); 
     pthread_mutex_unlock(&currentlyIdleMutex); 

     // wait for work from main 
     pthread_mutex_lock(&workReadyMutex); 
     while (!workReady) { 
      pthread_cond_wait(&workReadyCond , &workReadyMutex); 
     } 
     pthread_mutex_unlock(&workReadyMutex); 

     // Do the work 
     for (int i = start; i <= end; i++) { 
      arr[i] = arr[i] + 1; 
     } 

     // mark yourself as finished and signal to main 
     pthread_mutex_lock(&currentlyWorkingMutex); 
     currentlyWorking--; 
     pthread_cond_signal(&currentlyWorkingCond); 
     pthread_mutex_unlock(&currentlyWorkingMutex); 

     // Wait for permission to finish 
     pthread_mutex_lock(&canFinishMutex); 
     while (!canFinish) { 
      pthread_cond_wait(&canFinishCond , &canFinishMutex); 
     } 
     pthread_mutex_unlock(&canFinishMutex); 
    } 

    pthread_exit(NULL); 
} 

void *processArrayJoin(void *args) 
{ 
    struct thread_data *data = (struct thread_data *)args; 
    int *arr = data->arr; 
    int start = data->start; 
    int end = data->end; 

    // Do the work 
    for (int i = start; i <= end; i++) { 
     arr[i] = arr[i] + 1; 
    } 

    pthread_exit(NULL); 
} 

輸出是：

Join average : 0.153074 
Mutex average: 0.071588 
Mutex is 53% faster. 

再次感謝您。我非常感謝你的幫助！

Answer 1

您需要使用與join不同的同步技術，這很清楚。

不幸的是，您有很多選擇。一個是「同步障礙」，基本上是每個到達它的線程都會阻塞，直到它們全部到達它（您預先指定了線程的數量）。看看pthread_barrier。

另一種是使用條件變量/互斥對（pthread_cond_*）。當每個線程完成時，它需要互斥量，遞增計數，表示condvar。主線程在condvar上等待，直到計數達到預期的值。代碼如下所示：

// thread has finished 
mutex_lock 
++global_count 
// optional optimization: only execute the next line when global_count >= N 
cond_signal 
mutex_unlock 

// main is waiting for N threads to finish 
mutex_lock 
while (global_count < N) { 
    cond_wait 
} 
mutex_unlock 

另一種是用每個線程一個信號 - 當線程完成它張貼自己的旗語，並依次對每個信號燈的主線程等待而不是加入依次在每個線程。

您還需要同步才能重新啓動下一個作業的線程 - 這可能是與第一個同類型的第二個同步對象，因爲您有1個海報和N個服務員比其他方式。或者你可以（小心）重複使用同一個對象。

如果你已經嘗試過這些東西，並且你的代碼不起作用，那麼可能會問你關於你試過的代碼的一個新的具體問題。所有這些都足以完成這項任務。

Answer 2

要告訴所有線程開始工作，它可以像初始化爲零的全局整數變量一樣簡單，並且線程只是等到它不爲零。這樣你就不需要線程函數中的while (1)循環。

對於等待他們全部完成，pthread_join是最簡單的，因爲它實際上會阻塞，直到它加入的線程完成。在線程之後還需要清理系統內容（否則，線程的返回值將被存儲用於程序的其餘部分）。由於線程的所有pthread_t都是數組，所以只需循環一次即可。由於程序的這一部分沒有做其他任何事情，所以必須等到所有線程都完成後，才按順序等待它們。

Answer 3

有幾種可以使用的同步機制（例如條件變量）。我認爲最簡單的方法是使用pthread_barrier來同步線程的開始。

假設您希望所有線程在每次循環迭代中「同步」，您可以重新使用屏障。如果你需要更靈活的東西，一個條件變量可能更合適。

當你決定是時候讓線程結束了（你還沒有表明線程知道如何擺脫無限循環 - 一個簡單的共享變量可能用於這個;共享變量可能是一個原子類型或受互斥體保護），main()線程應使用pthread_join()等待所有線程完成。

Answer 4

您正在處理錯誤的抽象層次。這個問題已經解決了。您正在重新實現工作隊列+線程池。

OpenMP看起來很適合您的問題。它將#pragma註釋轉換爲線程代碼。我相信這會讓你直接表達你想做的事情。

使用libdispatch，您正在嘗試執行的操作將被表示爲針對併發隊列的dispatch_apply。這隱含地等待所有的子任務完成。在OS X下，它使用不可移植的pthread工作隊列接口實現;在FreeBSD下，我相信它直接管理一組pthread。

如果是便攜性問題促使您使用原始pthreads，請勿使用pthread barrier。障礙是基本POSIX線程之外的另一個擴展。例如OS X不支持它。欲瞭解更多信息，請參閱POSIX。

阻塞主線程直到所有子線程完成都可以使用受條件變量保護的計數來完成，或者更簡單地說，使用管道和阻塞讀取，其中要讀取的字節數與線程數匹配。每個線程在工作完成時寫入一個字節，然後休眠，直到它從主線程獲得新的工作。一旦每個線程寫完「我完成了！」，主線程就會解鎖。字節。

將工作傳遞給子線程可以使用保護工作描述符的互斥鎖和表示新工作的條件來完成。您可以使用所有線程從中繪製的單個工作描述符數組。在信號上，每個人都試圖抓住互斥體。在抓取互斥體時，它會將某些工作出隊，如果隊列不是空的，則重新發出信號，然後處理其工作，之後它會向主線程發出完成信號。

您可以重複使用這個「工作隊列」，通過排列結果來解鎖主線程，主線程一直等待，直到結果隊列長度與線程數相匹配;管道方法只是使用阻塞read爲你做這個計數。