linux线程(C)：信号量实现的线程安全队列

用信号量实现的线程安全队列。

简单有用的示例程序，

比起互斥量的实现在多线程时效率更好。

cir_queue.h

/*
 * \File
 *   cir_queue.h
 * \Brief
 *   circular queue
 */
#ifndef __CIR_QUEUE_H__
#define __CIR_QUEUE_H__

#define QUE_SIZE 8

typedef int DataType;
typedef struct cir_queue_t
{
  DataType data[QUE_SIZE];
  int front;
  int rear;
  int count;
}cir_queue_t;

extern sem_t queue_sem;

void init_cir_queue(cir_queue_t* q);
int  is_empty_cir_queue(cir_queue_t* q);
int  is_full_cir_queue(cir_queue_t* q);
void push_cir_queue(cir_queue_t* q, DataType x);
DataType pop_cir_queue(cir_queue_t* q);
DataType top_cir_queue(cir_queue_t* q);
void destroy_cir_queue(cir_queue_t* q);
void print_queue(cir_queue_t* q);

#endif

main.c

/*
 * \File
 *   main.c
 * \Breif
 *   Thread-safe circular-queue implemented by semaphore
 * \Author
 *   Hank.yan
 */
#include <stdio.h>
#include <stdlib.h>
#include <unistd.h>
#include <string.h>
#include <pthread.h>
#include <semaphore.h>

#include "cir_queue.h"

void* thread_queue(void *arg);


/*
 * \Func
 *   main
 */
int main(int argc, char* argv[])
{
  int res;
  cir_queue_t cq;
  DataType e;

  pthread_t a_thread, b_thread;
  void* thread_result;

  init_cir_queue(&cq);

  push_cir_queue(&cq, 1);
  push_cir_queue(&cq, 2);
  push_cir_queue(&cq, 3);

  print_queue(&cq);

  res = pthread_create(&a_thread, NULL, thread_queue, (void*)&cq);
  if (res != 0)
  {
    perror("Thread creation failed.");
    exit(EXIT_FAILURE);
  }

  e = pop_cir_queue(&cq);
  e = pop_cir_queue(&cq);
  print_queue(&cq);

  push_cir_queue(&cq, 9);
  push_cir_queue(&cq, 100);

  print_queue(&cq);

  res = pthread_create(&b_thread, NULL, thread_queue, (void*)&cq);
  if (res != 0)
  {
    perror("Thread creation failed.");
    exit(EXIT_FAILURE);
  }
  e = pop_cir_queue(&cq);


  push_cir_queue(&cq, 20);
  print_queue(&cq);

  printf("Waiting for thread to finish...\n");
  res = pthread_join(a_thread, &thread_result);
  if (res != 0)
  {
    perror("Thread join failed.");
    exit(EXIT_FAILURE);
  }
  print_queue(&cq);

  printf("Waiting for thread to finish...\n");
  res = pthread_join(b_thread, &thread_result);
  if (res != 0)
  {
    perror("Thread join failed.");
    exit(EXIT_FAILURE);
  }

  destroy_cir_queue(&cq);

  printf("Thread joined, it returned %s\n", (char*)thread_result);
  exit(EXIT_SUCCESS);
}


void *thread_queue(void *cirqueue)
{
  int flag;
  DataType element;

  print_queue((cir_queue_t*)cirqueue);

  flag = is_empty_cir_queue((cir_queue_t*)cirqueue);

  print_queue((cir_queue_t*)cirqueue);
  element = pop_cir_queue((cir_queue_t*)cirqueue);
  element = pop_cir_queue((cir_queue_t*)cirqueue);
  print_queue((cir_queue_t*)cirqueue);

  push_cir_queue((cir_queue_t*)cirqueue, 5);
  print_queue((cir_queue_t*)cirqueue);

  push_cir_queue((cir_queue_t*)cirqueue, 99);
  push_cir_queue((cir_queue_t*)cirqueue, 1000);
  push_cir_queue((cir_queue_t*)cirqueue, 88);

  print_queue((cir_queue_t*)cirqueue);

  pthread_exit("Thank you for the cpu time.");
}

cir_queue.c

/*
 * \File
 *   cir_queue.c
 */
#include <stdio.h>
#include <stdlib.h>
#include <unistd.h>
#include <string.h>
#include <pthread.h>
#include <semaphore.h>

#include "cir_queue.h"

sem_t queue_sem;
/*
 * \Func
 *
 */
void init_cir_queue(cir_queue_t *q)
{
  int res;

  /* Create semaphore */
  res = sem_init(&queue_sem, 0, QUE_SIZE);
  if (res != 0)
  {
    perror("Semaphore init failed.\n");
    exit(EXIT_FAILURE);
  }
  memset(q->data, 0, QUE_SIZE*sizeof(DataType));

  q->front = q->rear = 0;
  q->count = 0;
}

/*
 * \Func
 *
 */
int  is_empty_cir_queue(cir_queue_t *q)
{
  int empty_flag;

  sem_wait(&queue_sem);

  empty_flag = q->front == q->rear;

  sem_post(&queue_sem);

  return empty_flag;
}

/*
 * \Func
 *
 */
int  is_full_cir_queue(cir_queue_t *q)
{
  int full_flag;

  sem_wait(&queue_sem);

  full_flag = q->rear == QUE_SIZE - 1 + q->front;

  sem_post(&queue_sem);

  return full_flag;
}

/*
 * \Func
 *
 */
void push_cir_queue(cir_queue_t *q, DataType x)
{

  if (is_full_cir_queue(q))
  {
    printf("queue overflow.\n");
    return ;
  }

  sem_wait(&queue_sem);

  q->count++;
  q->data[q->rear] = x;
  q->rear = (q->rear+1) % QUE_SIZE;

  sem_post(&queue_sem);

}

/*
 * \Func
 *
 */
DataType pop_cir_queue(cir_queue_t *q)
{
  DataType temp;

  if (is_empty_cir_queue(q))
  {
    printf("queue empty.\n");
    return 0;
  }

  sem_wait(&queue_sem);

  temp = q->data[q->front];
  q->data[q->front] = 0;
  q->count--;
  q->front = (q->front+1) % QUE_SIZE;

  sem_post(&queue_sem);

  return temp;
}

/*
 * \Func
 *
 */
DataType top_cir_queue(cir_queue_t *q)
{
  DataType x;

  if (is_empty_cir_queue(q))
  {
    printf("queue is empty.\n");
    return 0;
  }

  sem_wait(&queue_sem);

  x = q->data[q->front];

  sem_post(&queue_sem);

  return x;
}

void destroy_cir_queue(cir_queue_t *q)
{
  sem_destroy(&queue_sem);

  return;
}

void print_queue(cir_queue_t* q)
{
  int index;
  if (is_empty_cir_queue(q))
  {
    printf("queue is empty.\n");
    return;
  }

  sem_wait(&queue_sem);
  printf("QUEUE: ");
  for (index = 0; index < QUE_SIZE; index++)
  {
    printf(" %d ", q->data[index]);
  }
  printf("\n");

  sem_post(&queue_sem);

  return;
}

makefile

OBJECTS = main.o cir_queue.o
CC = gcc
CFLAGS = -D_REENTRANT -lpthread -g -Wall


thrd_safe_queue: $(OBJECTS)
  $(CC) $(CFLAGS) -o thrd_safe_queue $(OBJECTS)

main.o: cir_queue.h
cir_queue.o: cir_queue.h


.PHONY:clean
clean:
  rm thrd_safe_queue $(OBJECTS)