libcm is a C development framework with an emphasis on audio signal processing applications.
您最多选择25个主题 主题必须以字母或数字开头,可以包含连字符 (-),并且长度不得超过35个字符

cmThread.c 49KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695696697698699700701702703704705706707708709710711712713714715716717718719720721722723724725726727728729730731732733734735736737738739740741742743744745746747748749750751752753754755756757758759760761762763764765766767768769770771772773774775776777778779780781782783784785786787788789790791792793794795796797798799800801802803804805806807808809810811812813814815816817818819820821822823824825826827828829830831832833834835836837838839840841842843844845846847848849850851852853854855856857858859860861862863864865866867868869870871872873874875876877878879880881882883884885886887888889890891892893894895896897898899900901902903904905906907908909910911912913914915916917918919920921922923924925926927928929930931932933934935936937938939940941942943944945946947948949950951952953954955956957958959960961962963964965966967968969970971972973974975976977978979980981982983984985986987988989990991992993994995996997998999100010011002100310041005100610071008100910101011101210131014101510161017101810191020102110221023102410251026102710281029103010311032103310341035103610371038103910401041104210431044104510461047104810491050105110521053105410551056105710581059106010611062106310641065106610671068106910701071107210731074107510761077107810791080108110821083108410851086108710881089109010911092109310941095109610971098109911001101110211031104110511061107110811091110111111121113111411151116111711181119112011211122112311241125112611271128112911301131113211331134113511361137113811391140114111421143114411451146114711481149115011511152115311541155115611571158115911601161116211631164116511661167116811691170117111721173117411751176117711781179118011811182118311841185118611871188118911901191119211931194119511961197119811991200120112021203120412051206120712081209121012111212121312141215121612171218121912201221122212231224122512261227122812291230123112321233123412351236123712381239124012411242124312441245124612471248124912501251125212531254125512561257125812591260126112621263126412651266126712681269127012711272127312741275127612771278127912801281128212831284128512861287128812891290129112921293129412951296129712981299130013011302130313041305130613071308130913101311131213131314131513161317131813191320132113221323132413251326132713281329133013311332133313341335133613371338133913401341134213431344134513461347134813491350135113521353135413551356135713581359136013611362136313641365136613671368136913701371137213731374137513761377137813791380138113821383138413851386138713881389139013911392139313941395139613971398139914001401140214031404140514061407140814091410141114121413141414151416141714181419142014211422142314241425142614271428142914301431143214331434143514361437143814391440144114421443144414451446144714481449145014511452145314541455145614571458145914601461146214631464146514661467146814691470147114721473147414751476147714781479148014811482148314841485148614871488148914901491149214931494149514961497149814991500150115021503150415051506150715081509151015111512151315141515151615171518151915201521152215231524152515261527152815291530153115321533153415351536153715381539154015411542154315441545154615471548154915501551155215531554155515561557155815591560156115621563156415651566156715681569157015711572157315741575157615771578157915801581158215831584158515861587158815891590159115921593159415951596159715981599160016011602160316041605160616071608160916101611161216131614161516161617161816191620162116221623162416251626162716281629163016311632163316341635163616371638163916401641164216431644164516461647164816491650165116521653165416551656165716581659166016611662166316641665166616671668166916701671167216731674167516761677167816791680168116821683168416851686168716881689169016911692169316941695169616971698169917001701170217031704170517061707170817091710171117121713171417151716171717181719172017211722172317241725172617271728172917301731173217331734173517361737173817391740174117421743174417451746174717481749175017511752175317541755175617571758175917601761176217631764176517661767176817691770177117721773177417751776177717781779178017811782178317841785178617871788178917901791179217931794179517961797179817991800180118021803180418051806180718081809181018111812181318141815181618171818181918201821182218231824182518261827182818291830183118321833183418351836183718381839184018411842184318441845184618471848184918501851185218531854185518561857185818591860186118621863186418651866186718681869187018711872187318741875187618771878187918801881188218831884188518861887188818891890189118921893189418951896189718981899190019011902190319041905190619071908190919101911191219131914191519161917191819191920192119221923192419251926192719281929193019311932193319341935193619371938193919401941194219431944194519461947194819491950195119521953195419551956195719581959196019611962196319641965196619671968196919701971197219731974197519761977197819791980198119821983198419851986198719881989199019911992199319941995199619971998199920002001200220032004200520062007200820092010201120122013201420152016201720182019202020212022202320242025202620272028202920302031203220332034203520362037203820392040204120422043204420452046204720482049205020512052205320542055205620572058205920602061206220632064206520662067206820692070207120722073207420752076207720782079208020812082208320842085208620872088208920902091
  1. #include "cmPrefix.h"
  2. #include "cmGlobal.h"
  3. #include "cmRpt.h"
  4. #include "cmErr.h"
  5. #include "cmMem.h"
  6. #include "cmMallocDebug.h"
  7. #include "cmThread.h"
  8. #include <pthread.h>
  9. #include <unistd.h> // usleep
  10. #ifdef OS_OSX
  11. #include <libkern/OSAtomic.h>
  12. #endif
  13. cmThreadH_t cmThreadNullHandle = {NULL};
  14. enum
  15. {
  16. kDoExitThFl = 0x01,
  17. kDoPauseThFl = 0x02,
  18. kDoRunThFl = 0x04
  19. };
  20. typedef struct
  21. {
  22. cmErr_t err;
  23. cmThreadFunc_t funcPtr;
  24. pthread_t pthreadH;
  25. cmThStateId_t state;
  26. void* funcParam;
  27. unsigned doFlags;
  28. unsigned pauseMicroSecs;
  29. unsigned waitMicroSecs;
  30. } cmThThread_t;
  31. cmThRC_t _cmThError( cmErr_t* err, cmThRC_t rc, int sysErr, const char* fmt, ... )
  32. {
  33. va_list vl;
  34. va_start(vl,fmt);
  35. cmErrVSysMsg(err,rc,sysErr,fmt,vl);
  36. va_end(vl);
  37. return rc;
  38. }
  39. void _cmThThreadCleanUpCallback(void* t)
  40. {
  41. ((cmThThread_t*)t)->state = kExitedThId;
  42. }
  43. void* _cmThThreadCallback(void* param)
  44. {
  45. cmThThread_t* t = (cmThThread_t*)param;
  46. // set a clean up handler - this will be called when the
  47. // thread terminates unexpectedly or pthread_cleanup_pop() is called.
  48. pthread_cleanup_push(_cmThThreadCleanUpCallback,t);
  49. while( cmIsFlag(t->doFlags,kDoExitThFl) == false )
  50. {
  51. if( t->state == kPausedThId )
  52. {
  53. cmSleepUs( t->pauseMicroSecs );
  54. if( cmIsFlag(t->doFlags,kDoRunThFl) )
  55. {
  56. t->doFlags = cmClrFlag(t->doFlags,kDoRunThFl);
  57. t->state = kRunningThId;
  58. }
  59. }
  60. else
  61. {
  62. if( t->funcPtr(t->funcParam)==false )
  63. break;
  64. if( cmIsFlag(t->doFlags,kDoPauseThFl) )
  65. {
  66. t->doFlags = cmClrFlag(t->doFlags,kDoPauseThFl);
  67. t->state = kPausedThId;
  68. }
  69. }
  70. }
  71. pthread_cleanup_pop(1);
  72. pthread_exit(NULL);
  73. return t;
  74. }
  75. cmThThread_t* _cmThThreadFromHandle( cmThreadH_t h )
  76. {
  77. cmThThread_t* tp = (cmThThread_t*)h.h;
  78. assert(tp != NULL);
  79. return tp->state==kNotInitThId ? NULL : tp;
  80. }
  81. cmThRC_t _cmThWaitForState( cmThThread_t* t, unsigned stateId )
  82. {
  83. unsigned waitTimeMicroSecs = 0;
  84. while( t->state != stateId && waitTimeMicroSecs < t->waitMicroSecs )
  85. {
  86. //cmSleepUs( t->waitMicroSecs );
  87. cmSleepUs( 15000 );
  88. waitTimeMicroSecs += 15000; //t->waitMicroSecs;
  89. }
  90. return t->state==stateId ? kOkThRC : kTimeOutThRC;
  91. }
  92. cmThRC_t cmThreadCreate( cmThreadH_t* hPtr, cmThreadFunc_t funcPtr, void* funcParam, cmRpt_t* rpt )
  93. {
  94. //pthread_attr_t attr;
  95. cmThRC_t rc = kOkThRC;
  96. cmThThread_t* tp = cmMemAllocZ( cmThThread_t, 1 );
  97. int sysErr;
  98. cmErrSetup(&tp->err,rpt,"Thread");
  99. tp->funcPtr = funcPtr;
  100. tp->funcParam = funcParam;
  101. tp->state = kPausedThId;
  102. tp->doFlags = 0;
  103. tp->pauseMicroSecs = 50000;
  104. tp->waitMicroSecs = 1000000;
  105. if((sysErr = pthread_create(&tp->pthreadH,NULL,_cmThThreadCallback, (void*)tp )) != 0 )
  106. {
  107. tp->state = kNotInitThId;
  108. rc = _cmThError(&tp->err,kCreateFailThRC,sysErr,"Thread create failed.");
  109. }
  110. hPtr->h = tp;
  111. return rc;
  112. }
  113. cmThRC_t cmThreadDestroy( cmThreadH_t* hPtr )
  114. {
  115. cmThRC_t rc = kOkThRC;
  116. if( hPtr==NULL || cmThreadIsValid(*hPtr)==false )
  117. return rc;
  118. cmThThread_t* t = _cmThThreadFromHandle(*hPtr );
  119. if( t == NULL )
  120. return kInvalidHandleThRC;
  121. // tell the thread to exit
  122. t->doFlags = cmSetFlag(t->doFlags,kDoExitThFl);
  123. // wait for the thread to exit and then deallocate the thread object
  124. if((rc = _cmThWaitForState(t,kExitedThId)) == kOkThRC )
  125. {
  126. cmMemFree(t);
  127. hPtr->h = NULL;
  128. }
  129. else
  130. {
  131. rc = _cmThError(&t->err,rc,0,"Thread timed out waiting for destroy.");
  132. }
  133. return rc;
  134. }
  135. cmThRC_t cmThreadPause( cmThreadH_t h, unsigned cmdFlags )
  136. {
  137. cmThRC_t rc = kOkThRC;
  138. bool pauseFl = cmIsFlag(cmdFlags,kPauseThFl);
  139. bool waitFl = cmIsFlag(cmdFlags,kWaitThFl);
  140. cmThThread_t* t = _cmThThreadFromHandle(h);
  141. unsigned waitId;
  142. if( t == NULL )
  143. return kInvalidHandleThRC;
  144. bool isPausedFl = t->state == kPausedThId;
  145. if( isPausedFl == pauseFl )
  146. return kOkThRC;
  147. if( pauseFl )
  148. {
  149. t->doFlags = cmSetFlag(t->doFlags,kDoPauseThFl);
  150. waitId = kPausedThId;
  151. }
  152. else
  153. {
  154. t->doFlags = cmSetFlag(t->doFlags,kDoRunThFl);
  155. waitId = kRunningThId;
  156. }
  157. if( waitFl )
  158. rc = _cmThWaitForState(t,waitId);
  159. if( rc != kOkThRC )
  160. _cmThError(&t->err,rc,0,"Thread timed out waiting for '%s'.", pauseFl ? "pause" : "un-pause");
  161. return rc;
  162. }
  163. cmThStateId_t cmThreadState( cmThreadH_t h )
  164. {
  165. cmThThread_t* tp = _cmThThreadFromHandle(h);
  166. if( tp == NULL )
  167. return kNotInitThId;
  168. return tp->state;
  169. }
  170. bool cmThreadIsValid( cmThreadH_t h )
  171. { return h.h != NULL; }
  172. unsigned cmThreadPauseTimeOutMicros( cmThreadH_t h )
  173. {
  174. cmThThread_t* tp = _cmThThreadFromHandle(h);
  175. return tp->pauseMicroSecs;
  176. }
  177. void cmThreadSetPauseTimeOutMicros( cmThreadH_t h, unsigned usecs )
  178. {
  179. cmThThread_t* tp = _cmThThreadFromHandle(h);
  180. tp->pauseMicroSecs = usecs;
  181. }
  182. unsigned cmThreadWaitTimeOutMicros( cmThreadH_t h )
  183. {
  184. cmThThread_t* tp = _cmThThreadFromHandle(h);
  185. return tp->waitMicroSecs;
  186. }
  187. void cmThreadSetWaitTimeOutMicros( cmThreadH_t h, unsigned usecs )
  188. {
  189. cmThThread_t* tp = _cmThThreadFromHandle(h);
  190. tp->waitMicroSecs = usecs;
  191. }
  192. bool _cmThreadTestCb( void* p )
  193. {
  194. unsigned* ip = (unsigned*)p;
  195. ip[0]++;
  196. return true;
  197. }
  198. void cmThreadTest(cmRpt_t* rpt)
  199. {
  200. cmThreadH_t th0;
  201. unsigned val = 0;
  202. if( cmThreadCreate(&th0,_cmThreadTestCb,&val,rpt) == kOkThRC )
  203. {
  204. if( cmThreadPause(th0,0) != kOkThRC )
  205. {
  206. cmRptPrintf(rpt,"Thread start failed.\n");
  207. return;
  208. }
  209. char c = 0;
  210. cmRptPrintf(rpt,"o=print p=pause s=state q=quit\n");
  211. while( c != 'q' )
  212. {
  213. c = (char)fgetc(stdin);
  214. fflush(stdin);
  215. switch(c)
  216. {
  217. case 'o':
  218. cmRptPrintf(rpt,"val: 0x%x\n",val);
  219. break;
  220. case 's':
  221. cmRptPrintf(rpt,"state=%i\n",cmThreadState(th0));
  222. break;
  223. case 'p':
  224. {
  225. cmRC_t rc;
  226. if( cmThreadState(th0) == kPausedThId )
  227. rc = cmThreadPause(th0,kWaitThFl);
  228. else
  229. rc = cmThreadPause(th0,kPauseThFl|kWaitThFl);
  230. if( rc == kOkThRC )
  231. cmRptPrintf(rpt,"new state:%i\n", cmThreadState(th0));
  232. else
  233. cmRptPrintf(rpt,"cmThreadPause() failed.");
  234. }
  235. break;
  236. case 'q':
  237. break;
  238. //default:
  239. //cmRptPrintf(rpt,"Unknown:%c\n",c);
  240. }
  241. }
  242. if( cmThreadDestroy(&th0) != kOkThRC )
  243. cmRptPrintf(rpt,"Thread destroy failed.\n");
  244. }
  245. }
  246. //-----------------------------------------------------------------------------
  247. //-----------------------------------------------------------------------------
  248. //-----------------------------------------------------------------------------
  249. typedef struct
  250. {
  251. cmErr_t err;
  252. pthread_mutex_t mutex;
  253. pthread_cond_t cvar;
  254. } cmThreadMutex_t;
  255. cmThreadMutexH_t kThreadMutexNULL = {NULL};
  256. cmThreadMutex_t* _cmThreadMutexFromHandle( cmThreadMutexH_t h )
  257. {
  258. cmThreadMutex_t* p = (cmThreadMutex_t*)h.h;
  259. assert(p != NULL);
  260. return p;
  261. }
  262. cmThRC_t cmThreadMutexCreate( cmThreadMutexH_t* hPtr, cmRpt_t* rpt )
  263. {
  264. int sysErr;
  265. cmThreadMutex_t* p = cmMemAllocZ( cmThreadMutex_t, 1 );
  266. cmErrSetup(&p->err,rpt,"Thread Mutex");
  267. if((sysErr = pthread_mutex_init(&p->mutex,NULL)) != 0 )
  268. return _cmThError(&p->err,kCreateFailThRC,sysErr,"Thread mutex create failed.");
  269. if((sysErr = pthread_cond_init(&p->cvar,NULL)) != 0 )
  270. return _cmThError(&p->err,kCreateFailThRC,sysErr,"Thread Condition var. create failed.");
  271. hPtr->h = p;
  272. return kOkThRC;
  273. }
  274. cmThRC_t cmThreadMutexDestroy( cmThreadMutexH_t* hPtr )
  275. {
  276. int sysErr;
  277. cmThreadMutex_t* p = _cmThreadMutexFromHandle(*hPtr);
  278. if( p == NULL )
  279. return kInvalidHandleThRC;
  280. if((sysErr = pthread_cond_destroy(&p->cvar)) != 0)
  281. return _cmThError(&p->err,kDestroyFailThRC,sysErr,"Thread condition var. destroy failed.");
  282. if((sysErr = pthread_mutex_destroy(&p->mutex)) != 0)
  283. return _cmThError(&p->err,kDestroyFailThRC,sysErr,"Thread mutex destroy failed.");
  284. cmMemFree(p);
  285. hPtr->h = NULL;
  286. return kOkThRC;
  287. }
  288. cmThRC_t cmThreadMutexTryLock( cmThreadMutexH_t h, bool* lockFlPtr )
  289. {
  290. cmThreadMutex_t* p = _cmThreadMutexFromHandle(h);
  291. if( p == NULL )
  292. return kInvalidHandleThRC;
  293. int sysErr = pthread_mutex_trylock(&p->mutex);
  294. switch(sysErr)
  295. {
  296. case EBUSY:
  297. *lockFlPtr = false;
  298. break;
  299. case 0:
  300. *lockFlPtr = true;
  301. break;
  302. default:
  303. return _cmThError(&p->err,kLockFailThRC,sysErr,"Thread mutex try-lock failed.");;
  304. }
  305. return kOkThRC;
  306. }
  307. cmThRC_t cmThreadMutexLock( cmThreadMutexH_t h )
  308. {
  309. cmThreadMutex_t* p = _cmThreadMutexFromHandle(h);
  310. if( p == NULL )
  311. return kInvalidHandleThRC;
  312. int sysErr = pthread_mutex_lock(&p->mutex);
  313. if( sysErr == 0 )
  314. return kOkThRC;
  315. return _cmThError(&p->err,kLockFailThRC,sysErr,"Thread mutex lock failed.");
  316. }
  317. cmThRC_t cmThreadMutexUnlock( cmThreadMutexH_t h )
  318. {
  319. cmThreadMutex_t* p = _cmThreadMutexFromHandle(h);
  320. if( p == NULL )
  321. return kInvalidHandleThRC;
  322. int sysErr = pthread_mutex_unlock(&p->mutex);
  323. if( sysErr == 0 )
  324. return kOkThRC;
  325. return _cmThError(&p->err,kUnlockFailThRC,sysErr,"Thread mutex unlock failed.");
  326. }
  327. bool cmThreadMutexIsValid( cmThreadMutexH_t h )
  328. { return h.h != NULL; }
  329. cmThRC_t cmThreadMutexWaitOnCondVar( cmThreadMutexH_t h, bool lockFl )
  330. {
  331. cmThreadMutex_t* p = _cmThreadMutexFromHandle(h);
  332. if( p == NULL )
  333. return kInvalidHandleThRC;
  334. int sysErr;
  335. if( lockFl )
  336. if( (sysErr=pthread_mutex_lock(&p->mutex)) != 0 )
  337. _cmThError(&p->err,kLockFailThRC,sysErr,"Thread lock failed on cond. var. wait.");
  338. if((sysErr = pthread_cond_wait(&p->cvar,&p->mutex)) != 0 )
  339. _cmThError(&p->err,kCVarWaitFailThRC,sysErr,"Thread cond. var. wait failed.");
  340. return kOkThRC;
  341. }
  342. cmThRC_t cmThreadMutexSignalCondVar( cmThreadMutexH_t h )
  343. {
  344. int sysErr;
  345. cmThreadMutex_t* p = _cmThreadMutexFromHandle(h);
  346. if( p == NULL )
  347. return kInvalidHandleThRC;
  348. if((sysErr = pthread_cond_signal(&p->cvar)) != 0 )
  349. return _cmThError(&p->err,kCVarSignalFailThRC,sysErr,"Thread cond. var. signal failed.");
  350. return kOkThRC;
  351. }
  352. //-----------------------------------------------------------------------------
  353. //-----------------------------------------------------------------------------
  354. //-----------------------------------------------------------------------------
  355. cmTsQueueH_t cmTsQueueNullHandle = { NULL };
  356. enum { cmTsQueueBufCnt = 2 };
  357. typedef struct
  358. {
  359. unsigned allocCnt; // count of bytes allocated for the buffer
  360. unsigned fullCnt; // count of bytes used in the buffer
  361. char* basePtr; // base of buffer memory
  362. unsigned* msgPtr; // pointer to first msg
  363. unsigned msgCnt;
  364. } cmTsQueueBuf;
  365. typedef struct
  366. {
  367. cmThreadMutexH_t mutexH;
  368. cmTsQueueBuf bufArray[cmTsQueueBufCnt];
  369. unsigned inBufIdx;
  370. unsigned outBufIdx;
  371. char* memPtr;
  372. cmTsQueueCb_t cbFunc;
  373. void* userCbPtr;
  374. } cmTsQueue_t;
  375. cmTsQueue_t* _cmTsQueueFromHandle( cmTsQueueH_t h )
  376. {
  377. cmTsQueue_t* p = h.h;
  378. assert(p != NULL);
  379. return p;
  380. }
  381. cmThRC_t _cmTsQueueDestroy( cmTsQueue_t* p )
  382. {
  383. cmThRC_t rc;
  384. if( p == NULL )
  385. return kInvalidHandleThRC;
  386. if( p->mutexH.h != NULL )
  387. if((rc = cmThreadMutexDestroy(&p->mutexH)) != kOkThRC )
  388. return rc;
  389. if( p->memPtr != NULL )
  390. cmMemPtrFree(&p->memPtr);
  391. cmMemPtrFree(&p);
  392. return kOkThRC;
  393. }
  394. cmThRC_t cmTsQueueCreate( cmTsQueueH_t* hPtr, unsigned bufByteCnt, cmTsQueueCb_t cbFunc, void* userCbPtr, cmRpt_t* rpt )
  395. {
  396. cmTsQueue_t* p = cmMemAllocZ( cmTsQueue_t, 1 );
  397. unsigned i;
  398. if( cmThreadMutexCreate(&p->mutexH,rpt) != kOkThRC )
  399. goto errLabel;
  400. p->memPtr = cmMemAllocZ( char, bufByteCnt*cmTsQueueBufCnt );
  401. p->outBufIdx = 0;
  402. p->inBufIdx = 1;
  403. p->cbFunc = cbFunc;
  404. p->userCbPtr = userCbPtr;
  405. for(i=0; i<cmTsQueueBufCnt; ++i)
  406. {
  407. p->bufArray[i].allocCnt = bufByteCnt;
  408. p->bufArray[i].fullCnt = 0;
  409. p->bufArray[i].basePtr = p->memPtr + (i*bufByteCnt);
  410. p->bufArray[i].msgPtr = NULL;
  411. p->bufArray[i].msgCnt = 0;
  412. }
  413. hPtr->h = p;
  414. return kOkThRC;
  415. errLabel:
  416. _cmTsQueueDestroy(p);
  417. return kCreateFailThRC;
  418. }
  419. cmThRC_t cmTsQueueDestroy( cmTsQueueH_t* hPtr )
  420. {
  421. cmThRC_t rc = kOkThRC;
  422. if( (hPtr != NULL) && cmTsQueueIsValid(*hPtr))
  423. if((rc = _cmTsQueueDestroy(_cmTsQueueFromHandle(*hPtr))) == kOkThRC )
  424. hPtr->h = NULL;
  425. return rc;
  426. }
  427. cmThRC_t cmTsQueueSetCallback( cmTsQueueH_t h, cmTsQueueCb_t cbFunc, void* cbArg )
  428. {
  429. cmTsQueue_t* p = _cmTsQueueFromHandle(h);
  430. p->cbFunc = cbFunc;
  431. p->userCbPtr = cbArg;
  432. return kOkThRC;
  433. }
  434. unsigned cmTsQueueAllocByteCount( cmTsQueueH_t h )
  435. {
  436. cmTsQueue_t* p = _cmTsQueueFromHandle(h);
  437. unsigned n = 0;
  438. if( cmThreadMutexLock(p->mutexH) == kOkThRC )
  439. {
  440. n = p->bufArray[ p->inBufIdx ].allocCnt;
  441. cmThreadMutexUnlock(p->mutexH);
  442. }
  443. return n;
  444. }
  445. unsigned cmTsQueueAvailByteCount( cmTsQueueH_t h )
  446. {
  447. cmTsQueue_t* p = _cmTsQueueFromHandle(h);
  448. unsigned n = 0;
  449. if(cmThreadMutexLock(p->mutexH) == kOkThRC )
  450. {
  451. n = p->bufArray[ p->inBufIdx ].allocCnt - p->bufArray[ p->inBufIdx].fullCnt;
  452. cmThreadMutexUnlock(p->mutexH);
  453. }
  454. return n;
  455. }
  456. cmThRC_t _cmTsQueueEnqueueMsg( cmTsQueueH_t h, const void* msgPtrArray[], unsigned msgByteCntArray[], unsigned arrayCnt )
  457. {
  458. cmThRC_t rc;
  459. cmTsQueue_t* p = _cmTsQueueFromHandle(h);
  460. if( p == NULL )
  461. return kInvalidHandleThRC;
  462. // lock the mutex
  463. if((rc = cmThreadMutexLock(p->mutexH)) == kOkThRC )
  464. {
  465. cmTsQueueBuf* b = p->bufArray + p->inBufIdx; // ptr to buf recd
  466. const char* ep = b->basePtr + b->allocCnt; // end of buf data space
  467. unsigned *mp = (unsigned*)(b->basePtr + b->fullCnt); // ptr to size of new msg space
  468. char* dp = (char*)(mp+1); // ptr to data area of new msg space
  469. unsigned ttlByteCnt = 0; // track size of msg data
  470. unsigned i = 0;
  471. // get the total size of the msg
  472. for(i=0; i<arrayCnt; ++i)
  473. ttlByteCnt += msgByteCntArray[i];
  474. // if the msg is too big for the queue buf
  475. if( dp + ttlByteCnt > ep )
  476. rc = kBufFullThRC;
  477. else
  478. {
  479. // for each segment of the incoming msg
  480. for(i=0; i<arrayCnt; ++i)
  481. {
  482. // get the size of the segment
  483. unsigned n = msgByteCntArray[i];
  484. // copy in the segment
  485. memcpy(dp,msgPtrArray[i],n);
  486. dp += n; //
  487. }
  488. assert(dp <= ep );
  489. // write the size ofthe msg into the buffer
  490. *mp = ttlByteCnt;
  491. // update the pointer to the first msg
  492. if( b->msgPtr == NULL )
  493. b->msgPtr = mp;
  494. // track the count of msgs in this buffer
  495. ++b->msgCnt;
  496. // update fullCnt last since dequeue uses fullCnt to
  497. // notice that a msg may be waiting
  498. b->fullCnt += sizeof(unsigned) + ttlByteCnt;
  499. }
  500. cmThreadMutexUnlock(p->mutexH);
  501. }
  502. return rc;
  503. }
  504. cmThRC_t cmTsQueueEnqueueSegMsg( cmTsQueueH_t h, const void* msgPtrArray[], unsigned msgByteCntArray[], unsigned arrayCnt )
  505. { return _cmTsQueueEnqueueMsg(h,msgPtrArray,msgByteCntArray,arrayCnt); }
  506. cmThRC_t cmTsQueueEnqueueMsg( cmTsQueueH_t h, const void* dataPtr, unsigned byteCnt )
  507. {
  508. const void* msgPtrArray[] = { dataPtr };
  509. unsigned msgByteCntArray[] = { byteCnt };
  510. return _cmTsQueueEnqueueMsg(h,msgPtrArray,msgByteCntArray,1);
  511. }
  512. cmThRC_t cmTsQueueEnqueueIdMsg( cmTsQueueH_t h, unsigned id, const void* dataPtr, unsigned byteCnt )
  513. {
  514. const void* msgPtrArray[] = { &id, dataPtr };
  515. unsigned msgByteCntArray[] = { sizeof(id), byteCnt };
  516. return _cmTsQueueEnqueueMsg(h,msgPtrArray,msgByteCntArray,2);
  517. }
  518. cmThRC_t _cmTsQueueDequeueMsg( cmTsQueue_t* p, void* retBuf, unsigned refBufByteCnt )
  519. {
  520. cmTsQueueBuf* b = p->bufArray + p->outBufIdx;
  521. // if the output buffer is empty - there is nothing to do
  522. if( b->fullCnt == 0 )
  523. return kBufEmptyThRC;
  524. assert( b->msgPtr != NULL );
  525. // get the output msg size and data
  526. unsigned msgByteCnt = *b->msgPtr;
  527. char* msgDataPtr = (char*)(b->msgPtr + 1);
  528. // transmit the msg via a callback
  529. if( retBuf == NULL && p->cbFunc != NULL )
  530. p->cbFunc(p->userCbPtr,msgByteCnt,msgDataPtr);
  531. else
  532. {
  533. // retBuf may be NULL if the func is being used by cmTsQueueDequeueByteCount()
  534. if( retBuf == NULL || msgByteCnt > refBufByteCnt )
  535. return kBufTooSmallThRC;
  536. // copy the msg to a buffer
  537. if( retBuf != NULL )
  538. memcpy(retBuf,msgDataPtr,msgByteCnt);
  539. }
  540. // update the buffer
  541. b->fullCnt -= sizeof(unsigned) + msgByteCnt;
  542. b->msgPtr = (unsigned*)(msgDataPtr + msgByteCnt);
  543. --(b->msgCnt);
  544. if( b->fullCnt == 0 )
  545. {
  546. assert(b->msgCnt == 0);
  547. b->msgPtr = NULL;
  548. }
  549. return kOkThRC;
  550. }
  551. cmThRC_t cmTsQueueDequeueMsg( cmTsQueueH_t h, void* retBuf, unsigned refBufByteCnt )
  552. {
  553. cmThRC_t rc;
  554. cmTsQueue_t* p = _cmTsQueueFromHandle(h);
  555. if( p == NULL )
  556. return kInvalidHandleThRC;
  557. // dequeue the next msg from the current output buffer
  558. if((rc =_cmTsQueueDequeueMsg( p, retBuf, refBufByteCnt )) != kBufEmptyThRC )
  559. return rc;
  560. // the current output buffer was empty
  561. cmTsQueueBuf* b = p->bufArray + p->inBufIdx;
  562. // if the input buffer has msg's ...
  563. if( b->fullCnt > 0 )
  564. {
  565. bool lockFl = false;
  566. // ...attempt to lock the mutex ...
  567. if( (cmThreadMutexTryLock(p->mutexH,&lockFl) == kOkThRC) && lockFl )
  568. {
  569. // ... swap the input and the output buffers ...
  570. unsigned tmp = p->inBufIdx;
  571. p->inBufIdx = p->outBufIdx;
  572. p->outBufIdx = tmp;
  573. // .. unlock the mutex
  574. cmThreadMutexUnlock(p->mutexH);
  575. // ... and dequeue the first msg from the new output buffer
  576. rc = _cmTsQueueDequeueMsg( p, retBuf, refBufByteCnt );
  577. }
  578. }
  579. return rc;
  580. }
  581. bool cmTsQueueMsgWaiting( cmTsQueueH_t h )
  582. {
  583. cmTsQueue_t* p = _cmTsQueueFromHandle(h);
  584. if( p == NULL )
  585. return false;
  586. if( p->bufArray[p->outBufIdx].fullCnt )
  587. return true;
  588. return p->bufArray[p->inBufIdx].fullCnt > 0;
  589. }
  590. unsigned cmTsQueueDequeueMsgByteCount( cmTsQueueH_t h )
  591. {
  592. cmTsQueue_t* p = _cmTsQueueFromHandle(h);
  593. if( p == NULL )
  594. return 0;
  595. // if output msgs are available then the msgPtr points to the size of the msg
  596. if( p->bufArray[p->outBufIdx].fullCnt )
  597. return *(p->bufArray[p->outBufIdx].msgPtr);
  598. // no msgs are waiting in the output buffer
  599. // force the buffers to swap - returns kBufEmptyThRC if there are
  600. // still no msgs waiting after the swap (the input buf was also empty)
  601. if( cmTsQueueDequeueMsg(h,NULL,0) == kBufTooSmallThRC )
  602. {
  603. // the buffers swapped so there must be msg waiting
  604. assert( p->bufArray[p->outBufIdx].fullCnt );
  605. return *(p->bufArray[p->outBufIdx].msgPtr);
  606. }
  607. return 0;
  608. }
  609. bool cmTsQueueIsValid( cmTsQueueH_t h )
  610. { return h.h != NULL; }
  611. //--------------------------------------------------------------------------------------------------
  612. //--------------------------------------------------------------------------------------------------
  613. //--------------------------------------------------------------------------------------------------
  614. #ifdef NOT_DEF
  615. enum { kThBufCnt=2 };
  616. typedef struct
  617. {
  618. char* buf;
  619. volatile unsigned ii;
  620. volatile unsigned oi;
  621. } cmThBuf_t;
  622. typedef struct
  623. {
  624. cmErr_t err;
  625. cmThBuf_t a[kThBufCnt];
  626. volatile unsigned ibi;
  627. unsigned bn;
  628. cmTsQueueCb_t cbFunc;
  629. void* cbArg;
  630. } cmTs1p1c_t;
  631. cmTs1p1c_t* _cmTs1p1cHandleToPtr( cmTs1p1cH_t h )
  632. {
  633. cmTs1p1c_t* p = (cmTs1p1c_t*)h.h;
  634. assert( p != NULL );
  635. return p;
  636. }
  637. cmThRC_t _cmTs1p1cDestroy( cmTs1p1c_t* p )
  638. {
  639. unsigned i;
  640. for(i=0; i<kThBufCnt; ++i)
  641. cmMemFree(p->a[i].buf);
  642. cmMemFree(p);
  643. return kOkThRC;
  644. }
  645. cmThRC_t cmTs1p1cCreate( cmTs1p1cH_t* hPtr, unsigned bufByteCnt, cmTsQueueCb_t cbFunc, void* cbArg, cmRpt_t* rpt )
  646. {
  647. cmThRC_t rc;
  648. if((rc = cmTs1p1cDestroy(hPtr)) != kOkThRC )
  649. return rc;
  650. unsigned i;
  651. cmTs1p1c_t* p = cmMemAllocZ(cmTs1p1c_t,1);
  652. cmErrSetup(&p->err,rpt,"TS 1p1c Queue");
  653. for(i=0; i<kThBufCnt; ++i)
  654. {
  655. p->a[i].buf = cmMemAllocZ(char,bufByteCnt);
  656. p->a[i].ii = 0;
  657. p->a[i].oi = bufByteCnt;
  658. }
  659. p->ibi = 0;
  660. p->bn = bufByteCnt;
  661. p->cbFunc = cbFunc;
  662. p->cbArg = cbArg;
  663. hPtr->h = p;
  664. return rc;
  665. }
  666. cmThRC_t cmTs1p1cDestroy( cmTs1p1cH_t* hp )
  667. {
  668. cmThRC_t rc = kOkThRC;
  669. if( hp == NULL || cmTs1p1cIsValid(*hp)==false )
  670. return kOkThRC;
  671. cmTs1p1c_t* p = _cmTs1p1cHandleToPtr(*hp);
  672. if(( rc = _cmTs1p1cDestroy(p)) != kOkThRC )
  673. return rc;
  674. hp->h = NULL;
  675. return rc;
  676. }
  677. cmThRC_t cmTs1p1cEnqueueSegMsg( cmTs1p1cH_t h, const void* msgPtrArray[], unsigned msgByteCntArray[], unsigned arrayCnt )
  678. {
  679. cmThRC_t rc = kOkThRC;
  680. unsigned mn = 0;
  681. unsigned i;
  682. cmTs1p1c_t* p = _cmTs1p1cHandleToPtr(h);
  683. cmThBuf_t* ib = p->a + p->ibi;
  684. // get the total count of bytes for this msg
  685. for(i=0; i<arrayCnt; ++i)
  686. mn += msgByteCntArray[i];
  687. unsigned dn = mn + sizeof(unsigned);
  688. // if the message is too big for even an empty buffer
  689. if( dn > p->bn )
  690. return cmErrMsg(&p->err,kBufFullThRC,"A msg containing %i bytes will never be able to fit in a queue with an empty size of %i bytes.",dn,p->bn);
  691. // if the msg won't fit in the current input buffer then try swapping buffers.
  692. if( ib->ii + dn > p->bn )
  693. {
  694. // get the current output buffer
  695. cmThBuf_t* ob = p->a + (p->ibi==0 ? 1 : 0);
  696. // Empty buffers will be set such that: oi==bn and ii==0.
  697. //
  698. // Note that setting ii to 0 in an output buffer is the last operation
  699. // performed on an empty output buffer. ii==0 is therefore the
  700. // signal that an output buffer can be reused for input.
  701. // if the output buffer is not empty - then an overflow occurred
  702. if( ob->ii != 0 )
  703. return cmErrMsg(&p->err,kBufFullThRC,"The msq queue cannot accept a %i byte msg into %i bytes.",dn, p->bn - ib->ii);
  704. // setup the initial output location of the new output buffer
  705. ib->oi = 0;
  706. // swap buffers
  707. p->ibi = (p->ibi + 1) % kThBufCnt;
  708. // get the new input buffer
  709. ib = ob;
  710. }
  711. // get a pointer to the base of the write location
  712. char* dp = ib->buf + ib->ii;
  713. // write the length of the message
  714. *(unsigned*)dp = mn;
  715. dp += sizeof(unsigned);
  716. // write the body of the message
  717. for(i=0; i<arrayCnt; ++i)
  718. {
  719. memcpy(dp,msgPtrArray[i],msgByteCntArray[i]);
  720. dp += msgByteCntArray[i];
  721. }
  722. // this MUST be executed last - we'll use 'dp' in the calculation
  723. // (even though ib->ii += dn would be more straight forward way
  724. // to accomplish the same thing) to prevent the optimizer from
  725. // moving the assignment prior to the for loop.
  726. ib->ii += dp - (ib->buf + ib->ii);
  727. return rc;
  728. }
  729. cmThRC_t cmTs1p1cEnqueueMsg( cmTsQueueH_t h, const void* dataPtr, unsigned byteCnt )
  730. { return cmTs1p1cEnqueueSegMsg(h,&dataPtr,&byteCnt,1); }
  731. unsigned cmTs1p1cAllocByteCount( cmTs1p1cH_t h )
  732. {
  733. cmTs1p1c_t* p = _cmTs1p1cHandleToPtr(h);
  734. return p->bn;
  735. }
  736. unsigned cmTs1p1cAvailByteCount( cmTs1p1cH_t h )
  737. {
  738. cmTs1p1c_t* p = _cmTs1p1cHandleToPtr(h);
  739. return p->bn - p->a[ p->ibi ].ii;
  740. }
  741. cmThRC_t cmTs1p1cDequeueMsg( cmTs1p1cH_t h, void* dataPtr, unsigned byteCnt )
  742. {
  743. cmThRC_t rc = kOkThRC;
  744. cmTs1p1c_t* p = _cmTs1p1cHandleToPtr(h);
  745. cmThBuf_t* ob = p->a + (p->ibi == 0 ? 1 : 0);
  746. // empty buffers always are set to: oi==bn && ii==0
  747. if( ob->oi >= ob->ii )
  748. return kBufEmptyThRC;
  749. // get the size of the msg
  750. unsigned mn = *(unsigned*)(ob->buf + ob->oi);
  751. // increment the current output location to the msg body
  752. ob->oi += sizeof(unsigned);
  753. // copy or send the msg
  754. if( dataPtr != NULL )
  755. {
  756. if( byteCnt < mn )
  757. return cmErrMsg(&p->err,kBufTooSmallThRC,"The return buffer constains too few bytes (%i) to contain %i bytes.",byteCnt,mn);
  758. memcpy(dataPtr, ob->buf + ob->oi, mn);
  759. }
  760. else
  761. {
  762. p->cbFunc(p->cbArg, mn, ob->buf + ob->oi );
  763. }
  764. ob->oi += mn;
  765. // if we are reading correctly ob->oi should land
  766. // exactly on ob->ii when the buffer is empty
  767. assert( ob->oi <= ob->ii );
  768. // if the buffer is empty
  769. if( ob->oi == ob->ii )
  770. {
  771. ob->oi = p->bn; // mark the buffer as empty
  772. ob->ii = 0; //
  773. }
  774. return rc;
  775. }
  776. unsigned cmTs1p1cDequeueMsgByteCount( cmTsQueueH_t h )
  777. {
  778. cmTs1p1c_t* p = _cmTs1p1cHandleToPtr(h);
  779. cmThBuf_t* ob = p->a + (p->ibi == 0 ? 1 : 0);
  780. // empty buffers always are set to: oi==bn && ii==0
  781. if( ob->oi >= ob->ii )
  782. return 0;
  783. // get the size of the msg
  784. return *(unsigned*)(ob->buf + ob->oi);
  785. }
  786. bool cmTs1p1cMsgWaiting( cmTsQueueH_t h )
  787. { return cmTs1p1cDequeueMsgByteCount(h) > 0; }
  788. bool cmTs1p1cIsValid( cmTs1p1cH_t h )
  789. { return h.h != NULL; }
  790. #endif
  791. //--------------------------------------------------------------------------------------------------
  792. //--------------------------------------------------------------------------------------------------
  793. //--------------------------------------------------------------------------------------------------
  794. typedef struct
  795. {
  796. volatile unsigned ii;
  797. cmErr_t err;
  798. char* buf;
  799. unsigned bn;
  800. cmTsQueueCb_t cbFunc;
  801. void* cbArg;
  802. volatile unsigned oi;
  803. } cmTs1p1c_t;
  804. cmTs1p1c_t* _cmTs1p1cHandleToPtr( cmTs1p1cH_t h )
  805. {
  806. cmTs1p1c_t* p = (cmTs1p1c_t*)h.h;
  807. assert( p != NULL );
  808. return p;
  809. }
  810. cmThRC_t cmTs1p1cCreate( cmTs1p1cH_t* hPtr, unsigned bufByteCnt, cmTsQueueCb_t cbFunc, void* cbArg, cmRpt_t* rpt )
  811. {
  812. cmThRC_t rc;
  813. if((rc = cmTs1p1cDestroy(hPtr)) != kOkThRC )
  814. return rc;
  815. cmTs1p1c_t* p = cmMemAllocZ(cmTs1p1c_t,1);
  816. cmErrSetup(&p->err,rpt,"1p1c Queue");
  817. p->buf = cmMemAllocZ(char,bufByteCnt+sizeof(unsigned));
  818. p->ii = 0;
  819. p->oi = 0;
  820. p->bn = bufByteCnt;
  821. p->cbFunc = cbFunc;
  822. p->cbArg = cbArg;
  823. hPtr->h = p;
  824. return rc;
  825. }
  826. cmThRC_t cmTs1p1cDestroy( cmTs1p1cH_t* hp )
  827. {
  828. cmThRC_t rc = kOkThRC;
  829. if( hp == NULL || cmTs1p1cIsValid(*hp)==false )
  830. return kOkThRC;
  831. cmTs1p1c_t* p = _cmTs1p1cHandleToPtr(*hp);
  832. cmMemFree(p->buf);
  833. cmMemFree(p);
  834. hp->h = NULL;
  835. return rc;
  836. }
  837. cmThRC_t cmTs1p1cEnqueueSegMsg( cmTs1p1cH_t h, const void* msgPtrArray[], unsigned msgByteCntArray[], unsigned arrayCnt )
  838. {
  839. cmThRC_t rc = kOkThRC;
  840. unsigned mn = 0;
  841. unsigned i;
  842. cmTs1p1c_t* p = _cmTs1p1cHandleToPtr(h);
  843. // get the total count of bytes for this msg
  844. for(i=0; i<arrayCnt; ++i)
  845. mn += msgByteCntArray[i];
  846. int dn = mn + sizeof(unsigned);
  847. int oi = p->oi;
  848. int bi = p->ii; // 'bi' is the idx of the leftmost cell which can be written
  849. int en = p->bn; // 'en' is the idx of the cell just to the right of the rightmost cell that can be written
  850. // note: If 'oi' marks the rightmost location then 'en' must be set
  851. // one cell to the left of 'oi', because 'ii' can never be allowed to
  852. // advance onto 'oi' - because 'oi'=='ii' marks an empty (NOT a full)
  853. // queue.
  854. //
  855. // If 'bn' marks the rightmost location then 'ii' can advance onto 'bn'
  856. // beause the true queue length is bn+1.
  857. // if we need to wrap
  858. if( en-bi < dn && oi<=bi )
  859. {
  860. bi = 0;
  861. en = oi - 1; // note if oi==0 then en is negative - see note above re: oi==ii
  862. assert( p->ii>=0 && p->ii <= p->bn );
  863. *(unsigned*)(p->buf + p->ii) = cmInvalidIdx; // mark the wrap location
  864. }
  865. // if oi is between ii and bn
  866. if( oi > bi )
  867. en = oi - 1; // never allow ii to advance onto oi - see note above
  868. // if the msg won't fit
  869. if( en - bi < dn )
  870. return cmErrMsg(&p->err,kBufFullThRC,"%i consecutive bytes is not available in the queue.",dn);
  871. // set the msg byte count - the msg byte cnt precedes the msg body
  872. char* dp = p->buf + bi;
  873. *(unsigned*)dp = dn - sizeof(unsigned);
  874. dp += sizeof(unsigned);
  875. // copy the msg into the buffer
  876. for(i=0,dn=0; i<arrayCnt; ++i)
  877. {
  878. memcpy(dp,msgPtrArray[i],msgByteCntArray[i]);
  879. dp += msgByteCntArray[i];
  880. dn += msgByteCntArray[i];
  881. }
  882. // incrementing p->ii must occur last - the unnecessary accumulation
  883. // of dn in the above loop is intended to prevent this line from
  884. // begin moved before the copy loop.
  885. p->ii = bi + dn + sizeof(unsigned);
  886. assert( p->ii >= 0 && p->ii <= p->bn);
  887. return rc;
  888. }
  889. cmThRC_t cmTs1p1cEnqueueMsg( cmTs1p1cH_t h, const void* dataPtr, unsigned byteCnt )
  890. { return cmTs1p1cEnqueueSegMsg(h,&dataPtr,&byteCnt,1); }
  891. unsigned cmTs1p1cAllocByteCount( cmTs1p1cH_t h )
  892. {
  893. cmTs1p1c_t* p = _cmTs1p1cHandleToPtr(h);
  894. return p->bn;
  895. }
  896. unsigned cmTs1p1cAvailByteCount( cmTs1p1cH_t h )
  897. {
  898. cmTs1p1c_t* p = _cmTs1p1cHandleToPtr(h);
  899. unsigned oi = p->oi;
  900. unsigned ii = p->ii;
  901. return oi < ii ? p->bn - ii + oi : oi - ii;
  902. }
  903. unsigned _cmTs1p1cDequeueMsgByteCount( cmTs1p1c_t* p )
  904. {
  905. // if the buffer is empty
  906. if( p->ii == p->oi )
  907. return 0;
  908. // get the length of the next msg
  909. unsigned mn = *(unsigned*)(p->buf + p->oi);
  910. // if the msg length is cmInvalidIdx ...
  911. if( mn == cmInvalidIdx )
  912. {
  913. p->oi = 0; // ... wrap to buf begin and try again
  914. return _cmTs1p1cDequeueMsgByteCount(p);
  915. }
  916. return mn;
  917. }
  918. cmThRC_t cmTs1p1cDequeueMsg( cmTs1p1cH_t h, void* dataPtr, unsigned byteCnt )
  919. {
  920. cmThRC_t rc = kOkThRC;
  921. cmTs1p1c_t* p = _cmTs1p1cHandleToPtr(h);
  922. unsigned mn;
  923. if((mn = _cmTs1p1cDequeueMsgByteCount(p)) == 0 )
  924. return kBufEmptyThRC;
  925. void* mp = p->buf + p->oi + sizeof(unsigned);
  926. if( dataPtr != NULL )
  927. {
  928. if( byteCnt < mn )
  929. return cmErrMsg(&p->err,kBufTooSmallThRC,"The return buffer constains too few bytes (%i) to contain %i bytes.",byteCnt,mn);
  930. memcpy(dataPtr,mp,mn);
  931. }
  932. else
  933. {
  934. p->cbFunc(p->cbArg,mn,mp);
  935. }
  936. p->oi += mn + sizeof(unsigned);
  937. return rc;
  938. }
  939. unsigned cmTs1p1cDequeueMsgByteCount( cmTs1p1cH_t h )
  940. {
  941. cmTs1p1c_t* p = _cmTs1p1cHandleToPtr(h);
  942. return _cmTs1p1cDequeueMsgByteCount(p);
  943. }
  944. bool cmTs1p1cMsgWaiting( cmTs1p1cH_t h )
  945. { return cmTs1p1cDequeueMsgByteCount(h) > 0; }
  946. bool cmTs1p1cIsValid( cmTs1p1cH_t h )
  947. { return h.h != NULL; }
  948. //============================================================================================================================
  949. bool cmThIntCAS( int* addr, int old, int new )
  950. {
  951. #ifdef OS_OSX
  952. int rv = OSAtomicCompareAndSwap32Barrier(old,new,addr);
  953. return rv;
  954. #endif
  955. #ifdef OS_LINUX
  956. return __sync_bool_compare_and_swap(addr,old,new);
  957. #endif
  958. }
  959. bool cmThUIntCAS( unsigned* addr, unsigned old, unsigned new )
  960. {
  961. #ifdef OS_OSX
  962. return OSAtomicCompareAndSwap32Barrier((int)old,(int)new,(int*)addr);
  963. #endif
  964. #ifdef OS_LINUX
  965. return __sync_bool_compare_and_swap(addr,old,new);
  966. #endif
  967. }
  968. bool cmThFloatCAS( float* addr, float old, float new )
  969. {
  970. #ifdef OS_OSX
  971. return OSAtomicCompareAndSwap32Barrier(*(int*)(&old),*(int*)(&new),(int*)addr );
  972. #endif
  973. #ifdef OS_LINUX
  974. return __sync_bool_compare_and_swap((unsigned*)addr, *(unsigned*)(&old),*(unsigned*)(&new));
  975. #endif
  976. }
  977. #define OS_VER_10_6
  978. bool cmThPtrCAS( void* addr, void* old, void* neww )
  979. {
  980. #ifdef OS_OSX
  981. // REMOVE THIS HACK AND USE OSAtomicCompareAndSwapPtrBarrier() WHEN
  982. // A 64 BIT BUILD IS POSSIBLE ON OS-X.
  983. typedef struct
  984. {
  985. union
  986. {
  987. void* addr;
  988. #ifdef OS_VER_10_6
  989. int64_t val;
  990. #else
  991. int val;
  992. #endif
  993. } u;
  994. } s_t;
  995. s_t ov,nv;
  996. ov.u.addr = old;
  997. nv.u.addr = neww;
  998. #ifdef OS_VER_10_6
  999. int rv = OSAtomicCompareAndSwap64Barrier(ov.u.val,nv.u.val,(int64_t*)addr);
  1000. #else
  1001. int rv = OSAtomicCompareAndSwap32Barrier(ov.u.val,nv.u.val,(int*)addr);
  1002. #endif
  1003. return rv;
  1004. #endif
  1005. #ifdef OS_LINUX
  1006. #ifdef OS_VER_10_6
  1007. return __sync_bool_compare_and_swap((long long*)addr, (long long)old, (long long)neww);
  1008. #else
  1009. return __sync_bool_compare_and_swap((int*)addr,(int)old,(int)neww);
  1010. #endif
  1011. #endif
  1012. }
  1013. void cmThIntIncr( int* addr, int incr )
  1014. {
  1015. #ifdef OS_OSX
  1016. OSAtomicAdd32Barrier(incr,addr);
  1017. #endif
  1018. #ifdef OS_LINUX
  1019. // ... could also use __sync_add_and_fetch() ...
  1020. __sync_fetch_and_add(addr,incr);
  1021. #endif
  1022. }
  1023. void cmThUIntIncr( unsigned* addr, unsigned incr )
  1024. {
  1025. #ifdef OS_OSX
  1026. OSAtomicAdd32Barrier((int)incr,(int*)addr);
  1027. #endif
  1028. #ifdef OS_LINUX
  1029. __sync_fetch_and_add(addr,incr);
  1030. #endif
  1031. }
  1032. void cmThFloatIncr(float* addr, float incr )
  1033. {
  1034. float old,new;
  1035. do
  1036. {
  1037. old = *addr;
  1038. new = old + incr;
  1039. }while( cmThFloatCAS(addr,old,new)==0 );
  1040. }
  1041. void cmThIntDecr( int* addr, int decr )
  1042. {
  1043. #ifdef OS_OSX
  1044. OSAtomicAdd32Barrier(-decr,addr);
  1045. #endif
  1046. #ifdef OS_LINUX
  1047. __sync_fetch_and_sub(addr,decr);
  1048. #endif
  1049. }
  1050. void cmThUIntDecr( unsigned* addr, unsigned decr )
  1051. {
  1052. #ifdef OS_OSX
  1053. OSAtomicAdd32Barrier(-((int)decr),(int*)addr);
  1054. #endif
  1055. #ifdef OS_LINUX
  1056. __sync_fetch_and_sub(addr,decr);
  1057. #endif
  1058. }
  1059. void cmThFloatDecr(float* addr, float decr )
  1060. {
  1061. float old,new;
  1062. do
  1063. {
  1064. old = *addr;
  1065. new = old - decr;
  1066. }while( cmThFloatCAS(addr,old,new)==0 );
  1067. }
  1068. //============================================================================================================================
  1069. //
  1070. //
  1071. typedef pthread_t cmThreadId_t;
  1072. typedef struct
  1073. {
  1074. cmThreadId_t id; // id of this thread as returned by pthread_self()
  1075. char* buf; // buf[bn]
  1076. int ii; // input index
  1077. int oi; // output index (oi==ii == empty buffer)
  1078. } cmTsBuf_t;
  1079. // msg header - which is actually written AFTER the msg it is associated with
  1080. typedef struct cmTsHdr_str
  1081. {
  1082. int mn; // length of the msg
  1083. int ai; // buffer index
  1084. struct cmTsHdr_str* link; // pointer to next msg
  1085. } cmTsHdr_t;
  1086. typedef struct
  1087. {
  1088. cmErr_t err;
  1089. int bn; // bytes per thread buffer
  1090. cmTsBuf_t* a; // a[an] buffer array
  1091. unsigned an; // length of a[] - one buffer per thread
  1092. cmTsQueueCb_t cbFunc;
  1093. void* cbArg;
  1094. cmTsHdr_t* ilp; // prev msg hdr record
  1095. cmTsHdr_t* olp; // prev msg hdr record (wait for olp->link to be set to go to next record)
  1096. } cmTsMp1c_t;
  1097. cmTsMp1cH_t cmTsMp1cNullHandle = cmSTATIC_NULL_HANDLE;
  1098. void _cmTsMp1cPrint( cmTsMp1c_t* p )
  1099. {
  1100. unsigned i;
  1101. for(i=0; i<p->an; ++i)
  1102. printf("%2i ii:%3i oi:%3i\n",i,p->a[i].ii,p->a[i].oi);
  1103. }
  1104. cmTsMp1c_t* _cmTsMp1cHandleToPtr( cmTsMp1cH_t h )
  1105. {
  1106. cmTsMp1c_t* p = (cmTsMp1c_t*)h.h;
  1107. assert(p != NULL);
  1108. return p;
  1109. }
  1110. unsigned _cmTsMp1cBufIndex( cmTsMp1c_t* p, cmThreadId_t id )
  1111. {
  1112. unsigned i;
  1113. for(i=0; i<p->an; ++i)
  1114. if( p->a[i].id == id )
  1115. return i;
  1116. p->an = i+1;
  1117. p->a = cmMemResizePZ(cmTsBuf_t,p->a,p->an);
  1118. p->a[i].buf = cmMemAllocZ(char,p->bn);
  1119. p->a[i].id = id;
  1120. return i;
  1121. }
  1122. cmThRC_t cmTsMp1cDestroy( cmTsMp1cH_t* hp )
  1123. {
  1124. if( hp == NULL || cmTsMp1cIsValid(*hp) == false )
  1125. return kOkThRC;
  1126. cmTsMp1c_t* p = _cmTsMp1cHandleToPtr(*hp);
  1127. unsigned i;
  1128. for(i=0; i<p->an; ++i)
  1129. cmMemFree(p->a[i].buf);
  1130. cmMemPtrFree(&p->a);
  1131. cmMemFree(p);
  1132. hp->h = NULL;
  1133. return kOkThRC;
  1134. }
  1135. cmThRC_t cmTsMp1cCreate( cmTsMp1cH_t* hp, unsigned bufByteCnt, cmTsQueueCb_t cbFunc, void* cbArg, cmRpt_t* rpt )
  1136. {
  1137. cmThRC_t rc;
  1138. if((rc = cmTsMp1cDestroy(hp)) != kOkThRC )
  1139. return rc;
  1140. cmTsMp1c_t* p = cmMemAllocZ(cmTsMp1c_t,1);
  1141. cmErrSetup(&p->err,rpt,"TsMp1c Queue");
  1142. p->a = NULL;
  1143. p->an = 0;
  1144. p->bn = bufByteCnt;
  1145. p->cbFunc = cbFunc;
  1146. p->cbArg = cbArg;
  1147. p->ilp = NULL;
  1148. p->olp = NULL;
  1149. hp->h = p;
  1150. return rc;
  1151. }
  1152. void cmTsMp1cSetCbFunc( cmTsMp1cH_t h, cmTsQueueCb_t cbFunc, void* cbArg )
  1153. {
  1154. cmTsMp1c_t* p = _cmTsMp1cHandleToPtr(h);
  1155. p->cbFunc = cbFunc;
  1156. p->cbArg = cbArg;
  1157. }
  1158. cmTsQueueCb_t cmTsMp1cCbFunc( cmTsMp1cH_t h )
  1159. {
  1160. cmTsMp1c_t* p = _cmTsMp1cHandleToPtr(h);
  1161. return p->cbFunc;
  1162. }
  1163. void* cmTsMp1cCbArg( cmTsMp1cH_t h )
  1164. {
  1165. cmTsMp1c_t* p = _cmTsMp1cHandleToPtr(h);
  1166. return p->cbArg;
  1167. }
  1168. //#define CAS(addr,old,new) __sync_bool_compare_and_swap(addr,old,new)
  1169. //#define CAS(addr,old,neww) cmThPtrCAS(addr,old,neww)
  1170. cmThRC_t cmTsMp1cEnqueueSegMsg( cmTsMp1cH_t h, const void* msgPtrArray[], unsigned msgByteCntArray[], unsigned arrayCnt )
  1171. {
  1172. cmThRC_t rc = kOkThRC;
  1173. unsigned mn = 0;
  1174. cmTsMp1c_t* p = _cmTsMp1cHandleToPtr(h);
  1175. unsigned ai = _cmTsMp1cBufIndex( p, pthread_self() );
  1176. cmTsBuf_t* b = p->a + ai;
  1177. int i,bi,ei;
  1178. cmTsHdr_t hdr;
  1179. // Use a stored oi for the duration of this function.
  1180. // b->oi may be changed by the dequeue thread but storing it here
  1181. // at least prevents it from changing during the course of the this function.
  1182. // Note: b->oi is only used to check for buffer full. Even if it changes
  1183. // it would only mean that more bytes were available than calculated based
  1184. // on the stored value. A low estimate of the actual bytes available is
  1185. // never unsafe.
  1186. volatile int oi = b->oi;
  1187. // get the total count of bytes for this msg
  1188. for(i=0; i<arrayCnt; ++i)
  1189. mn += msgByteCntArray[i];
  1190. // dn = count of msg bytes + count of header bytes
  1191. int dn = mn + sizeof(hdr);
  1192. // if oi is ahead of ii in the queue then we must write
  1193. // in the area between ii and oi
  1194. if( oi > b->ii )
  1195. {
  1196. ei = oi-1; // (never allow ii to equal oi (that's the empty condition))
  1197. bi = b->ii;
  1198. }
  1199. else // otherwise oi is same or before ii in the queue and we have the option to wrap
  1200. {
  1201. // if the new msg will not fit at the end of the queue ....
  1202. if( b->ii + dn > p->bn )
  1203. {
  1204. bi = 0; // ... then wrap to the beginning
  1205. ei = oi-1; // (never allow ii to equal oi (that's the empty condition))
  1206. }
  1207. else
  1208. {
  1209. ei = p->bn; // otherwise write at the current location
  1210. bi = b->ii;
  1211. }
  1212. }
  1213. if( bi + dn > ei )
  1214. return cmErrMsg(&p->err,kBufFullThRC,"%i consecutive bytes is not available in the queue.",dn);
  1215. char* dp = b->buf + bi;
  1216. // write the msg
  1217. for(i=0; i<arrayCnt; ++i)
  1218. {
  1219. memcpy(dp,msgPtrArray[i],msgByteCntArray[i]);
  1220. dp += msgByteCntArray[i];
  1221. }
  1222. // setup the msg header
  1223. hdr.ai = ai;
  1224. hdr.mn = mn;
  1225. hdr.link = NULL;
  1226. // write the msg header (following the msg body in memory)
  1227. cmTsHdr_t* hp = (cmTsHdr_t*)dp;
  1228. memcpy(hp,&hdr,sizeof(hdr));
  1229. // increment the buffers input index
  1230. b->ii = bi + dn;
  1231. // update the link list head to point to this msg hdr
  1232. cmTsHdr_t* old_hp, *new_hp;
  1233. do
  1234. {
  1235. old_hp = p->ilp;
  1236. new_hp = hp;
  1237. }while(!cmThPtrCAS(&p->ilp,old_hp,new_hp));
  1238. // link the prev recd to this recd
  1239. if( old_hp != NULL )
  1240. old_hp->link = hp;
  1241. // if this is the first record written by this queue then prime the output list
  1242. do
  1243. {
  1244. old_hp = p->olp;
  1245. new_hp = hp;
  1246. if( old_hp != NULL )
  1247. break;
  1248. }while(!cmThPtrCAS(&p->olp,old_hp,new_hp));
  1249. //printf("%p %p %i\n",p->ilp,p->olp,p->olp->mn);
  1250. return rc;
  1251. }
  1252. cmThRC_t cmTsMp1cEnqueueMsg( cmTsMp1cH_t h, const void* dataPtr, unsigned byteCnt )
  1253. { return cmTsMp1cEnqueueSegMsg(h,&dataPtr,&byteCnt,1); }
  1254. unsigned cmTsMp1cAllocByteCount( cmTsMp1cH_t h )
  1255. {
  1256. cmTsMp1c_t* p = _cmTsMp1cHandleToPtr(h);
  1257. return p->bn;
  1258. }
  1259. unsigned cmTsMp1cAvailByteCount( cmTsMp1cH_t h )
  1260. {
  1261. cmTsMp1c_t* p = _cmTsMp1cHandleToPtr(h);
  1262. unsigned ai = _cmTsMp1cBufIndex(p,pthread_self());
  1263. const cmTsBuf_t* b = p->a + ai;
  1264. if( b->oi > b->ii )
  1265. return b->oi - b->ii - 1;
  1266. return (p->bn - b->ii) + b->oi - 1;
  1267. }
  1268. unsigned _cmTsMp1cNextMsgByteCnt( cmTsMp1c_t* p )
  1269. {
  1270. if( p->olp == NULL )
  1271. return 0;
  1272. // if the current msg has not yet been read
  1273. if( p->olp->mn != 0 )
  1274. return p->olp->mn;
  1275. // if the current msg has been read but a new next msg has been linked
  1276. if( p->olp->mn == 0 && p->olp->link != NULL )
  1277. {
  1278. // advance the buffer output msg past the prev msg header
  1279. char* hp = (char*)(p->olp + 1);
  1280. p->a[p->olp->ai].oi = hp - p->a[p->olp->ai].buf;
  1281. // advance msg pointer to point to the new msg header
  1282. p->olp = p->olp->link;
  1283. // return the size of the new msg
  1284. return p->olp->mn;
  1285. }
  1286. return 0;
  1287. }
  1288. cmThRC_t cmTsMp1cDequeueMsg( cmTsMp1cH_t h, void* dataPtr, unsigned byteCnt )
  1289. {
  1290. cmTsMp1c_t* p = _cmTsMp1cHandleToPtr(h);
  1291. // if there are no messages waiting
  1292. if( _cmTsMp1cNextMsgByteCnt(p) == 0 )
  1293. return kBufEmptyThRC;
  1294. char* hp = (char*)p->olp;
  1295. char* dp = hp - p->olp->mn; // the msg body is before the msg hdr
  1296. if( dataPtr == NULL )
  1297. {
  1298. p->cbFunc(p->cbArg,p->olp->mn,dp);
  1299. }
  1300. else
  1301. {
  1302. if( p->olp->mn > byteCnt )
  1303. return cmErrMsg(&p->err,kBufTooSmallThRC,"The return buffer constains too few bytes (%i) to contain %i bytes.",byteCnt,p->olp->mn);
  1304. memcpy(dataPtr,dp,p->olp->mn);
  1305. }
  1306. // advance the buffers output index past the msg body
  1307. p->a[p->olp->ai].oi = hp - p->a[p->olp->ai].buf;
  1308. // mark the msg as read
  1309. p->olp->mn = 0;
  1310. return kOkThRC;
  1311. }
  1312. bool cmTsMp1cMsgWaiting( cmTsMp1cH_t h )
  1313. {
  1314. cmTsMp1c_t* p = _cmTsMp1cHandleToPtr(h);
  1315. return _cmTsMp1cNextMsgByteCnt(p) != 0;
  1316. }
  1317. unsigned cmTsMp1cDequeueMsgByteCount( cmTsMp1cH_t h )
  1318. {
  1319. cmTsMp1c_t* p = _cmTsMp1cHandleToPtr(h);
  1320. return _cmTsMp1cNextMsgByteCnt(p);
  1321. }
  1322. bool cmTsMp1cIsValid( cmTsMp1cH_t h )
  1323. { return h.h != NULL; }
  1324. //============================================================================================================================/
  1325. //
  1326. // cmTsQueueTest()
  1327. //
  1328. // param recd for use by _cmTsQueueCb0() and
  1329. // the msg record passed between the sender
  1330. // threads and the receiver thread
  1331. typedef struct
  1332. {
  1333. unsigned id;
  1334. cmTsQueueH_t qH;
  1335. int val;
  1336. } _cmTsQCbParam_t;
  1337. // Generate a random number and put it in a TS queue
  1338. bool _cmTsQueueCb0(void* param)
  1339. {
  1340. _cmTsQCbParam_t* p = (_cmTsQCbParam_t*)param;
  1341. p->val = rand(); // generate a random number
  1342. // send the msg
  1343. if( cmTsQueueEnqueueMsg( p->qH, p, sizeof(_cmTsQCbParam_t)) == kOkThRC )
  1344. printf("in:%i %i\n",p->id,p->val);
  1345. else
  1346. printf("in error %i\n",p->id);
  1347. cmSleepUs(100*1000);
  1348. return true;
  1349. }
  1350. // Monitor a TS queue for incoming messages from _cmTsQueueCb1()
  1351. bool _cmTsQueueCb1(void* param)
  1352. {
  1353. // the thread param is a ptr to the TS queue to monitor.
  1354. cmTsQueueH_t* qp = (cmTsQueueH_t*)param;
  1355. cmThRC_t rc;
  1356. _cmTsQCbParam_t msg;
  1357. // dequeue any waiting messages
  1358. if((rc = cmTsQueueDequeueMsg( *qp, &msg, sizeof(msg))) == kOkThRC )
  1359. printf("out:%i %i\n",msg.id,msg.val);
  1360. else
  1361. {
  1362. if( rc != kBufEmptyThRC )
  1363. printf("out error:%i\n", rc);
  1364. }
  1365. return true;
  1366. }
  1367. // Test the TS queue by starting sender threads (threads 0 & 1)
  1368. // and a receiver thread (thread 2) and sending messages
  1369. // from the sender to the receiver.
  1370. void cmTsQueueTest( cmRpt_t* rpt )
  1371. {
  1372. cmThreadH_t th0=cmThreadNullHandle,th1=cmThreadNullHandle,th2=cmThreadNullHandle;
  1373. cmTsQueueH_t q=cmTsQueueNullHandle;
  1374. _cmTsQCbParam_t param0, param1;
  1375. // create a TS Queue
  1376. if( cmTsQueueCreate(&q,100,NULL,NULL,rpt) != kOkThRC )
  1377. goto errLabel;
  1378. // create thread 0
  1379. param0.id = 0;
  1380. param0.qH = q;
  1381. if( cmThreadCreate(&th0,_cmTsQueueCb0,&param0,rpt) != kOkThRC )
  1382. goto errLabel;
  1383. // create thread 1
  1384. param1.id = 1;
  1385. param1.qH = q;
  1386. if( cmThreadCreate(&th1,_cmTsQueueCb0,&param1,rpt) != kOkThRC )
  1387. goto errLabel;
  1388. // create thread 2
  1389. if( cmThreadCreate(&th2,_cmTsQueueCb1,&q,rpt) != kOkThRC )
  1390. goto errLabel;
  1391. // start thread 0
  1392. if( cmThreadPause(th0,0) != kOkThRC )
  1393. goto errLabel;
  1394. // start thread 1
  1395. if( cmThreadPause(th1,0) != kOkThRC )
  1396. goto errLabel;
  1397. // start thread 2
  1398. if( cmThreadPause(th2,0) != kOkThRC )
  1399. goto errLabel;
  1400. printf("any key to quit.");
  1401. getchar();
  1402. errLabel:
  1403. if( cmThreadIsValid(th0) )
  1404. if( cmThreadDestroy(&th0) != kOkThRC )
  1405. printf("Error destroying thread 0\n");
  1406. if( cmThreadIsValid(th1) )
  1407. if( cmThreadDestroy(&th1) != kOkThRC )
  1408. printf("Error destroying thread 1\n");
  1409. if( cmThreadIsValid(th2) )
  1410. if( cmThreadDestroy(&th2) != kOkThRC )
  1411. printf("Error destroying thread 1\n");
  1412. if( cmTsQueueIsValid(q) )
  1413. if( cmTsQueueDestroy(&q) != kOkThRC )
  1414. printf("Error destroying queue\n");
  1415. }
  1416. //============================================================================================================================/
  1417. //
  1418. // cmTs1p1cTest()
  1419. //
  1420. // param recd for use by _cmTsQueueCb0() and
  1421. // the msg record passed between the sender
  1422. // threads and the receiver thread
  1423. typedef struct
  1424. {
  1425. unsigned id;
  1426. cmTs1p1cH_t qH;
  1427. int val;
  1428. } _cmTs1p1cCbParam_t;
  1429. cmTs1p1cH_t cmTs1p1cNullHandle = cmSTATIC_NULL_HANDLE;
  1430. // Generate a random number and put it in a TS queue
  1431. bool _cmTs1p1cCb0(void* param)
  1432. {
  1433. _cmTs1p1cCbParam_t* p = (_cmTs1p1cCbParam_t*)param;
  1434. p->val = rand(); // generate a random number
  1435. // send the msg
  1436. if( cmTs1p1cEnqueueMsg( p->qH, p, sizeof(_cmTs1p1cCbParam_t)) == kOkThRC )
  1437. printf("in:%i %i\n",p->id,p->val);
  1438. else
  1439. printf("in error %i\n",p->id);
  1440. ++p->id;
  1441. cmSleepUs(100*1000);
  1442. return true;
  1443. }
  1444. // Monitor a TS queue for incoming messages from _cmTs1p1cCb1()
  1445. bool _cmTs1p1cCb1(void* param)
  1446. {
  1447. // the thread param is a ptr to the TS queue to monitor.
  1448. cmTs1p1cH_t* qp = (cmTs1p1cH_t*)param;
  1449. cmThRC_t rc;
  1450. _cmTs1p1cCbParam_t msg;
  1451. // dequeue any waiting messages
  1452. if((rc = cmTs1p1cDequeueMsg( *qp, &msg, sizeof(msg))) == kOkThRC )
  1453. printf("out:%i %i\n",msg.id,msg.val);
  1454. else
  1455. {
  1456. if( rc != kBufEmptyThRC )
  1457. printf("out error:%i\n", rc);
  1458. }
  1459. return true;
  1460. }
  1461. // Test the TS queue by starting sender threads (threads 0 & 1)
  1462. // and a receiver thread (thread 2) and sending messages
  1463. // from the sender to the receiver.
  1464. void cmTs1p1cTest( cmRpt_t* rpt )
  1465. {
  1466. cmThreadH_t th0=cmThreadNullHandle,th1=cmThreadNullHandle,th2=cmThreadNullHandle;
  1467. cmTs1p1cH_t q=cmTs1p1cNullHandle;
  1468. _cmTs1p1cCbParam_t param1;
  1469. // create a TS Queue
  1470. if( cmTs1p1cCreate(&q,28*2,NULL,NULL,rpt) != kOkThRC )
  1471. goto errLabel;
  1472. // create thread 1
  1473. param1.id = 0;
  1474. param1.qH = q;
  1475. if( cmThreadCreate(&th1,_cmTs1p1cCb0,&param1,rpt) != kOkThRC )
  1476. goto errLabel;
  1477. // create thread 2
  1478. if( cmThreadCreate(&th2,_cmTs1p1cCb1,&q,rpt) != kOkThRC )
  1479. goto errLabel;
  1480. // start thread 1
  1481. if( cmThreadPause(th1,0) != kOkThRC )
  1482. goto errLabel;
  1483. // start thread 2
  1484. if( cmThreadPause(th2,0) != kOkThRC )
  1485. goto errLabel;
  1486. printf("any key to quit.");
  1487. getchar();
  1488. errLabel:
  1489. if( cmThreadIsValid(th0) )
  1490. if( cmThreadDestroy(&th0) != kOkThRC )
  1491. printf("Error destroying thread 0\n");
  1492. if( cmThreadIsValid(th1) )
  1493. if( cmThreadDestroy(&th1) != kOkThRC )
  1494. printf("Error destroying thread 1\n");
  1495. if( cmThreadIsValid(th2) )
  1496. if( cmThreadDestroy(&th2) != kOkThRC )
  1497. printf("Error destroying thread 1\n");
  1498. if( cmTs1p1cIsValid(q) )
  1499. if( cmTs1p1cDestroy(&q) != kOkThRC )
  1500. printf("Error destroying queue\n");
  1501. }
  1502. //============================================================================================================================/
  1503. //
  1504. // cmTsMp1cTest()
  1505. //
  1506. // param recd for use by _cmTsQueueCb0() and
  1507. // the msg record passed between the sender
  1508. // threads and the receiver thread
  1509. typedef struct
  1510. {
  1511. unsigned id;
  1512. cmTsMp1cH_t qH;
  1513. int val;
  1514. } _cmTsMp1cCbParam_t;
  1515. unsigned _cmTsMp1cVal = 0;
  1516. // Incr the global value _cmTsMp1cVal and put it in a TS queue
  1517. bool _cmTsMp1cCb0(void* param)
  1518. {
  1519. _cmTsMp1cCbParam_t* p = (_cmTsMp1cCbParam_t*)param;
  1520. //p->val = __sync_fetch_and_add(&_cmTsMp1cVal,1);
  1521. cmThUIntIncr(&_cmTsMp1cVal,1);
  1522. p->val = _cmTsMp1cVal;
  1523. // send the msg
  1524. if( cmTsMp1cEnqueueMsg( p->qH, p, sizeof(_cmTsMp1cCbParam_t)) == kOkThRC )
  1525. printf("in:%i %i\n",p->id,p->val);
  1526. else
  1527. printf("in error %i\n",p->id);
  1528. cmSleepUs(100*1000);
  1529. return true;
  1530. }
  1531. // Monitor a TS queue for incoming messages from _cmTsMp1cCb1()
  1532. bool _cmTsMp1cCb1(void* param)
  1533. {
  1534. // the thread param is a ptr to the TS queue to monitor.
  1535. cmTsMp1cH_t* qp = (cmTsMp1cH_t*)param;
  1536. cmThRC_t rc;
  1537. _cmTsMp1cCbParam_t msg;
  1538. // dequeue any waiting messages
  1539. if((rc = cmTsMp1cDequeueMsg( *qp, &msg, sizeof(msg))) == kOkThRC )
  1540. printf("out - cons id:%i val:%i\n",msg.id,msg.val);
  1541. else
  1542. {
  1543. if( rc != kBufEmptyThRC )
  1544. printf("out error:%i\n", rc);
  1545. }
  1546. return true;
  1547. }
  1548. // Test the TS queue by starting sender threads (threads 0 & 1)
  1549. // and a receiver thread (thread 2) and sending messages
  1550. // from the sender to the receiver.
  1551. void cmTsMp1cTest( cmRpt_t* rpt )
  1552. {
  1553. cmThreadH_t th0=cmThreadNullHandle,th1=cmThreadNullHandle,th2=cmThreadNullHandle;
  1554. cmTsMp1cH_t q=cmTsMp1cNullHandle;
  1555. _cmTsMp1cCbParam_t param0, param1;
  1556. // create a TS Queue
  1557. if( cmTsMp1cCreate(&q,1000,NULL,NULL,rpt) != kOkThRC )
  1558. goto errLabel;
  1559. // create thread 0 - producer 0
  1560. param0.id = 0;
  1561. param0.qH = q;
  1562. if( cmThreadCreate(&th0,_cmTsMp1cCb0,&param0,rpt) != kOkThRC )
  1563. goto errLabel;
  1564. // create thread 1 - producer 1
  1565. param1.id = 1;
  1566. param1.qH = q;
  1567. if( cmThreadCreate(&th1,_cmTsMp1cCb0,&param1,rpt) != kOkThRC )
  1568. goto errLabel;
  1569. // create thread 2 - consumer 0
  1570. if( cmThreadCreate(&th2,_cmTsMp1cCb1,&q,rpt) != kOkThRC )
  1571. goto errLabel;
  1572. // start thread 0
  1573. if( cmThreadPause(th0,0) != kOkThRC )
  1574. goto errLabel;
  1575. // start thread 1
  1576. if( cmThreadPause(th1,0) != kOkThRC )
  1577. goto errLabel;
  1578. // start thread 2
  1579. if( cmThreadPause(th2,0) != kOkThRC )
  1580. goto errLabel;
  1581. printf("any key to quit.");
  1582. getchar();
  1583. errLabel:
  1584. if( cmThreadIsValid(th0) )
  1585. if( cmThreadDestroy(&th0) != kOkThRC )
  1586. printf("Error destroying thread 0\n");
  1587. if( cmThreadIsValid(th1) )
  1588. if( cmThreadDestroy(&th1) != kOkThRC )
  1589. printf("Error destroying thread 1\n");
  1590. if( cmThreadIsValid(th2) )
  1591. if( cmThreadDestroy(&th2) != kOkThRC )
  1592. printf("Error destroying thread 1\n");
  1593. if( cmTsMp1cIsValid(q) )
  1594. if( cmTsMp1cDestroy(&q) != kOkThRC )
  1595. printf("Error destroying queue\n");
  1596. }
  1597. void cmSleepUs( unsigned microseconds )
  1598. { usleep(microseconds); }
  1599. void cmSleepMs( unsigned milliseconds )
  1600. { cmSleepUs(milliseconds*1000); }