亚洲免费在线-亚洲免费在线播放-亚洲免费在线观看-亚洲免费在线观看视频-亚洲免费在线看-亚洲免费在线视频

超輕量 pthread 集結點實現

系統 2267 0

我需要的 pthread 線程集結點功能,使用同一集結點的線程將通過 rend_wait 函數等待,當集結點到達指定數量的線程后同時激發繼續執行。使用 pthread 的 mutex 和 cond 超輕量實現。下面 rend.h 是集結點實現,rendezvous.c 是測試應用。

C代碼
  1. /*
  2. *rend.h
  3. *
  4. *Createdon:2009-11-14
  5. *Author:liuzy(lzy.dev@gmail.com)
  6. */
  7. #ifndefREND_H_
  8. #defineREND_H_
  9. #include<pthread.h>
  10. #include<assert.h>
  11. struct rend_t{
  12. volatile int count;
  13. pthread_mutex_tcount_lock;
  14. pthread_cond_tready;
  15. };
  16. #defineDECLARE_REND(name,count)/
  17. struct rend_tname={(count),PTHREAD_MUTEX_INITIALIZER,PTHREAD_COND_INITIALIZER}
  18. int rend_init( struct rend_t*prend, int count){
  19. int ret=0;
  20. assert(prend);
  21. prend->count=count;
  22. if ((ret=pthread_mutex_init(&prend->count_lock,NULL)))
  23. return ret;
  24. if ((ret=pthread_cond_init(&prend->ready,NULL)))
  25. return ret;
  26. return EXIT_SUCCESS;
  27. }
  28. int rend_wait( struct rend_t*prend){
  29. int ret=0;
  30. assert(prend);
  31. if ((ret=pthread_mutex_lock(&prend->count_lock)))
  32. return ret;
  33. /*checkcountvalueisreadytoweakupblockcode*/
  34. if (prend->count==1){
  35. if ((ret=pthread_cond_broadcast(&prend->ready)))
  36. return ret;
  37. if ((ret=pthread_mutex_unlock(&prend->count_lock)))
  38. return ret;
  39. } else {
  40. prend->count--;
  41. ret=pthread_cond_wait(&prend->ready,&prend->count_lock);
  42. prend->count++;
  43. if (ret){
  44. pthread_mutex_unlock(&prend->count_lock);
  45. return ret;
  46. }
  47. if ((ret=pthread_mutex_unlock(&prend->count_lock)))
  48. return ret;
  49. }
  50. return EXIT_SUCCESS;
  51. }
  52. int rend_free( struct rend_t*prend){
  53. int ret=0;
  54. assert(prend);
  55. prend->count=0;
  56. if ((ret=pthread_mutex_destroy(&prend->count_lock)))
  57. return ret;
  58. if ((ret=pthread_cond_destroy(&prend->ready)))
  59. return ret;
  60. return EXIT_SUCCESS;
  61. }
  62. #endif/*REND_H_*/

rend 使用更簡單:

  1. 定義/初始化 rend_t 集結點對象。DECLARE_REND 宏用于靜態定義,rend_init 函數可以對動態創建的集結點結構初始化;
  2. pthread 線程通過調用 rend_wait 函數 P/V 集結狀態。集結關系的線程要 P/V 在同一個 rend_t 集結對象上;
  3. 釋放集結對象,rend_free 函數。

以上函數都是成功返回 0,出錯返回 errno 值(非 0)。

C代碼
  1. /*
  2. ==============================
  3. Name:rendezvous.c
  4. Author:liuzy(lzy.dev@gmail.com)
  5. Version:0.1
  6. ==============================
  7. */
  8. #include<stdio.h>
  9. #include<stdlib.h>
  10. #include<stdarg.h>/*va_list*/
  11. #include<unistd.h>
  12. #include<string.h>
  13. #include<errno.h>/*errno*/
  14. #include<syslog.h>/*forsyslog(2)andlevel*/
  15. #include<pthread.h>
  16. #include"rend.h"
  17. static int daemon_proc=0; /*forsysloginerr_doit*/
  18. #defineMAXLINE4096/*maxtextlinelength*/
  19. void err_doit( int errnoflag, int level, const char *fmt, va_list ap){
  20. char buf[MAXLINE+1]={0};
  21. int errno_save=errno,n=0;
  22. #ifdefHAVE_VSNPRINTF
  23. vsnprintf(buf,MAXLINE,fmt,ap);
  24. #else
  25. vsprintf(buf,fmt,ap);
  26. #endif/*HAVE_VSNPRINTF*/
  27. n=strlen(buf);
  28. if (errnoflag)
  29. snprintf(buf+n,MAXLINE-n, ":%s" ,strerror(errno_save));
  30. strcat(buf, "/n" );
  31. if (daemon_proc){
  32. syslog(level, "%s" ,buf);
  33. } else {
  34. fflush(stdout);
  35. fputs(buf,stderr);
  36. fflush(stderr);
  37. }
  38. return ;
  39. }
  40. void err_msg( const char *fmt,...){
  41. va_list ap;
  42. va_start(ap,fmt);
  43. err_doit(0,LOG_INFO,fmt,ap);
  44. va_end(ap);
  45. return ;
  46. }
  47. void err_sys( const char *fmt,...){
  48. va_list ap;
  49. va_start(ap,fmt);
  50. err_doit(1,LOG_ERR,fmt,ap);
  51. va_end(ap);
  52. exit(EXIT_FAILURE);
  53. }
  54. #defineTHREAD_COUNT100/*rendezvoustestthreadworkers*/
  55. struct worker_arg{
  56. int worker_id;
  57. struct rend_t*prend;
  58. };
  59. static void *pthread_worker( void *arg){
  60. struct worker_arg*parg=( struct worker_arg*)arg;
  61. err_msg( "worker#%drunning." ,( int )parg->worker_id);
  62. srand(parg->worker_id*2);
  63. sleep(rand()%5);
  64. rend_wait(parg->prend); /*workersrendezvous*/
  65. err_msg( "worker#%dexiting." ,( int )parg->worker_id);
  66. return EXIT_SUCCESS;
  67. }
  68. int main( void ){
  69. int idx=0;
  70. void *exitcode=NULL;
  71. pthread_tthds[THREAD_COUNT];
  72. struct worker_argarg[THREAD_COUNT];
  73. DECLARE_REND(rend,THREAD_COUNT);
  74. err_msg( "workerscreating." );
  75. for (idx=0;idx<THREAD_COUNT;idx++){
  76. arg[idx].prend=&rend;
  77. arg[idx].worker_id=idx;
  78. if (pthread_create(thds+idx,NULL,pthread_worker,( void *)&arg[idx]))
  79. err_sys( "worker#%dcreateerror." ,idx);
  80. }
  81. puts( "workersexiting." );
  82. for (idx=0;idx<THREAD_COUNT;idx++)
  83. if (pthread_join(thds[idx],&exitcode)||(exitcode!=EXIT_SUCCESS))
  84. err_msg( "worker#%dexiterror." ,idx);
  85. err_msg( "alldone.exit0." );
  86. rend_free(&rend);
  87. return EXIT_SUCCESS;
  88. }

看了下 semaphore os syscall 及其 infrastructure,也許以后還需要進程間(非 pthread)集結時用得上。kernel 實現的超強啊,呵呵~

// 2009.11.17 14:34 添加 ////

快速用戶空間互斥鎖(Futex)
快速用戶空間互斥鎖(fast userspace mutex,Futex)是快速的用戶空間的鎖,是對傳統的System V同步方式的一種替代,傳統同步方式如:信號量、文件鎖和消息隊列,在每次鎖訪問時需要進行系統調用。而futex僅在有競爭的操作時才用系統調用訪問內核,這樣,在競爭出現較少的情況下,可以大幅度地減少工作負載
futex在非競爭情況下可從用戶空間獲取和釋放,不需要進入內核。與信號量類似,它有一個可以原子增減的計數器,進程可以等待計數器值變為正數。用戶進程通過系統調用對資源的競爭作一個公斷。
futex 是一個用戶空間的整數值,被多個線程或進程共享。Futex的系統調用對該整數值時進行操作,仲裁競爭的訪問。 glibc中的NPTL庫封裝了futex 系統調用,對futex接口進行了抽象。用戶通過NPTL庫像傳統編程一樣地使用線程同步API函數,而不會感覺到futex的存在。
futex 的實現機制是:如果當前進程訪問臨界區時,該臨界區正被另一個進程使用,當前進程將鎖用一個值標識,表示“有一個等待者正掛起”,并且調用 sys_futex(FUTEX_WAIT)等待其他進程釋放它。內核在內部創建futex隊列,以便以后與喚醒者匹配等待者。當臨界區擁有者線程釋放了 futex,它通過變量值發出通知表示還有多個等待者在掛起,并調用系統調用sys_futex(FUTEX_WAKE)喚醒它們。一旦所有等待者已獲取資源并釋放鎖時,futex回到非競爭狀態,并沒有內核狀態與它相關。
robust futex是為了解決futex鎖崩潰而對futex進行了增強。例如:當一個進程在持有pthread_mutex_t鎖正與其他進程發生競爭時,進程因某種意外原因而提前退出,如:進程發生段錯誤,或者被用戶用shell命令kill -9-ed”強行退出,此時,需要有一種機制告訴等待者“鎖的最一個持有者已經非正常地退出”。“
為了解決此類問題,NPTL創建了robust mutex用戶空間API pthread_mutex_lock(),如果鎖的擁有者進程提前退出,pthread_mutex_lock()返回一個錯誤值,新的擁有者進程可以決定是否可以安全恢復被鎖保護的數據。

有幾點不還不理解:

  1. “futex 如果說是一個用戶空間的整數值,那怎么被多個進程共享?Futex 系統調用在 kernel 態怎么操作該值并仲裁競爭?這是那種直接映射到 userspace 的 kernel 地址么。 這個需要程序間通過 mmap 在共享段中訪問,與 futex 沒什么關系。
  2. 這個“robust futex”機制指的應該就是 SVRx 傳統 sem IPC 里的 SEM_UNDO flag 吧?

一篇不錯的文章,引發對 glibc nptl 實現源碼的探索:

關于信號量與線程互斥鎖的區別與實現

超輕量 pthread 集結點實現


更多文章、技術交流、商務合作、聯系博主

微信掃碼或搜索:z360901061

微信掃一掃加我為好友

QQ號聯系: 360901061

您的支持是博主寫作最大的動力,如果您喜歡我的文章,感覺我的文章對您有幫助,請用微信掃描下面二維碼支持博主2元、5元、10元、20元等您想捐的金額吧,狠狠點擊下面給點支持吧,站長非常感激您!手機微信長按不能支付解決辦法:請將微信支付二維碼保存到相冊,切換到微信,然后點擊微信右上角掃一掃功能,選擇支付二維碼完成支付。

【本文對您有幫助就好】

您的支持是博主寫作最大的動力,如果您喜歡我的文章,感覺我的文章對您有幫助,請用微信掃描上面二維碼支持博主2元、5元、10元、自定義金額等您想捐的金額吧,站長會非常 感謝您的哦!!!

發表我的評論
最新評論 總共0條評論
主站蜘蛛池模板: 亚洲午夜日韩高清一区 | 亚洲 国产 路线1路线2路线 | 久久99久久精品国产只有 | 亚洲日韩欧美一区二区在线 | 成人爽视频 | 亚洲精品国产专区一区 | 曰本女人一级毛片看一级毛 | 精品香蕉一区二区三区 | 欧美精品一区二区三区在线 | 五月婷婷开心中文字幕 | 欧美日韩免费在线观看 | 久久www香蕉免费人成 | 国产不卡视频在线观看 | 99久久精品无码一区二区毛片 | 国产亚洲精品福利在线 | 一级特黄aaa免费 | 久久久久国产精品免费免费 | 一级黄色免费毛片 | 中文一级片| 九9热这里只有真品 | 精品久久久中文字幕二区 | 5g影院天天5g天天爽精品 | 午夜一级毛片免费视频 | 看真人一级毛多毛片 | 国产区精品一区二区不卡中文 | 麻豆国产高清精品国在线 | 亚洲国产影院 | 在线亚洲精品国产波多野结衣 | 久久国产精品国语对白 | 国产精品一区二区三区免费 | 久久最新 | 国产一区二区精品久久凹凸 | 人人爱天天做夜夜爽88 | 免费国产一区二区三区四区 | www.免费黄色 | 色综合天天综久久久噜噜噜久久〔 | 欧美一级毛片日本 | 国产在线视频不卡 | 牛人盗摄一区二区三区视频 | 日本毛片免费看 | 99精品视频免费在线观看 |