[Queue] Thread 간 통신을 위한 Async Queue (비동기 큐)

AsyncQueue라고 표현하면 뭔가 이해하기 쉽지 않습니다. AsyncQueue의 목적은 실제 Queue로 사용하기 보단 Multi-thread 환경에서 thread간 데이터 송수신을 위해 사용합니다.

AsyncQueue에 대해 살펴보겠습니다. 저는 AsyncQueue를 pthread(mutex, cond) 기반으로 작성하였습니다. 

Header에 include되어 있는 다른 header와 source 파일은 아래 URL을 참고해주세요.

https://github.com/linuxias/linux-system-programming/tree/master/DataStruct

아래 Header를 보시면 SAsyncQueue 구조체에 mutex, cond, SQueue (자체적으로 구현한 Queue) 등이 있습니다. 아직 미완성이지만 기본적인 동작은 합니다 ^^;;

s_async_queue_push()와 s_async_queue_pop() 함수를 주의깊게 보시면 됩니다. 

s_async_queue.h

#ifndef __S_ASYNC_QUEUE_H__
#define __S_ASYNC_QUEUE_H__
 
#include <pthread.h>
 
#include "s_macros.h"
#include "s_queue.h"
 
typedef struct _SAsyncQueue SAsyncQueue;
 
struct _SAsyncQueue {
    pthread_mutex_t mtx;
    pthread_cond_t cond;
    SQueue queue;
    unsigned int waiting_threads;
    unsigned int ref_count;
};
 
SAsyncQueue *s_async_queue_create(void);
 
void s_async_queue_destroy(SAsyncQueue *queue);
 
void s_async_queue_push(SAsyncQueue *queue, void *data);
 
void *s_async_queue_pop(SAsyncQueue *queue);
 
#endif /* __S_AYNC_QUEUE_H__  */

s_async_queue.c

#include <stdlib.h>
 
#include <pthread.h>
 
#include "s_async_queue.h"
 
SAsyncQueue *s_async_queue_create(void)
{
    SAsyncQueue *queue;
 
    queue = (SAsyncQueue *)calloc(1, sizeof(SAsyncQueue));
    if (queue == NULL)
        return NULL;
 
    pthread_mutex_init(&queue->mtx, NULL);
    pthread_cond_init(&queue->cond, NULL);
    queue->ref_count = 1;
 
    return queue;
}
 
void s_async_queue_destroy(SAsyncQueue *queue)
{
    if S_UNLIKELY (queue == NULL)
        return;
 
    pthread_mutex_destroy(&queue->mtx);
    pthread_cond_destroy(&queue->cond);
    s_queue_destroy(&queue->queue);
 
    free(queue);
}
 
void s_async_queue_push(SAsyncQueue *queue, void *data)
{
    if S_UNLIKELY (queue == NULL)
        return;
    if S_UNLIKELY (data == NULL)
        return;
 
    pthread_mutex_lock(&queue->mtx);
    s_queue_push(&queue->queue, data);
    if (queue->waiting_threads > 0)
        pthread_cond_signal(&queue->cond);
    pthread_mutex_unlock(&queue->mtx);
}
 
void *s_async_queue_pop(SAsyncQueue *queue)
{
    void *ret_data;
 
    pthread_mutex_lock(&queue->mtx);
    if (!s_queue_peek_tail_link(&queue->queue)) {
        queue->waiting_threads++;
 
        while (!s_queue_peek_tail_link(&queue->queue))
            pthread_cond_wait(&queue->cond, &queue->mtx);
 
        queue->waiting_threads--;
    }
 
    ret_data = s_queue_pop(&queue->queue);
    pthread_mutex_unlock(&queue->mtx);
 
    return ret_data;
}