libcm is a C development framework with an emphasis on audio signal processing applications.
Du kannst nicht mehr als 25 Themen auswählen Themen müssen mit entweder einem Buchstaben oder einer Ziffer beginnen. Sie können Bindestriche („-“) enthalten und bis zu 35 Zeichen lang sein.

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. }