C中的非忙阻塞队列实现答案

【问题标题】：Non-busy blocking Queue Implementation in CC中的非忙阻塞队列实现
【发布时间】：2013-04-26 15:19:06
【问题描述】：

我正在尝试在 C 中实现一个队列，它会导致进程不忙地等待，直到队列中有一个元素可以使用。我尝试了两种不同的方法来实现这一目标。

我遇到的第一个问题是入队/出队操作是否有 if 条件来检查边界（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\n");
        }
    }       
    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\n");
        }
    }   
    strcpy(ptr,q->q[ q->first]);
    q->first = (q->first+1) % QUEUESIZE;
    q->count = q->count - 1;
}

【问题讨论】：

哪个操作系统？ Linux？（POSIX？）
@Amigable Clark Kant 实际上使用的是 IBM i (AS/400)，但它支持类似于 *nix 的 POSIX 接口。
我改了标签，现在可能会有更多人签到。
Srsly？你在 OS/400 下编程？伙计，那个平台是惊人的！正如您在问题中所回避的那样， condvar 和 mutex 将使这变得非常简单。 dequeue 中的 while 语句看起来有点奇怪。如果队列计数器为零，sen_wait 应该阻塞。如果有的话，你应该继续努力，也许在那里检查 retval。
有具体原因吗？理想情况下，信号量上的资源计数始终反映队列计数。除非队列本身也以初始占用者开始。

标签： c locking queue posix ibm-midrange

【解决方案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);

【讨论】：

mhhh 我认为q->count 可能存在竞争条件，如果您有超过 2 个进程（线程），q->first 和 q->last 可能存在竞争条件。

【解决方案2】：

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

诀窍是，您需要两个条件变量来覆盖完整和空等待状态。实现很简单，但很难涵盖极端情况，更难进行良好的测试。

我准备了一个示例，它是历史悠久的 Apache apr_queue 实现，但依赖项被剥离为仅 pthreads：https://github.com/chrismerck/rpa_queue

【讨论】：