碼上快樂

相關內容    簡體    繁體

linux下C語言實現多線程通信—環形緩沖區,可用於生產者(producer)/消費者(consumer)【轉】

本文轉載自   查看原文   2018-01-11 10:31   1538    【linux內核】/ 【嵌入式基礎】/ 【C語言】

轉自：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》源碼


免責聲明！

本站轉載的文章為個人學習借鑒使用，本站對版權不負任何法律責任。如果侵犯了您的隱私權益，請聯系本站郵箱yoyou2525@163.com刪除。



猜您在找 生產者消費者模式下的並發無鎖環形緩沖區 環形緩沖區的設計及其在生產者消費者模式下的使用（並發有鎖環形隊列） 環形緩沖區類(C++)絕對可用(轉) 線程安全的環形緩沖區實現 linux內核環形緩沖區【轉】 C#環形緩沖區（隊列）完全實現 C++ 環形緩沖區的實現 linux網絡編程--Circular Buffer(Ring Buffer) 環形緩沖區的設計與實現【轉】 再談多線程模型之生產者消費者（多生產者和多消費者 ）（c++11實現） [多線程]環形緩沖區以及多線程條件同步
 
粵ICP備18138465號   © 2018-2025 CODEPRJ.COM