如何使用atomic_compare_exchange_weak解决生产者-消费者问题？

Question

我正在用C实现'生产者-消费者问题'，它解决了多线程时冗余生产、冗余消费、消费未生产项的同步问题。

bool expected = false; 
while (!atomic_compare_exchange_weak(&lock, &expected, true))
    expected = false;

为了防止进程重复访问，使用了atomic_compare_exchange_weak

我知道 atomic_compare_exchange_weak 可以在一个进程进入临界区时阻止另一个进程的访问。但是，进程的重复访问继续发生。我怎么解决这个问题？下面是我的生产者和消费者代码。

生产者代码

void *producer(void *arg) {
    int i = *(int *)arg;
    int item;
    
    while(alive) {

        while(counter >= BUFSIZE);
        
        expected_p = false;
        while (!atomic_compare_exchange_weak(&lock_p, &expected_p, true))
            expected_p = false;
        
        item = next_item++;
        
        buffer[in] = item;
        in = (in + 1) % BUFSIZE;
        counter++;  
            
        if (task_log[item][0] == -1) {
            task_log[item][0] = i;
            produced++;
            lock_p = false;
        }
        else {
            printf("ERROR\n");
            continue;
        } 
        printf("<P%d,%d>", i, item); 
    }
    pthread_exit(NULL); 
}

消费者代码

void *consumer(void *arg)
{
    int i = *(int *)arg;
    int item;

    while (alive) {

        while (counter <= 0);

        expected_c = false;
        while (!atomic_compare_exchange_weak(&lock_c, &expected_c, true)) 
            expected_c = false;
        
        item = buffer[out];
        out = (out + 1) % BUFSIZE;    
        counter--; 
        
        if (task_log[item][0] == -1) {
            printf("ERROR\n");
            continue;
        }
        else if (task_log[item][1] == -1) {
            task_log[item][1] = i;
            consumed++;
            lock_c = false; 
        }
        else {
            printf("ERROR\n");
            continue;
        }
        printf("<C%d,%d>\n", i, item);   
    }
    pthread_exit(NULL);
}

全球价值

int buffer[BUFSIZE];
int in = 0;
int out = 0;
int counter = 0;
int next_item = 0;

bool expected_p = false;
bool expected_c = false;
atomic_bool lock_p = false;
atomic_bool lock_c = false;

Answer 1

只需使用 pthread_cond 和 pthread_mutex 而不是尝试用原子做一些疯狂的事情。下面的代码使用真正的锁可能比使用原子原语有更多的开销，但它更安全且更易于维护。此外，条件变量代替自旋循环将显着提高两个线程的性能——克服使用真正锁的开销。

声明一个互斥量和一对条件变量：

pthread_mutex_t mut = PTHREAD_MUTEX_INITIALIZER;
pthread_cond_t cond_can_write, cand_can_read;

然后启动两个线程，修改如下：

void *producer(void *arg) {

    int i = *(int *)arg;
    int item;
    
    while(alive) {

        pthread_mutex_lock(&mut);  // TAKE THE LOCK

        // wait for counter to go below BUFSIZE
        while (counter >= BUFSIZE) {
            // atomically release lock and wait for consumer thread to signal
            pthread_cond_wait(&cond_can_write, &mut);

           // when pthread_cond_wait returns, this thread has reacquired the lock
        }

        // DO YOUR PRODUCE LOGIC HERE        

        pthread_mutex_unlock(&mut); // EXIT THE LOCK

        // signal to consumer that it can read
        pthread_cond_broadcast(&cond_can_read);
    }
    pthread_exit(NULL); 
}

void *consumer(void *arg)
{
    int i = *(int *)arg;
    int item;

    while (alive) {

        pthread_mutex_lock(&mut);   // TAKE THE LOCK

        // Wait for the counter to go above 0
        while (counter <= 0) {
            // atomically release lock and wait for producer thread to signal
            pthread_cond_wait(&cond_can_read, &mut);

            // when pthread_cond_wait returns, this thread has reacquired the lock
        }

        // DO YOUR CONSUME LOGIC HERE

        pthread_mutex_unlock(&mut);  // EXIT THE LOCK

        // signal to producer that it can write again
        pthread_cond_broadcast(&cond_write);

    }
    pthread_exit(NULL);
}