c - 让我发疯的 Pthread 数据竞赛

Question

几周前我发布了一个与此类似的问题，我在使用 C 中的 pthreads 的 N-queens 程序中找不到数据竞争。Why is my multithreaded C program not working on macOS, but completely fine on Linux?

我在帖子的评论部分得到了一些建议，我真的尽力根据它们进行更正。我听了几天建议，更改了一些部分，但数据竞争仍然存在，我就是不明白为什么。关键部分内有用于生产和消费数量的计数器。在查看代码和分析代码时，我感到完全盲目，我知道消耗太多但围绕该代码片段的同步应该以我的知识是正确的，但显然有些地方不对。非常感谢外部输入。

这是我正在使用的代码，我不确定如何减小它的大小以重现问题。我使用 MacBook Pro 2020 x86_64 架构在 macOS Monterey (12.1) 上使用 gcc (clang-1205.0.22.11) 编译它。

编译:gcc -o 8q 8q.c*
运行:./8q <consumers> <N> , NxN 棋盘, N 个皇后放置
参数:./8q 2 4足以突出问题(应该产生 2 个解决方案，但每隔一次运行产生 3 个以上的解决方案，即存在重复的解决方案
注意:使用 ./8q 2 4 运行程序应该给出 2 个解决方案，1820 个生产和 1820 个消费。

#ifndef _REENTRANT
    #define _REENTRANT
#endif

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


typedef struct stack_buf {
    int positions[8];
    int top;
} stack_buf;

typedef struct global_buf {
    int positions[8];
    int buf_empty;
    long prod_done;
    int last_done;
} global_buf;

typedef struct print_buf {
    int qpositions[100][8];
    int top;
} print_buf;

stack_buf queen_comb = { {0}, 0 };
print_buf printouts = { {{0}}, 0 };
global_buf global = { {0}, 1, 0, 0 };
int N; //NxN board and N queens to place
long productions = 0;
long consumptions = 0;

pthread_mutex_t buffer_mutex, print_mutex;
pthread_cond_t empty, filled;


/* ##########################################################################################
   ################################## VALIDATION FUNCTIONS ##################################
   ########################################################################################## */

/* Validate that no queens are placed on the same row */
int valid_rows(int qpositions[]) {
    int rows[N];
    memset(rows, 0, N*sizeof(int));
    int row;
    for (int i = 0; i < N; i++) {
        row = qpositions[i] / N;
        if (rows[row] == 0) rows[row] = 1;
        else return 0;
    }
    return 1;
}

/* Validate that no queens are placed in the same column */
int valid_columns(int qpositions[]) {
    int columns[N];
    memset(columns, 0, N*sizeof(int));
    int column;
    for (int i = 0; i < N; i++) {
        column = qpositions[i] % N;
        if (columns[column] == 0) columns[column] = 1;
        else return 0;
    }
    return 1;
}

/* Validate that left and right diagonals aren't used by another queen */
int valid_diagonals(int qpositions[]) {
    int left_bottom_diagonals[N];
    int right_bottom_diagonals[N];
    int row, col, temp_col, temp_row, fill_value, index;

    for (int queen = 0; queen < N; queen++) {
        row = qpositions[queen] / N;
        col = qpositions[queen] % N;
        
        /* position --> left down diagonal endpoint (index) */
        fill_value = col < row ? col : row; // closest to bottom or left wall
        temp_row = row - fill_value;
        temp_col = col - fill_value;
        index = temp_row * N + temp_col; // board position
        for (int i = 0; i < queen; i++) { // check if interference occurs
            if (left_bottom_diagonals[i] == index) return 0;
        }
        left_bottom_diagonals[queen] = index; // no interference

        /* position --> right down diagonal endpoint (index) */
        fill_value = (N-1) - col < row ? N - col - 1 : row; // closest to bottom or right wall
        temp_row = row - fill_value;
        temp_col = col + fill_value;
        index = temp_row * N + temp_col; // board position
        for (int i = 0; i < queen; i++) { // check if interference occurs
            if (right_bottom_diagonals[i] == index) return 0;
        }
        right_bottom_diagonals[queen] = index; // no interference
    }
    return 1;
}

/* ##########################################################################################
   #################################### HELPER FUNCTION(S) ####################################
   ########################################################################################## */

/* print the collected solutions  */
void print(print_buf printouts) {
    static int solution_number = 1;
    int placement;

    pthread_mutex_lock(&print_mutex);
    for (int sol = 0; sol < printouts.top; sol++) { // all solutions
        printf("Solution %d: [ ", solution_number++);
        for (int pos = 0; pos < N; pos++) {
            printf("%d ", printouts.qpositions[sol][pos]+1);
        } 
        printf("]\n");

        printf("Placement:\n");
        for (int i = 1; i <= N; i++) { // rows
            printf("[ ");
            placement = printouts.qpositions[sol][N-i];
            for (int j = (N-i)*N; j < (N-i)*N+N; j++) { // physical position
                if (j == placement) {
                    printf(" Q ");
                } else printf("%2d ", j+1);
            }
            printf("]\n");
        }
        printf("\n");
    }
    pthread_mutex_unlock(&print_mutex);
}


/* ##########################################################################################
   #################################### THREAD FUNCTIONS ####################################
   ########################################################################################## */

/* entry point for each worker (consumer) workers will 
check each queen's row, column and diagonal to evaluate 
satisfactory placements */
void *eval_positioning(void *id) {
    long thr_id = (long)id;
    int qpositions[N]; // on stack (thread-private)
    
    while (1) {
        pthread_mutex_lock(&buffer_mutex);
        while (global.buf_empty == 1) { // no element
            if (global.last_done) { // last done, no combinations left 
                pthread_mutex_unlock(&buffer_mutex);
                pthread_cond_signal(&filled);
                return NULL;
            }
            if (global.prod_done) {
                global.last_done = 1;
                break;
            }
            pthread_cond_wait(&filled, &buffer_mutex);
        }

        memcpy(qpositions, global.positions, N*sizeof(int)); // copy to local scope
        global.buf_empty = 1;
        consumptions++;

        pthread_mutex_unlock(&buffer_mutex);
        pthread_cond_signal(&empty);
        
        if (valid_rows(qpositions) && valid_columns(qpositions) && valid_diagonals(qpositions)) {
            pthread_mutex_lock(&print_mutex);
            memcpy(printouts.qpositions[printouts.top], qpositions, N*sizeof(int));  /* save for printing later */
            printouts.top++;
            pthread_mutex_unlock(&print_mutex);
        }
    }
    return NULL;
}

/* recursively generate all possible queen_combs */
void rec_positions(int pos, int queens) {
    if (queens == 0) { // base case
        pthread_mutex_lock(&buffer_mutex);
        while (global.buf_empty == 0) { // while production hasn't been consumed
            pthread_cond_wait(&empty, &buffer_mutex);
        }
        memcpy(global.positions, queen_comb.positions, N*sizeof(int));
        productions++;
        global.buf_empty = 0;
        pthread_mutex_unlock(&buffer_mutex);
        pthread_cond_signal(&filled);
        return;
    }

    for (int i = pos; i <= N*N - queens; i++) {
        queen_comb.positions[queen_comb.top++] = i;
        rec_positions(i+1, queens-1);
        queen_comb.top--;
    }
}

/* binomial coefficient | without order, without replacement
8 queens on 8x8 board: 4'426'165'368 queen combinations */
void *generate_positions(void *arg) {
    rec_positions(0, N);
    pthread_mutex_lock(&buffer_mutex);
    global.prod_done = 1;
    pthread_mutex_unlock(&buffer_mutex);
    pthread_cond_broadcast(&filled); //wake all to 
    return NULL;
}

/* ##########################################################################################
   ########################################## MAIN ##########################################
   ########################################################################################## */

/* main procedure of the program */
int main(int argc, char *argv[]) {
    if (argc < 3) {
        printf("usage: ./8q <workers> <board width/height>\n");
        exit(1);
    }

    int workers = atoi(argv[1]);
    N = atoi(argv[2]);

    if (N < 2 || N > 8) {
        printf("Wrong input! 2 <= N <= 8\n");
        return 0;
    }

    struct timeval start, stop;
    double elapsed;
    pthread_t consumers[workers];
    pthread_t producer;

    printf("\n");
    
    gettimeofday(&start, NULL);

    pthread_create(&producer, NULL, generate_positions, NULL);
    for (long i = 0; i < workers; i++) {
        pthread_create(&consumers[i], NULL, eval_positioning, (void*)i+1);
    }

    pthread_join(producer, NULL);
    for (int i = 0; i < workers; i++) {
        pthread_join(consumers[i], NULL);
        char id[2];
        sprintf(id, "%d", i+1);
        write(1, id, strlen(id));
        write(1, " done\n\n", 6);
    }

    gettimeofday(&stop, NULL);
    elapsed = stop.tv_sec - start.tv_sec;
    elapsed += (stop.tv_usec - start.tv_usec) / (double)1000000;
    
    /* go through all valid solutions and print */
    print(printouts);
    
    printf("board: %dx%d, workers: %d (+1), exec time: %fs, solutions: %d\n", N, N, workers, elapsed, printouts.top);
    printf("productions:  %ld\nconsumptions: %ld\n", productions, consumptions);
    return 0;
}

Answer 1

您没有初始化互斥量和条件变量。在 pthread API 中使用时结果为 UB。有两种方法可以做到这一点，最简单的就是使用适当的初始化器:

pthread_mutex_t buffer_mutex = PTHREAD_MUTEX_INITIALIZER;
pthread_mutex_t print_mutex = PTHREAD_MUTEX_INITIALIZER;
pthread_cond_t empty = PTHREAD_COND_INITIALIZER;
pthread_cond_t filled = PTHREAD_COND_INITIALIZER;

无关，但值得一提的是，last_done 意识形态不是必需的。这可以仅使用 buf_empty 和 prod_done 状态来完成。具体来说:

void *eval_positioning(void *tid)
{
    int qpositions[N]; // on stack (thread-private)

    while (1)
    {
        pthread_mutex_lock(&buffer_mutex);

        // while still producing *and* the buffer is empty
        while (!global.prod_done && global.buf_empty)
            pthread_cond_wait(&filled, &buffer_mutex);

        // if both are true, we're done. nothing to process, and
        //  there never will be (e.g. prod_done)
        if (global.prod_done && global.buf_empty)
        {
            // signal anyone else waiting on that mutex+cvar
            pthread_cond_signal(&filled);
            break;
        }

        // if we have a buffer to process (even if prod_done is true)
        else if (!global.buf_empty)
        {
            // make local copy of buffer
            memcpy(qpositions, global.positions, sizeof qpositions);
            ++consumptions;

            // mark global buffer as ready-to-receive
            global.buf_empty = 1;
            pthread_cond_signal(&empty);
            pthread_mutex_unlock(&buffer_mutex);

            // if validated...
            if (valid_rows(qpositions) && valid_columns(qpositions) && valid_diagonals(qpositions))
            {
                // record and bump the printout counter.
                pthread_mutex_lock(&print_mutex);
                int row = printouts.top++;
                pthread_mutex_unlock(&print_mutex);

                // this need not be protected by the mutex. we "own"
                //  this now, and can just blast away.
                memcpy(printouts.qpositions[row], qpositions, sizeof qpositions);
            }
        }
        else
        {
            pthread_mutex_unlock(&buffer_mutex);
        }
    }

    // make sure we unlock this
    pthread_mutex_unlock(&buffer_mutex);

    return tid;
}

正确初始化并发 Material 和上述 eval 处理器后，这是一致的输出:

输出

1 done
2 done
Solution 1: [ 2 8 9 15 ]
Placement:
[ 13 14  Q 16 ]
[  Q 10 11 12 ]
[  5  6  7  Q ]
[  1  Q  3  4 ]

Solution 2: [ 3 5 12 14 ]
Placement:
[ 13  Q 15 16 ]
[  9 10 11  Q ]
[  Q  6  7  8 ]
[  1  2  Q  4 ]

board: 4x4, workers: 2 (+1), exec time: 0.013001s, solutions: 2
productions:  1820
consumptions: 1820

_{对微不足道的笔记本电脑性能数字表示歉意}