Linux平台上C语言实现异步队列的两种方法

Linux上目前有两种事件通知方式&＃xff0c;一种是线程条件变量&＃xff0c;一种是利用eventfd实现事件通知&＃xff0c;下面介绍一下利用这两种方法实现异步队列的方法。 ###线程条件变量 ####相关函数介绍

• pthread_cond_init:初始化一个线程条件变量。
• pthread_cond_wait:等待条件触发。
• pthread_cond_signal:通知一个线程&＃xff0c;线程条件发生。
• pthread_cond_timedwait:等待条件触发&＃xff0c;可以设置超时时间。
• pthread_cond_reltimedwait_np:和pthread_cond_timedwait使用基本相同&＃xff0c;区别是使用的是相对时间间隔而不是绝对时间间隔。
• pthread_cond_broadcast:通知所有等待线程&＃xff0c;线程条件发生。
• pthread_cond_destroy:销毁条件变量。

####唤醒丢失问题 如果线程未持有与条件相关联的互斥锁&＃xff0c;则调用 pthread_cond_signal() 或 pthread_cond_broadcast() 会产生唤醒丢失错误。满足以下所有条件时&＃xff0c;即会出现唤醒丢失问题&＃xff1a;

• 一个线程调用 pthread_cond_signal() 或 pthread_cond_broadcast()
• 另一个线程已经测试了该条件&＃xff0c;但是尚未调用 pthread_cond_wait()
• 没有正在等待的线程

信号不起作用&＃xff0c;因此将会丢失&＃xff0c;仅当修改所测试的条件但未持有与之相关联的互斥锁时&＃xff0c;才会出现此问题。只要仅在持有关联的互斥锁同时修改所测试的条件&＃xff0c;即可调用 pthread_cond_signal() 和 pthread_cond_broadcast()&＃xff0c;而无论这些函数是否持有关联的互斥锁。

####线程条件变量使用方法

get_resources(int amount)
{ pthread_mutex_lock(&rsrc_lock); while (resources }add_resources(int amount)
{pthread_mutex_lock(&rsrc_lock);resources &＃43;&＃61; amount;pthread_cond_broadcast(&rsrc_add);pthread_mutex_unlock(&rsrc_lock);}


###eventfd

int eventfd(unsigned int initval, int flags);

eventfd是Linux提供内核态的事件等待/通知机制&＃xff0c;内核维护了一个8字节的整型数&＃xff0c;该整型数由initval来初始化&＃xff0c;flags参数可以由以下值位或而来&＃xff1a;

• EFD_CLOEXEC:设置该描述符的O_CLOEXEC标志。
• EFD_NONBLOCK:设置描述符为非阻塞模式。
• EFD_SEMAPHORE:设置描述符为信号量工作模式&＃xff0c;在此模式下&＃xff0c;read模式会使整型数减1并返回数值1。

当内核维护的8字节整型数为0时&＃xff0c;read操作会阻塞&＃xff0c;如果为fd设置为非阻塞模式&＃xff0c;则返回EAGAIN错误。

###简单的唤醒队列

下面我们实现一个简单的环形队列&＃xff1a;

#define default_size 1024typedef struct queue
{int header;int tail;int size;int capcity;void **_buf;
} queue_t;queue_t *queue_create(int size)
{queue_t *q &＃61; malloc(sizeof (queue_t));if (q !&＃61; NULL){if (size > 0){q->_buf &＃61; malloc(size);q->capcity &＃61; size;}else{q->_buf &＃61; malloc(default_size * sizeof (void *));q->capcity &＃61; default_size;}q->header &＃61; q->tail &＃61; q->size &＃61; 0;}return q;
}int queue_is_full(queue_t *q)
{return q->size &＃61;&＃61; q->capcity;
}int queue_is_empty(queue_t *q)
{return q->size &＃61;&＃61; 0;
}void queue_push_tail(queue_t *q, void *data)
{if (!queue_is_full(q)){q->_buf[q->tail] &＃61; data;q->tail &＃61; (q->tail &＃43; 1) % q->capcity;q->size&＃43;&＃43;;}
}void *queue_pop_head(queue_t *q)
{void *data &＃61; NULL;if (!queue_is_empty(q)){data &＃61; q->_buf[(q->header)];q->header &＃61; (q->header &＃43; 1) % q->capcity;q->size--;}return data;
}int *queue_free(queue_t *q)
{free(q->_buf);free(q);
}


###线程变量实现的异步队列

typedef struct async_queue
{pthread_mutex_t mutex;pthread_cond_t cond;int waiting_threads;queue_t *_queue;
} async_queue_t;
async_queue_t *async_queue_create(int size)
{async_queue_t *q &＃61; malloc(sizeof (async_queue_t));q->_queue &＃61; queue_create(size);q->waiting_threads &＃61; 0;pthread_mutex_init(&(q->mutex), NULL);pthread_cond_init(&(q->cond), NULL);return q;
}void async_queue_push_tail(async_queue_t *q, void *data)
{if (!queue_is_full(q->_queue)){pthread_mutex_lock(&(q->mutex));queue_push_tail(q->_queue, data);if (q->waiting_threads > 0){pthread_cond_signal(&(q->cond));}pthread_mutex_unlock(&(q->mutex));}}void *async_queue_pop_head(async_queue_t *q, struct timeval *tv)
{void *retval &＃61; NULL;pthread_mutex_lock(&(q->mutex));if (queue_is_empty(q->_queue)){q->waiting_threads&＃43;&＃43;;while (queue_is_empty(q->_queue)){pthread_cond_wait(&(q->cond), &(q->mutex));}q->waiting_threads--;}retval &＃61; queue_pop_head(q->_queue);pthread_mutex_unlock(&(q->mutex));return retval;
}void async_queue_free(async_queue_t *q)
{queue_free(q->_queue);pthread_cond_destroy(&(q->cond));pthread_mutex_destroy(&(q->mutex));free(q);
}


###eventfd实现的异步队列

typedef struct async_queue
{int efd; //event fdfd_set rdfds; //for selectqueue_t *_queue;
} async_queue_t;
async_queue_t *async_queue_create(int size)
{async_queue_t *q &＃61; malloc(sizeof (async_queue_t));q->efd &＃61; eventfd(0, EFD_SEMAPHORE|EFD_NONBLOCK);q->_queue &＃61; queue_create(size);FD_ZERO(&(q->rdfds));FD_SET(q->efd, &(q->rdfds));return q;
}void async_queue_push_tail(async_queue_t *q, void *data)
{unsigned long long i &＃61; 1;if (!queue_is_full(q->_queue)){queue_push_tail(q->_queue, data);write(q->efd, &i, sizeof (i));}
}void *async_queue_pop_head(async_queue_t *q, struct timeval *tv)
{unsigned long long i &＃61; 0;void *data &＃61; NULL;if (select(q->efd &＃43; 1, &(q->rdfds), NULL, NULL, tv) &＃61;&＃61; 0){return data;}else{read(q->efd, &i, sizeof (i));return queue_pop_head(q->_queue);}
}void async_queue_free(async_queue_t *q)
{queue_free(q->_queue);close(q->efd);free(q);
}


###总结 两种实现方法线程条件变量比较复杂&＃xff0c;但是性能略高&＃xff0c;而eventfd实现简单&＃xff0c;但是性能略低。