zoukankan      html  css  js  c++  java

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

    转自：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 ! ");
    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 ! ");
    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 ! ");
    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 ! ");
    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 ");
    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 !!! ");
    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 ",data);
    95.     debug("enqueue ");
    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!!! ");
    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 ",data);
    137.     debug("dequeue ");
    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 ",*ptr);

    58.         if(cbuf_empty(&cmd) && line2_finish && line1_finish)
    59.         {
    60.             printf("quit ");
    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》源码
  • 相关阅读:
    2级搭建类203-Oracle 19c SI ASM 静默搭建（OEL7.7）
    2级搭建类EM-Oracle EMCC 13c Release 3 在 OEL 7.7 上的搭建
    1级搭建类112-Oracle 19c SI FS（CentOS 8）
    0级搭建类013-CentOS 8.x 安装
    List添加map，后添加的map覆盖前面的问题
    mysql插入数据报错1366
    oracle ora-12514解决办法
    easyUI 创建详情页dialog
    Server Tomcat v7.0 Server at localhost failed to start.
    maven项目启动报错；class path resource [com/ssm/mapping/] cannot be resolved to URL because it does not exist
  • 原文地址：https://www.cnblogs.com/sky-heaven/p/8267211.html
Copyright © 2011-2022 走看看