码迷，mamicode.com
首页 > 编程语言 > 详细

linux下C语言实现多线程通信—环形缓冲区,可用于生产者(producer)/消费者(consumer)【转】

时间：2018-01-11 16:11:58      阅读：216      评论：0      收藏：0      [点我收藏+]

标签：一个   ret   保存   structure   capacity   ifd   多线程   int   元素   

转自：http://blog.chinaunix.net/uid-28458801-id-4262445.html

操作系统：ubuntu10.04

前言：
    在嵌入式开发中，只要是带操作系统的，在其上开发产品应用，基本都需要用到多线程。
    为了提高效率，尽可能的提高并发率。因此，线程之间的通信就是问题的核心。
    根据当前产品需要，使用 环形缓冲区 解决。

一，环形缓冲区的实现
    1，cbuf.h

点击(此处)折叠或打开

  1. #ifndef __CBUF_H__
  2. #define __CBUF_H__

  4. #ifdef __cplusplus
  5. extern "C" {
  6. #endif

  8. /* Define to prevent recursive inclusion
  9. -------------------------------------*/
  10. #include "types.h"
  11. #include "thread.h"


  14. typedef    struct _cbuf
  15. {
  16.     int32_t        size;            /* 当前缓冲区中存放的数据的个数 */
  17.     int32_t        next_in;        /* 缓冲区中下一个保存数据的位置 */
  18.     int32_t        next_out;        /* 从缓冲区中取出下一个数据的位置 */
  19.     int32_t        capacity;        /* 这个缓冲区的可保存的数据的总个数 */
  20.     mutex_t        mutex;            /* Lock the structure */
  21.     cond_t        not_full;        /* Full -> not full condition */
  22.     cond_t        not_empty;        /* Empty -> not empty condition */
  23.     void        *data[CBUF_MAX];/* 缓冲区中保存的数据指针 */
  24. }cbuf_t;


  27. /* 初始化环形缓冲区 */
  28. extern    int32_t        cbuf_init(cbuf_t *c);

  30. /* 销毁环形缓冲区 */
  31. extern    void        cbuf_destroy(cbuf_t    *c);

  33. /* 压入数据 */
  34. extern    int32_t        cbuf_enqueue(cbuf_t *c,void *data);

  36. /* 取出数据 */
  37. extern    void*        cbuf_dequeue(cbuf_t *c);


  40. /* 判断缓冲区是否为满 */
  41. extern    bool        cbuf_full(cbuf_t    *c);

  43. /* 判断缓冲区是否为空 */
  44. extern    bool        cbuf_empty(cbuf_t *c);

  46. /* 获取缓冲区可存放的元素的总个数 */
  47. extern    int32_t        cbuf_capacity(cbuf_t *c);


  50. #ifdef __cplusplus
  51. }
  52. #endif

  54. #endif
  55. /* END OF FILE
  56. ---------------------------------------------------------------*/


    2，cbuf.c

点击(此处)折叠或打开

  1. #include "cbuf.h"



  5. /* 初始化环形缓冲区 */
  6. int32_t        cbuf_init(cbuf_t *c)
  7. {
  8.     int32_t    ret = OPER_OK;

  10.     if((ret = mutex_init(&c->mutex)) != OPER_OK)    
  11.     {
  12. #ifdef DEBUG_CBUF
  13.     debug("cbuf init fail ! mutex init fail !\n");
  14. #endif
  15.         return ret;
  16.     }

  18.     if((ret = cond_init(&c->not_full)) != OPER_OK)    
  19.     {
  20. #ifdef DEBUG_CBUF
  21.     debug("cbuf init fail ! cond not full init fail !\n");
  22. #endif
  23.         mutex_destroy(&c->mutex);
  24.         return ret;
  25.     }

  27.     if((ret = cond_init(&c->not_empty)) != OPER_OK)
  28.     {
  29. #ifdef DEBUG_CBUF
  30.     debug("cbuf init fail ! cond not empty init fail !\n");
  31. #endif
  32.         cond_destroy(&c->not_full);
  33.         mutex_destroy(&c->mutex);
  34.         return ret;
  35.     }

  37.     c->size     = 0;
  38.     c->next_in    = 0;
  39.     c->next_out = 0;
  40.     c->capacity    = CBUF_MAX;

  42. #ifdef DEBUG_CBUF
  43.     debug("cbuf init success !\n");
  44. #endif

  46.     return ret;
  47. }


  50. /* 销毁环形缓冲区 */
  51. void        cbuf_destroy(cbuf_t    *c)
  52. {
  53.     cond_destroy(&c->not_empty);
  54.     cond_destroy(&c->not_full);
  55.     mutex_destroy(&c->mutex);

  57. #ifdef DEBUG_CBUF
  58.     debug("cbuf destroy success \n");
  59. #endif
  60. }



  64. /* 压入数据 */
  65. int32_t        cbuf_enqueue(cbuf_t *c,void *data)
  66. {
  67.     int32_t    ret = OPER_OK;

  69.     if((ret = mutex_lock(&c->mutex)) != OPER_OK)    return ret;

  71.     /*
  72.      * Wait while the buffer is full.
  73.      */
  74.     while(cbuf_full(c))
  75.     {
  76. #ifdef DEBUG_CBUF
  77.     debug("cbuf is full !!!\n");
  78. #endif
  79.         cond_wait(&c->not_full,&c->mutex);
  80.     }

  82.     c->data[c->next_in++] = data;
  83.     c->size++;
  84.     c->next_in %= c->capacity;

  86.     mutex_unlock(&c->mutex);

  88.     /*
  89.      * Let a waiting consumer know there is data.
  90.      */
  91.     cond_signal(&c->not_empty);

  93. #ifdef DEBUG_CBUF
  94. //    debug("cbuf enqueue success ,data : %p\n",data);
  95.     debug("enqueue\n");
  96. #endif

  98.     return ret;
  99. }



  103. /* 取出数据 */
  104. void*        cbuf_dequeue(cbuf_t *c)
  105. {
  106.     void     *data     = NULL;
  107.     int32_t    ret     = OPER_OK;

  109.     if((ret = mutex_lock(&c->mutex)) != OPER_OK)    return NULL;

  111.        /*
  112.      * Wait while there is nothing in the buffer
  113.      */
  114.     while(cbuf_empty(c))
  115.     {
  116. #ifdef DEBUG_CBUF
  117.     debug("cbuf is empty!!!\n");
  118. #endif
  119.         cond_wait(&c->not_empty,&c->mutex);
  120.     }

  122.     data = c->data[c->next_out++];
  123.     c->size--;
  124.     c->next_out %= c->capacity;

  126.     mutex_unlock(&c->mutex);


  129.     /*
  130.      * Let a waiting producer know there is room.
  131.      * 取出了一个元素，又有空间来保存接下来需要存储的元素
  132.      */
  133.     cond_signal(&c->not_full);

  135. #ifdef DEBUG_CBUF
  136. //    debug("cbuf dequeue success ,data : %p\n",data);
  137.     debug("dequeue\n");
  138. #endif

  140.     return data;
  141. }


  144. /* 判断缓冲区是否为满 */
  145. bool        cbuf_full(cbuf_t    *c)
  146. {
  147.     return (c->size == c->capacity);
  148. }

  150. /* 判断缓冲区是否为空 */
  151. bool        cbuf_empty(cbuf_t *c)
  152. {
  153.     return (c->size == 0);
  154. }

  156. /* 获取缓冲区可存放的元素的总个数 */
  157. int32_t        cbuf_capacity(cbuf_t *c)
  158. {
  159.     return c->capacity;
  160. }



二，辅助文件
    为了提高程序的移植性，对线程相关进行封装。
    1，thread.h

点击(此处)折叠或打开

  1. #ifndef __THREAD_H__
  2. #define __THREAD_H__

  4. #ifdef __cplusplus
  5. extern "C" {
  6. #endif

  8. /* Define to prevent recursive inclusion
  9. -------------------------------------*/
  10. #include "types.h"





  16. typedef    struct _mutex
  17. {
  18.     pthread_mutex_t        mutex;
  19. }mutex_t;


  22. typedef    struct _cond
  23. {
  24.     pthread_cond_t        cond;
  25. }cond_t;


  28. typedef    pthread_t        tid_t;
  29. typedef    pthread_attr_t    attr_t;
  30. typedef    void*    (* thread_fun_t)(void*);


  33. typedef    struct _thread
  34. {
  35.     tid_t            tid;
  36.     cond_t            *cv;
  37.     int32_t            state;
  38.     int32_t            stack_size;
  39.     attr_t         attr;
  40.     thread_fun_t    fun;
  41. }thread_t;



  45. /* mutex */
  46. extern    int32_t        mutex_init(mutex_t    *m);
  47. extern    int32_t        mutex_destroy(mutex_t    *m);
  48. extern    int32_t        mutex_lock(mutex_t    *m);
  49. extern    int32_t        mutex_unlock(mutex_t    *m);


  52. /* cond */
  53. extern    int32_t        cond_init(cond_t    *c);
  54. extern    int32_t        cond_destroy(cond_t    *c);
  55. extern    int32_t        cond_signal(cond_t *c);
  56. extern    int32_t        cond_wait(cond_t    *c,mutex_t *m);



  60. /* thread */
  61. /* 线程的创建，其属性的设置等都封装在里面 */
  62. extern    int32_t        thread_create(thread_t *t);
  63. //extern    int32_t        thread_init(thread_t    *t);

  65. #define    thread_join(t, p)     pthread_join(t, p)
  66. #define    thread_self()        pthread_self()
  67. #define    thread_sigmask        pthread_sigmask


  70. #ifdef __cplusplus
  71. }
  72. #endif

  74. #endif
  75. /* END OF FILE
  76. ---------------------------------------------------------------*/


    2，thread.c

点击(此处)折叠或打开

  1. #include "thread.h"




  6. /* mutex */
  7. int32_t        mutex_init(mutex_t    *m)
  8. {
  9.     int32_t        ret = OPER_OK;

  11.     if((ret = pthread_mutex_init(&m->mutex, NULL)) != 0)
  12.         ret = -THREAD_MUTEX_INIT_ERROR;

  14.     return ret;
  15. }


  18. int32_t        mutex_destroy(mutex_t    *m)
  19. {
  20.     int32_t        ret = OPER_OK;

  22.     if((ret = pthread_mutex_destroy(&m->mutex)) != 0)
  23.         ret = -MUTEX_DESTROY_ERROR;

  25.     return ret;
  26. }



  30. int32_t        mutex_lock(mutex_t    *m)
  31. {
  32.     int32_t        ret = OPER_OK;

  34.     if((ret = pthread_mutex_lock(&m->mutex)) != 0)
  35.         ret = -THREAD_MUTEX_LOCK_ERROR;

  37.     return ret;
  38. }



  42. int32_t        mutex_unlock(mutex_t    *m)
  43. {
  44.     int32_t        ret = OPER_OK;

  46.     if((ret = pthread_mutex_unlock(&m->mutex)) != 0)
  47.         ret = -THREAD_MUTEX_UNLOCK_ERROR;
  48.     
  49.     return ret;
  50. }






  57. /* cond */
  58. int32_t        cond_init(cond_t    *c)
  59. {
  60.     int32_t        ret = OPER_OK;

  62.     if((ret = pthread_cond_init(&c->cond, NULL)) != 0)
  63.         ret = -THREAD_COND_INIT_ERROR;

  65.     return ret;
  66. }



  70. int32_t        cond_destroy(cond_t    *c)
  71. {
  72.     int32_t        ret = OPER_OK;

  74.     if((ret = pthread_cond_destroy(&c->cond)) != 0)
  75.         ret = -COND_DESTROY_ERROR;
  76.     
  77.     return ret;
  78. }



  82. int32_t        cond_signal(cond_t *c)
  83. {
  84.     int32_t        ret = OPER_OK;


  87.     if((ret = pthread_cond_signal(&c->cond)) != 0)
  88.         ret = -COND_SIGNAL_ERROR;
  89.     
  90.     return ret;
  91. }




  96. int32_t        cond_wait(cond_t    *c,mutex_t *m)
  97. {
  98.     int32_t        ret = OPER_OK;

  100.     if((ret = pthread_cond_wait(&c->cond, &m->mutex)) != 0)
  101.         ret = -COND_WAIT_ERROR;    
  102.     
  103.     return ret;
  104. }



三，测试
    1，测试代码

点击(此处)折叠或打开

  1. /*
  2.  * cbuf begin
  3.  */
  4. #define        OVER    (-1)

  6. static        cbuf_t    cmd;
  7. static        int        line_1[200];
  8. static        int        line_2[200];
  9. //static        int        temp = 0;

  11. static        bool    line1_finish = false;
  12. static        bool    line2_finish = false;

  14. void*    producer_1(void *data)
  15. {
  16.     int32_t    i = 0;

  18.     for(i = 0; i < 200; i++)
  19.     {
  20.         line_1[i] = i+1000;
  21.         cbuf_enqueue(&cmd, &line_1[i]);

  23.         if(0 == (i % 9)) sleep(1);
  24.     }

  26.     line1_finish = true;

  28.     return NULL;
  29. }

  31. void*    producer_2(void *data)
  32. {
  33.     int32_t    i = 0;

  35.     for(i = 0; i < 200; i++)
  36.     {
  37.         line_2[i] = i+20000;
  38.         cbuf_enqueue(&cmd, &line_2[i]);

  40.         if(0 == (i % 9)) sleep(1);
  41.     }

  43.     line2_finish = true;

  45.     return NULL;
  46. }


  49. void*    consumer(void *data)
  50. {
  51.     int32_t        *ptr = NULL;

  53.     while(1)
  54.     {
  55.         ptr = cbuf_dequeue(&cmd);
  56.         printf("%d\n",*ptr);

  58.         if(cbuf_empty(&cmd) && line2_finish && line1_finish)
  59.         {
  60.             printf("quit\n");
  61.             break;
  62.         }
  63.     }

  65.     return NULL;
  66. }


  69. void    test_cbuf_oper(void)
  70. {
  71.     pthread_t    l_1;
  72.     pthread_t    l_2;
  73.     pthread_t    c;
  74.     
  75.     cbuf_init(&cmd);

  77.     pthread_create(&l_1,NULL,producer_1,0);
  78.     pthread_create(&l_2,NULL,producer_2,0);
  79.     pthread_create(&c,NULL,consumer,0);

  81.     pthread_join(l_1,NULL);
  82.     pthread_join(l_2,NULL);
  83.     pthread_join(c,NULL);

  85.     cbuf_destroy(&cmd);
  86. }


  89. void    test_cbuf(void)
  90. {
  91.     test_cbuf_oper();
  92. }


  95. /*
  96.  * cbuf end
  97.  */


    2，测试结果
技术分享图片


四，参考文件
1，《bareos-master》源码
2，《nginx》源码

linux下C语言实现多线程通信—环形缓冲区,可用于生产者(producer)/消费者(consumer)【转】

标签：一个   ret   保存   structure   capacity   ifd   多线程   int   元素   

原文地址：https://www.cnblogs.com/sky-heaven/p/8267211.html
踩
(0)
赞
(0)
   
举报
评论 一句话评论（0）
登录后才能评论！
分享档案
更多>
2021年07月29日 (22)
2021年07月28日 (40)
2021年07月27日 (32)
2021年07月26日 (79)
2021年07月23日 (29)
2021年07月22日 (30)
2021年07月21日 (42)
2021年07月20日 (16)
2021年07月19日 (90)
2021年07月16日 (35)
周排行
mamicode.com排行更多图片
更多
友情链接
兰亭集智   国之画   百度统计   站长统计   阿里云   chrome插件   新版天听网
关于我们 - 联系我们 - 留言反馈
© 2014 mamicode.com 版权所有  联系我们:gaon5@hotmail.com
迷上了代码！