libcm is a C development framework with an emphasis on audio signal processing applications.
Nevar pievienot vairāk kā 25 tēmas Tēmai ir jāsākas ar burtu vai ciparu, tā var saturēt domu zīmes ('-') un var būt līdz 35 simboliem gara.

cmThread.c 47KB

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