c - 线程缓冲区 : how to prevent racing between clients and have client and server working on buffer simultaneously

Question

我编写了一个程序，它有几个线程客户端，可以将信息存储到循环缓冲区中；服务器可以从循环缓冲区读取(并“删除”)消息。我遇到的问题是当缓冲区已满时；当前正在写入缓冲区的客户端必须释放互斥体(服务器然后从缓冲区中删除所有数据)，但是当释放互斥体时，客户端竞争和已经忙碌的客户端不一定是第一个继续传输的；所以他的信息变得困惑。有人知道如何解决这个问题吗？或者您有更好的方法建议吗？最终我想制作一个多线程 tcp 服务器，将消息存储在缓冲区中，然后由服务器重新传输给所有客户端，这样你就知道我将用它做什么。另外，我是否可以不锁定缓冲区，以便服务器在需要时连续检索数据，或者这是不可能的？

#include <stdio.h>
#include <malloc.h>
#include <string.h>
#include <pthread.h>
#include <unistd.h>

struct cBuf{
    int     size;   /* maximum number of slots */
    int     start;  /* index of oldest element (last element) */
    int     end;    /* index at which to write new element (first element) */
    char    *buf;  /* buffer array */
};

struct cBuf cb;

pthread_mutex_t mutex = PTHREAD_MUTEX_INITIALIZER;

/* Initialize buffer*/
void buf_Init(struct cBuf *cb, int size) {
    cb->size  = size + 1; /* empty element */
    cb->start = 0;
    cb->end   = 0; 
    cb->buf = (char *)calloc(cb->size, sizeof(char)); //pointer to beginning of allocated memory
}
/* Clean up buffer*/
void buf_Free(struct cBuf *cb) {
    free(cb->buf);
}

/* Is the the buffer full? if it equals size meaning the buffer is full: end will point to zero (ex. 7%7 is 0) */ 
int buf_IsFull(struct cBuf *cb) {
    return (cb->end + 1) % cb->size == cb->start; 
}

/* Is the buffer empty? Yes if both start & end are zero (are equal) */
int cbIsEmpty(struct cBuf *cb) {
    return cb->end == cb->start; 
}

/* Write an element, overwriting oldest element if buffer is full. App can
   choose to avoid the overwrite by checking cbIsFull(). */
/* When inserting an element acces needs to be blocked (only 1 insert at a time =>mutexes 
 * When buffer is full it should return -1 so we can block the request*/
int buf_Insert(struct cBuf *cb, char *elem) {
    if (buf_IsFull(cb)){
        //buffer is full; cannot write into the buffer
        //sleep(2);
        return 0;
    } else {
        /* When writing to the buffer, the buffer has to be locked first (will block when already locked)*/
        pthread_mutex_lock(&mutex);
        cb->buf[cb->end] = *elem; //we put the new character in the last free slot
        /* we move the index for the next slot one further */
        cb->end = (cb->end + 1) % cb->size; 
        pthread_mutex_unlock(&mutex);
        return 1;
    }
}

/* Read oldest element. Ensure !cbIsEmpty() first. */
int buf_Read(struct cBuf *cb, char *out) {
    if (!cbIsEmpty(cb)){
        /* When reading from the buffer, the buffer has to be locked first (will block when already locked)*/
        pthread_mutex_lock(&mutex);
        *out = cb->buf[cb->start];
        cb->start = (cb->start + 1) % cb->size; //are news oldest character is now the next one
        pthread_mutex_unlock(&mutex);
        return 1;
    }
    //buffer is emtpy
    return 0;
}

void * client(void *cb){
    //sleep (2);
    #ifdef DEBUG
        printf("\nDEBUG (Thread ID:%lu) - Entering thread\t",(unsigned long)pthread_self());
    #endif

    /* main thread does not need to wait and this way all resources are released when terminating*/
    pthread_detach(pthread_self());

    struct cBuf *myData;
    myData = (struct cBuf*) cb;
    int i;

    /* client will post message in the buffer */
    char input[]="Hello World!";

    //printf("I:\t");
    for (i=0; i < sizeof(input); ++ i){ 
        if (buf_Insert(myData, &input[i]) ){
            //printf("[I]%c",input[i]);
        } else {
            //the buffer is full; wait until it has been emptied + go back a step because of lost character
            i--;    
            #ifdef DEBUG
            printf("\nDEBUG (Main Thread) - Full Buffer");  
            #endif
        }
    }
    return 0;
}

int main(void) {
    int bufferSize = 20; /* arbitrary size */
    char out;
    pthread_t thread;
    int j;

    buf_Init(&cb, bufferSize);

    //we make 1 client for starters
    for (j = 0; j<4; j++){
        if(pthread_create (&thread,NULL, client, (void *) &cb) !=0){
            #ifdef DEBUG
            printf("\nDEBUG (Main Thread) - Error while creating thread");
            #endif
        } else {
            #ifdef DEBUG
            printf("\nDEBUG (Main Thread) - Thread created");
            #endif
        }
    }

    while (1){
        /* Server will read (and thus remove) the message */
        if (cbIsEmpty(&cb)){
            //sleep(2);
            #ifdef DEBUG
            printf("\nDEBUG (Main Thread) - Buffer is empty");
            #endif
        } else {
            printf("\tO:\t");
            while (buf_Read(&cb,&out)){
                printf("%c", out);
            }
            #ifdef DEBUG        
            if (buf_Read(&cb, &out) == 0){
                printf("\nDEBUG (Main Thread) - Finished reading");
            }
            #endif
            printf("\n");   
        }
    }

    //empty the buffer; free the allocated memory
    buf_Free(&cb);
    return 0;
}

Answer 1

您所描述的是 pthread_cond_t 的典型用例，除了 pthread_mutex_t 之外，您还可以使用它来有选择地发出信号(和检查)线程。