C中的非繁忙阻塞队列实现

Posted 2021-04-08

我试图在C中实现一个队列，导致进程非忙等待，直到队列中有一个元素要消耗。为了达到这个目的，我尝试了两种不同的方法。

我遇到的第一个问题是，如果enqueue / dequeue操作有条件来检查边界（if（q-> count == QUEUESIZE）），对sem_wait的调用将立即返回，因为没有其他进程获得锁定。

如果我将条件更改为while（q-> count == QUEUESIZE），我相信消费者进程将“忙等待”，直到生产者进程发布信号量，这不是我实现的目标，并通过测试，我发现消费者进程不会获得锁定并继续。

我认为我很接近，但我似乎无法弄清楚如何解决这些问题。我考虑过添加条件变量或pthread_mutex，但是在添加额外的复杂性之前想要耗尽信号量选项。

#define QUEUESIZE 48

typedef struct 
{           
    char q[QUEUESIZE+1][150];
    int first;                      
    int last;                       
    int count;                      
    sem_t *lock;                    
} Queue;


init_queue(Queue *q, sem_t *l)
{
    q->first = 0;
    q->last = QUEUESIZE-1;
    q->count = 0;
    q->lock = l;
}

enqueue(Queue *q, char x[150])
{
    while(q->count == QUEUESIZE)
        sem_wait(q->lock);

    if (q->count == 0)
    {
        if (sem_post(q->lock) == -1)
        {
            printf("Thread failed to unlock semaphore
");
        }
    }       
    q->last = (q->last+1) % QUEUESIZE;
    strcpy(q->q[ q->last ],x);    
    q->count = q->count + 1;
}

dequeue(Queue *q,char *ptr)
{
    char x[150];
    while(q->count == 0)
        sem_wait(q->lock);

    if (q->count == QUEUESIZE) 
    {
        if (sem_post(q->lock) == -1)
        {
            printf("Thread failed to unlock semaphore
");
        }
    }   
    strcpy(ptr,q->q[ q->first]);
    q->first = (q->first+1) % QUEUESIZE;
    q->count = q->count - 1;
}

答案

根据要求，这是我的解决方案。

#define QUEUESIZE 50

typedef struct 
{           
    char q[QUEUESIZE][150];
    int first;                      
    int last;                       
    int count;                      
    sem_t *full;
    sem_t *empty;
    sem_t *excl;

} Queue;


void init_queue(Queue *q, sem_t *f,sem_t *e, sem_t *ee,)
{
    q->first = 0;
    q->last = QUEUESIZE-1;
    q->count = 0;
    q->full = f;
    q->empty = e;
    q->excl = ee; 
}

void enqueue(Queue *q, char x[150])
{
    sem_wait(q->empty);
    sem_wait(q->excl);

    q->last = (q->last+1) % QUEUESIZE;
    strcpy(q->q[ q->last ],x);    
    q->count = q->count + 1;

    sem_post(q->excl);
    sem_post(q->full);
}

void dequeue(Queue *q,char *ptr)
{
    sem_wait(q->full);
    sem_wait(q->excl);

    strcpy(ptr,q->q[ q->first]);
    q->first = (q->first+1) % QUEUESIZE;
    q->count = q->count - 1;

    sem_post(q->excl);
    sem_post(q->empty);
}

我初始化信号量如下：

sem_init(full,1,0);
sem_init(empty,1,49);
sem_init(dequeue_excl,1,1);
sem_init(enqueue_excl,1,1);

另一答案

正如您在使用信号量的示例中所认识到的那样，您需要操作系统的一些支持才能实现此目的。如果您使用的是支持POSIX消息队列的操作系统，则可以依赖它。否则，您可以使用pthread条件变量作为实现的基础。

诀窍是，您需要两个条件变量来覆盖完整和空等待状态。实施很简单，但很难覆盖极端情况，甚至更难以测试。

我已经准备了一个示例，这是一个历史悠久的Apache apr_queue实现，但依赖关系被剥离到只有pthreads：https://github.com/chrismerck/rpa_queue