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.

cmRtNet.c 30KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695696697698699700701702703704705706707708709710711712713714715716717718719720721722723724725726727728729730731732733734735736737738739740741742743744745746747748749750751752753754755756757758759760761762763764765766767768769770771772773774775776777778779780781782783784785786787788789790791792793794795796797798799800801802803804805806807808809810811812813814815816817818819820821822823824825826827828829830831832833834835836837838839840841842843844845846847848849850851852853854855856857858859860861862863864865866867868869870871872873874875876877878879880881882883884885886887888889890891892893894895896897898899900901902903904905906907908909910911912913914915916917918919920921922923924925926927928929930931932933934935936937938939940941942943944945946947948949950951952953954955956957958959960961962963964965966967968969970971972973974975976977978979980981982983984985986987988989990991992993994995996997998999100010011002100310041005100610071008100910101011101210131014101510161017101810191020102110221023102410251026102710281029103010311032103310341035103610371038103910401041104210431044104510461047104810491050105110521053105410551056105710581059106010611062106310641065106610671068106910701071107210731074107510761077107810791080108110821083108410851086108710881089109010911092109310941095109610971098109911001101110211031104110511061107110811091110111111121113111411151116111711181119112011211122112311241125112611271128
  1. #include "cmGlobal.h"
  2. #include "cmRpt.h"
  3. #include "cmErr.h"
  4. #include "cmCtx.h"
  5. #include "cmMem.h"
  6. #include "cmMallocDebug.h"
  7. #include "cmLinkedHeap.h"
  8. #include "cmUdpPort.h"
  9. #include "cmRtSysMsg.h"
  10. #include "cmRtNet.h"
  11. #include "cmTime.h"
  12. #include "cmText.h"
  13. // flags for cmRtNetNode_t.flags;
  14. enum
  15. {
  16. kLocalNodeNetFl = 0x01,
  17. kValidNodeNetFl = 0x02
  18. };
  19. // flags for cmRtNet_t.flags
  20. enum
  21. {
  22. kReportSyncNetFl = 0x01
  23. };
  24. struct cmRtNetNode_str;
  25. typedef struct cmRtNetEnd_str
  26. {
  27. cmChar_t* label;
  28. unsigned id;
  29. struct cmRtNetNode_str* np; // Owner node.
  30. struct cmRtNetEnd_str* link;
  31. } cmRtNetEnd_t;
  32. struct cmRtNet_str;
  33. typedef struct cmRtNetNode_str
  34. {
  35. unsigned rtSubIdx; // rtSubIdx of the sub-system which owns this node
  36. cmChar_t* label; // Node label.
  37. struct sockaddr_in sockaddr; // Socket address
  38. cmChar_t* addr; // IP Address (human readable)
  39. cmUdpPort_t port; // Socket port
  40. unsigned flags; // See kXXXNodeNetFl flags above.
  41. unsigned endPtIdx; // tracks the next endpoint to send during sync-mode
  42. unsigned endPtCnt; // local-node=actual cnt of endpt's remote-node:expected cnt of endpt's
  43. cmTimeSpec_t lastSendTime; // Time of last message sent
  44. cmRtNetEnd_t* ends; // End point list for this node
  45. struct cmRtNetNode_str* link;
  46. } cmRtNetNode_t;
  47. typedef struct cmRtNet_str
  48. {
  49. cmErr_t err; // Error state object
  50. unsigned flags; // See kXXXNetFl above.
  51. unsigned rtSubIdx; // rtSubIdx of the owning sub-system
  52. cmUdpH_t udpH; // UDP port handle
  53. cmUdpCallback_t cbFunc; // Client callback to receive incoming messages from network.
  54. void* cbArg; //
  55. cmRtNetNode_t* nodes; // Node list.
  56. cmRtNetNode_t* localNode; // Pointer to local node (which is also in node list)
  57. unsigned udpRecvBufByteCnt; // UDP port receive buffer size.
  58. unsigned udpTimeOutMs; // UDP time-out period
  59. cmChar_t* bcastAddr; // Network broadcast address
  60. } cmRtNet_t;
  61. cmRtNetH_t cmRtNetNullHandle = cmSTATIC_NULL_HANDLE;
  62. cmRtNetEndptH_t cmRtNetEndptNullHandle = cmSTATIC_NULL_HANDLE;
  63. cmRtNet_t* _cmRtNetHandleToPtr( cmRtNetH_t h )
  64. {
  65. cmRtNet_t* p = (cmRtNet_t*)h.h;
  66. assert( p != NULL );
  67. return p;
  68. }
  69. cmRtNetEnd_t* _cmRtNetEndptHandleToPtr( cmRtNetEndptH_t h )
  70. {
  71. cmRtNetEnd_t* p = (cmRtNetEnd_t*)h.h;
  72. assert( p != NULL );
  73. return p;
  74. }
  75. void _cmRtNetVRpt( cmRtNet_t* p, const cmChar_t* fmt, va_list vl )
  76. {
  77. if( cmIsFlag(p->flags,kReportSyncNetFl) )
  78. cmRptVPrintf(p->err.rpt,fmt,vl);
  79. }
  80. void _cmRtNetRpt( cmRtNet_t* p, const cmChar_t* fmt, ... )
  81. {
  82. va_list vl;
  83. va_start(vl,fmt);
  84. _cmRtNetVRpt(p,fmt,vl);
  85. va_end(vl);
  86. }
  87. cmRtNetNode_t* _cmRtNetFindNode( cmRtNet_t* p, const cmChar_t* label )
  88. {
  89. if( label == NULL )
  90. return NULL;
  91. cmRtNetNode_t* np = p->nodes;
  92. for(; np!=NULL; np=np->link)
  93. if( strcmp(label,np->label)==0)
  94. return np;
  95. return NULL;
  96. }
  97. cmRtNetNode_t* _cmRtNetFindNodeFromSockAddr( cmRtNet_t* p, const struct sockaddr_in* saddr )
  98. {
  99. if( saddr == NULL )
  100. return NULL;
  101. cmRtNetNode_t* np = p->nodes;
  102. for(; np!=NULL; np=np->link)
  103. if( np->sockaddr.sin_addr.s_addr == saddr->sin_addr.s_addr && np->sockaddr.sin_port == saddr->sin_port )
  104. return np;
  105. return NULL;
  106. }
  107. void _cmRtNetFreeNode( cmRtNetNode_t* np )
  108. {
  109. cmRtNetEnd_t* ep = np->ends;
  110. while( ep != NULL )
  111. {
  112. cmRtNetEnd_t* nep = ep->link;
  113. cmMemFree(ep->label);
  114. cmMemFree(ep);
  115. ep = nep;
  116. }
  117. cmMemFree(np->label);
  118. cmMemFree(np->addr);
  119. cmMemFree(np);
  120. }
  121. void _cmRtNetReleaseNodes( cmRtNet_t* p )
  122. {
  123. cmRtNetNode_t* np = p->nodes;
  124. while( np != NULL )
  125. {
  126. cmRtNetNode_t* nnp = np->link;
  127. _cmRtNetFreeNode(np);
  128. np = nnp;
  129. }
  130. p->nodes = NULL;
  131. p->localNode = NULL;
  132. }
  133. cmRtNetRC_t _cmRtNetReleaseNode( cmRtNet_t* p, cmRtNetNode_t* np )
  134. {
  135. cmRtNetNode_t* cnp = p->nodes;
  136. cmRtNetNode_t* pnp = NULL;
  137. while( cnp != NULL )
  138. {
  139. cmRtNetNode_t* nnp = cnp->link;
  140. if( np == cnp )
  141. {
  142. if( pnp == NULL )
  143. p->nodes = np->link;
  144. else
  145. pnp->link = np->link;
  146. _cmRtNetFreeNode(np);
  147. return kOkNetRC;
  148. }
  149. pnp = np;
  150. cnp = nnp;
  151. }
  152. assert(0);
  153. return cmErrMsg(&p->err,kNodeNotFoundNetRC,"Node to release not found.");
  154. }
  155. cmRtNetRC_t _cmRtNetCreateNode( cmRtNet_t* p, const cmChar_t* label, unsigned rtSubIdx, const cmChar_t* addr, cmUdpPort_t port, const struct sockaddr_in* saddr, unsigned flags, unsigned endPtCnt )
  156. {
  157. cmRtNetRC_t rc = kOkNetRC;
  158. cmRtNetNode_t* np;
  159. if( cmTextIsEmpty(label) )
  160. return cmErrMsg(&p->err,kInvalidLabelNetRC,"A null or blank node label was encountered.");
  161. if((np = _cmRtNetFindNode(p,label)) != NULL )
  162. return cmErrMsg(&p->err,kDuplLabelNetRC,"The node label '%s' is already in use.",cmStringNullGuard(label));
  163. np = cmMemAllocZ(cmRtNetNode_t,1);
  164. np->label = cmMemAllocStr(label);
  165. if( saddr != NULL )
  166. np->sockaddr = *saddr;
  167. np->rtSubIdx = rtSubIdx;
  168. np->addr = addr==NULL ? NULL : cmMemAllocStr(addr);
  169. np->port = port;
  170. np->flags = flags;
  171. np->endPtCnt = endPtCnt;
  172. np->link = p->nodes;
  173. p->nodes = np;
  174. return rc;
  175. }
  176. cmRtNetEnd_t* _cmRtNetFindNodeEnd(cmRtNetNode_t* np, const cmChar_t* endPtLabel )
  177. {
  178. cmRtNetEnd_t* ep = np->ends;
  179. for(; ep!=NULL; ep=ep->link)
  180. if( strcmp(ep->label,endPtLabel)==0 )
  181. return ep;
  182. return NULL;
  183. }
  184. cmRtNetEnd_t* _cmRtNetIndexToEndpoint( cmRtNet_t* p, cmRtNetNode_t* np, unsigned endIndex )
  185. {
  186. cmRtNetEnd_t* ep = np->ends;
  187. unsigned i;
  188. for(i=0; ep!=NULL; ep=ep->link)
  189. {
  190. if( i == endIndex )
  191. return ep;
  192. ++i;
  193. }
  194. return NULL;
  195. }
  196. cmRtNetRC_t _cmRtNetCreateEndpoint( cmRtNet_t* p, cmRtNetNode_t* np, const cmChar_t* endPtLabel, unsigned endPtId )
  197. {
  198. if( endPtLabel == NULL )
  199. return cmErrMsg(&p->err,kInvalidLabelNetRC,"A null or blank node label was encountered.");
  200. if( _cmRtNetFindNodeEnd( np, endPtLabel) != NULL)
  201. return cmErrMsg(&p->err,kDuplEndNetRC,"A duplicate endpoint ('%s') was encountered on node '%s'.",endPtLabel,np->label);
  202. cmRtNetRC_t rc = kOkNetRC;
  203. cmRtNetEnd_t* ep = cmMemAllocZ(cmRtNetEnd_t,1);
  204. ep->label = cmMemAllocStr(endPtLabel);
  205. ep->id = endPtId;
  206. ep->np = np;
  207. ep->link = np->ends;
  208. np->ends = ep;
  209. return rc;
  210. }
  211. unsigned _cmRtNetNodeEndpointCount( cmRtNetNode_t* np )
  212. {
  213. cmRtNetEnd_t* ep = np->ends;
  214. unsigned n = 0;
  215. for(; ep!=NULL; ep=ep->link)
  216. ++n;
  217. return n;
  218. }
  219. unsigned _cmRtNetSyncMsgSerialByteCount( const cmRtNetSyncMsg_t* m )
  220. {
  221. return sizeof(cmRtNetSyncMsg_t) + (m->label==NULL ? 1 : strlen(m->label) + 1);
  222. }
  223. cmRtNetRC_t _cmRtNetSerializeSyncMsg( cmRtNet_t* p, const cmRtNetSyncMsg_t* m, void* buf, unsigned n )
  224. {
  225. unsigned bn = _cmRtNetSyncMsgSerialByteCount(m);
  226. char* b = (char*)buf;
  227. if( bn > n )
  228. return cmErrMsg(&p->err,kBufToSmallNetRC,"Serialize buffer too small.");
  229. memcpy(b,m,sizeof(*m));
  230. strcpy(b + sizeof(*m),m->label==NULL ? "" : m->label);
  231. return kOkNetRC;
  232. }
  233. cmRtNetRC_t _cmRtNetDeserializeSyncMsg( const void* buf, unsigned n, cmRtNetSyncMsg_t* m )
  234. {
  235. assert( n > sizeof(*m));
  236. memcpy(m,buf,sizeof(*m));
  237. const cmRtNetSyncMsg_t* mp = (const cmRtNetSyncMsg_t*)buf;
  238. const cmChar_t* s = ((const cmChar_t*)(mp)) + mp->hdrByteCnt;
  239. m->label = cmMemAllocStr(s);
  240. return kOkNetRC;
  241. }
  242. cmRtNetRC_t _cmRtNetSendSyncMsg( cmRtNet_t* p, cmRtNetNode_t* np, cmRtNetSelId_t selId, const cmChar_t* msgLabel, unsigned msgId, unsigned msgRtSubIdx )
  243. {
  244. cmRtNetSyncMsg_t m;
  245. cmRtNetRC_t rc = kOkNetRC;
  246. cmUdpRC_t udpRC = kOkUdpRC;
  247. m.hdr.rtSubIdx = cmInvalidIdx;
  248. m.hdr.selId = kNetSyncSelRtId;
  249. m.selId = selId;
  250. m.hdrByteCnt = sizeof(cmRtNetSyncMsg_t);
  251. m.label = msgLabel;
  252. m.id = msgId;
  253. m.rtSubIdx = msgRtSubIdx;
  254. // determine size of msg to send
  255. unsigned n = _cmRtNetSyncMsgSerialByteCount(&m);
  256. cmChar_t buf[n];
  257. // serialize msg into buf[]
  258. if((rc = _cmRtNetSerializeSyncMsg(p,&m,buf,n)) != kOkNetRC )
  259. return rc;
  260. // send the msg
  261. if( selId == kHelloSelNetId )
  262. udpRC = cmUdpSend2(p->udpH, buf, n, p->bcastAddr, np->port );
  263. else
  264. udpRC = cmUdpSendTo(p->udpH, buf, n, &np->sockaddr );
  265. // check for send errors
  266. if( udpRC != kOkUdpRC )
  267. {
  268. rc = cmErrMsg(&p->err,kUdpPortFailNetRC,"Sync msg. send on UDP port failed.");
  269. }
  270. else
  271. {
  272. // record the last send time
  273. cmTimeGet(&np->lastSendTime);
  274. }
  275. return rc;
  276. }
  277. cmRtNetRC_t _cmRtNetFree( cmRtNet_t* p )
  278. {
  279. cmRtNetRC_t rc = kOkNetRC;
  280. if( cmUdpFree(&p->udpH) != kOkUdpRC )
  281. cmErrMsg(&p->err,kUdpPortFailNetRC,"UDP Port free failed.");
  282. _cmRtNetReleaseNodes(p);
  283. cmMemFree(p->bcastAddr);
  284. cmMemFree(p);
  285. return rc;
  286. }
  287. const cmRtNetNode_t* _cmRtNetIndexToRemoteNode( cmRtNet_t* p, unsigned idx )
  288. {
  289. const cmRtNetNode_t* np = p->nodes;
  290. unsigned i = 0;
  291. for(; np!=NULL; np=np->link)
  292. if( np != p->localNode )
  293. {
  294. if( i == idx )
  295. return np;
  296. ++i;
  297. }
  298. return NULL;
  299. }
  300. const cmRtNetEnd_t* _cmRtNetIndexToEndpt( const cmRtNetNode_t* np, unsigned endIdx )
  301. {
  302. const cmRtNetEnd_t* ep = np->ends;
  303. unsigned i = 0;
  304. for(; ep!=NULL; ep=ep->link,++i)
  305. if( i==endIdx )
  306. return ep;
  307. return NULL;
  308. }
  309. const cmRtNetEnd_t* _cmRtNetFindEndpt( cmRtNet_t* p, unsigned nodeIdx, unsigned epIdx )
  310. {
  311. const cmRtNetNode_t* np;
  312. const cmRtNetEnd_t* ep;
  313. if((np = _cmRtNetIndexToRemoteNode( p, nodeIdx )) == NULL )
  314. return NULL;
  315. if((ep = _cmRtNetIndexToEndpt( np, epIdx )) == NULL )
  316. return NULL;
  317. return ep;
  318. }
  319. const cmChar_t* cmRtNetSyncMsgLabel( const cmRtNetSyncMsg_t* m )
  320. {
  321. if( m->selId==kNodeSelNetId || m->selId==kEndpointSelNetId )
  322. return (const cmChar_t*)(m+1);
  323. return "";
  324. }
  325. cmRtNetRC_t cmRtNetAlloc( cmCtx_t* ctx, cmRtNetH_t* hp, unsigned rtSubIdx, cmUdpCallback_t cbFunc, void* cbArg )
  326. {
  327. cmRtNetRC_t rc;
  328. if((rc = cmRtNetFree(hp)) != kOkNetRC )
  329. return rc;
  330. cmRtNet_t* p = cmMemAllocZ(cmRtNet_t,1);
  331. cmErrSetup(&p->err,&ctx->rpt,"cmRtNet");
  332. // allocate the UDP port
  333. if(cmUdpAlloc(ctx,&p->udpH) != kOkUdpRC )
  334. {
  335. cmErrMsg(&p->err,kUdpPortFailNetRC,"UDP Port allocate failed.");
  336. goto errLabel;
  337. }
  338. p->rtSubIdx = rtSubIdx;
  339. p->udpTimeOutMs = 50;
  340. p->udpRecvBufByteCnt = 8192;
  341. p->cbFunc = cbFunc;
  342. p->cbArg = cbArg;
  343. hp->h = p;
  344. errLabel:
  345. if(rc != kOkNetRC )
  346. _cmRtNetFree(p);
  347. return rc;
  348. }
  349. cmRtNetRC_t cmRtNetFree( cmRtNetH_t* hp )
  350. {
  351. cmRtNetRC_t rc = kOkNetRC;
  352. if( hp==NULL || cmRtNetIsValid(*hp)==false )
  353. return rc;
  354. cmRtNet_t* p = _cmRtNetHandleToPtr(*hp);
  355. if((rc = _cmRtNetFree(p)) != kOkNetRC )
  356. return rc;
  357. hp->h = NULL;
  358. return rc;
  359. }
  360. const cmChar_t* cmRtNetLocalHostName( cmRtNetH_t h )
  361. {
  362. cmRtNet_t* p = _cmRtNetHandleToPtr(h);
  363. return cmUdpHostName(p->udpH);
  364. }
  365. bool cmRtNetIsValid( cmRtNetH_t h )
  366. { return h.h !=NULL; }
  367. cmUdpH_t cmRtNetUdpPortHandle( cmRtNetH_t h )
  368. {
  369. cmRtNet_t* p = _cmRtNetHandleToPtr(h);
  370. return p->udpH;
  371. }
  372. cmRtNetRC_t _cmRtNetSendEndpointReplyMsg( cmRtNet_t* p, cmRtNetNode_t* np, cmRtNetSelId_t srcSelId )
  373. {
  374. cmRtNetRC_t rc = kOkNetRC;
  375. cmRtNetEnd_t* ep = NULL;
  376. const cmChar_t* msgLabel = NULL;
  377. unsigned msgId = cmInvalidId;
  378. unsigned msgRtSubIdx = cmInvalidIdx;
  379. cmRtNetSelId_t selId = kEndpointSelNetId;
  380. const cmChar_t* rptLabel = "endpoint";
  381. if( np == NULL )
  382. return cmErrMsg(&p->err,kNodeNotFoundNetRC,"The net node associated with an endpoint reply was not found.");
  383. // if we got here by receiving a 'done' msg from the remote node ...
  384. if( srcSelId == kDoneSelNetId )
  385. {
  386. // ... then mark the remote node as having recieved all endpoints
  387. unsigned n;
  388. if((n = _cmRtNetNodeEndpointCount(np)) != np->endPtCnt )
  389. rc = cmErrMsg(&p->err,kNodeEndCntErrNetRC,"The node '%s' expected %i endpoints but received %i.",cmStringNullGuard(np->label),np->endPtCnt,n);
  390. else
  391. np->flags = cmSetFlag(np->flags,kValidNodeNetFl);
  392. }
  393. // attempt to get the next local endpoint to send ...
  394. if((ep = _cmRtNetIndexToEndpoint(p,p->localNode,np->endPtIdx)) != NULL )
  395. {
  396. msgLabel = ep->label; // ... send next local endpoint
  397. msgId = ep->id;
  398. msgRtSubIdx = ep->np->rtSubIdx;
  399. }
  400. else // .... all local endpoints have been sent
  401. {
  402. selId = kInvalidSelNetId;
  403. rptLabel = "done";
  404. // verify that no endpoints are available
  405. if( np->endPtIdx < p->localNode->endPtCnt )
  406. rc = cmErrMsg(&p->err,kSyncFailNetRC,"More endpoints are available to send but are not reachable.");
  407. else
  408. {
  409. // if the remote node still has endpts to send then continue
  410. // sending 'done' messages.
  411. if( np->endPtIdx==p->localNode->endPtCnt || srcSelId != kDoneSelNetId )
  412. selId = kDoneSelNetId;
  413. }
  414. }
  415. // selId is set to kInvalidSelNetId when we encounter the (stop) criteria
  416. if( selId != kInvalidSelNetId )
  417. {
  418. if((rc = _cmRtNetSendSyncMsg(p,np,selId,msgLabel,msgId,msgRtSubIdx)) != kOkNetRC )
  419. rc = cmErrMsg(&p->err,rc,"Send '%s' to %s:%s:%i failed.",rptLabel,cmStringNullGuard(np->label),cmStringNullGuard(np->addr),np->port);
  420. else
  421. _cmRtNetRpt(p,"Sent %s.\n",cmStringNullGuard(rptLabel));
  422. np->endPtIdx += 1;
  423. }
  424. return rc;
  425. }
  426. bool _cmRtNetIsSyncModeMsg( const void* data, unsigned dataByteCnt )
  427. {
  428. cmRtNetSyncMsg_t* m = (cmRtNetSyncMsg_t*)data;
  429. return dataByteCnt >= sizeof(cmRtSysMsgHdr_t) && m->hdr.selId == kNetSyncSelRtId;
  430. }
  431. // When the network message recieve function (See cmRtNetAlloc() 'cbFunc')
  432. // receives a message with the cmRtSysMsgHdr_t.selId == kNetSyncSelRtId
  433. // it should call this function to update the current sync state of the
  434. // cmRtNet.
  435. cmRtNetRC_t _cmRtNetSyncModeRecv( cmRtNet_t* p, const char* data, unsigned dataByteCnt, const struct sockaddr_in* fromAddr )
  436. {
  437. cmRtNetRC_t rc = kOkNetRC;
  438. cmRtNetSyncMsg_t m;
  439. m.label = NULL;
  440. assert( _cmRtNetIsSyncModeMsg(data,dataByteCnt));
  441. if( _cmRtNetDeserializeSyncMsg(data,dataByteCnt,&m) != kOkNetRC )
  442. {
  443. rc = cmErrMsg(&p->err,rc,"Net sync. receive failed due to deserialize fail.");
  444. goto errLabel;
  445. }
  446. _cmRtNetRpt(p,"recv from:%s\n",cmUdpAddrToString(p->udpH, fromAddr ));
  447. assert( m.hdr.selId == kNetSyncSelRtId );
  448. // attempt to locate the remote node which sent the msg
  449. cmRtNetNode_t* np = _cmRtNetFindNodeFromSockAddr(p,fromAddr);
  450. switch( m.selId )
  451. {
  452. case kHelloSelNetId:
  453. // if this is a response to a broadcast from the local node then ignore it
  454. if(m.label!=NULL && p->localNode->label!=NULL && (np = _cmRtNetFindNode(p,m.label)) != NULL && strcmp(p->localNode->label,m.label)==0 )
  455. {
  456. const cmChar_t* fromAddrStr = cmUdpAddrToString(p->udpH,fromAddr);
  457. const cmChar_t* localAddrStr = cmUdpAddrToString(p->udpH,cmUdpLocalAddr(p->udpH));
  458. if( fromAddrStr!=NULL && localAddrStr!=NULL && strcmp(fromAddrStr,localAddrStr)!=0)
  459. cmErrMsg(&p->err,kDuplLocalNetRC,"The node label '%s' appears to be duplicated at address %s and locally.",cmStringNullGuard(m.label),fromAddrStr);
  460. np->sockaddr = *fromAddr;
  461. goto errLabel;
  462. }
  463. // fall through
  464. case kNodeSelNetId:
  465. {
  466. // if the node already exists ...
  467. if( np != NULL )
  468. {
  469. // ... delete it because we are about to get new info. about it.
  470. if((rc = _cmRtNetReleaseNode(p,np )) != kOkNetRC )
  471. goto errLabel;
  472. }
  473. // create a node proxy to represent the remote node
  474. // (Note:m.id == remote node endpoint count (i.e. the count of endpoints expected for the remote node.))
  475. if(( rc = _cmRtNetCreateNode(p,m.label,m.rtSubIdx,NULL,0,fromAddr,0,m.id)) != kOkNetRC )
  476. goto errLabel;
  477. np = p->nodes; // newest node is always the first node
  478. // send response
  479. switch( m.selId )
  480. {
  481. case kHelloSelNetId:
  482. _cmRtNetRpt(p,"rcv hello\n"); // reply with local node
  483. rc = _cmRtNetSendSyncMsg( p, np, kNodeSelNetId, p->localNode->label, p->localNode->endPtCnt, p->localNode->rtSubIdx );
  484. break;
  485. case kNodeSelNetId:
  486. _cmRtNetRpt(p,"rcv node\n");
  487. _cmRtNetSendEndpointReplyMsg( p, np, m.selId ); // reply with first endpoint
  488. break;
  489. default:
  490. assert(0);
  491. }
  492. }
  493. break;
  494. case kDoneSelNetId:
  495. //case kEndpointAckSelNetId:
  496. rc = _cmRtNetSendEndpointReplyMsg(p,np,m.selId);
  497. break;
  498. case kEndpointSelNetId:
  499. {
  500. cmRtNetEnd_t* ep;
  501. // verify the remote node exists.
  502. if( np == NULL )
  503. {
  504. rc = cmErrMsg(&p->err,kNodeNotFoundNetRC,"The net node associated with an endpoint receive was not found.");
  505. goto errLabel;
  506. }
  507. // attempt to find the end point
  508. if((ep = _cmRtNetFindNodeEnd(np, m.label)) != NULL )
  509. ep->id = m.id; // the endpoint was found update the endPtId
  510. else
  511. {
  512. // create a local proxy for the endpoint
  513. if((rc = _cmRtNetCreateEndpoint(p,np,m.label,m.id)) != kOkNetRC )
  514. goto errLabel;
  515. }
  516. // reply with a local endpoint or 'done' msg
  517. rc = _cmRtNetSendEndpointReplyMsg( p, np, m.selId );
  518. }
  519. break;
  520. default:
  521. assert(0);
  522. break;
  523. }
  524. errLabel:
  525. cmMemFree((cmChar_t*)m.label);
  526. return rc;
  527. }
  528. unsigned _cmRtNetAddrToNodeIndex( cmRtNet_t* p, const struct sockaddr_in* addr )
  529. {
  530. unsigned i;
  531. cmRtNetNode_t* np = p->nodes;
  532. for(i=0; np!=NULL; np=np->link,++i)
  533. if( cmUdpAddrIsEqual( addr, &np->sockaddr ) )
  534. return i;
  535. return cmInvalidIdx;
  536. }
  537. // This is called in the context of cmRtNetReceive().
  538. void _cmRtNetRecv( void* cbArg, const char* data, unsigned dataByteCnt, const struct sockaddr_in* fromAddr )
  539. {
  540. cmRtNet_t* p = (cmRtNet_t*)cbArg;
  541. // if this is a sync msg - then handle it here
  542. if( _cmRtNetIsSyncModeMsg(data,dataByteCnt))
  543. _cmRtNetSyncModeRecv(p,data,dataByteCnt,fromAddr);
  544. else
  545. {
  546. // All non-sync messages arriving here are prefixed by a cmRtNetMsg_t header - fill in the source addr here.
  547. // NOTE: the source addr could be filled in by the sender but this would increase the size
  548. // of the msg. Instead we choose the more time consuming method of looking up the
  549. // soure node here - hmmmm????.
  550. cmRtNetMsg_t* hdr = (cmRtNetMsg_t*)(data);
  551. hdr->srcNodeIdx = _cmRtNetAddrToNodeIndex(p,fromAddr);
  552. }
  553. p->cbFunc(p->cbArg,data,dataByteCnt,fromAddr);
  554. }
  555. cmRtNetRC_t cmRtNetInitialize( cmRtNetH_t h, const cmChar_t* bcastAddr, const cmChar_t* nodeLabel, const cmChar_t* ipAddr, cmUdpPort_t port )
  556. {
  557. cmRtNet_t* p = _cmRtNetHandleToPtr(h);
  558. cmRtNetRC_t rc;
  559. // release the local node
  560. if((rc = cmRtNetFinalize(h)) != kOkNetRC )
  561. goto errLabel;
  562. if( cmTextIsEmpty(bcastAddr) )
  563. {
  564. rc = cmErrMsg(&p->err,kInvalidArgNetRC,"The 'broadcast address' is not valid.");
  565. goto errLabel;
  566. }
  567. // if this is the local node then initialze the local socket
  568. if( cmUdpInit(p->udpH,port,kNonBlockingUdpFl | kBroadcastUdpFl,_cmRtNetRecv,p,NULL,0,p->udpRecvBufByteCnt,p->udpTimeOutMs) != kOkUdpRC )
  569. {
  570. rc = cmErrMsg(&p->err,kUdpPortFailNetRC,"The UDP port initialization failed.");
  571. goto errLabel;
  572. }
  573. // create the local node
  574. if((rc = _cmRtNetCreateNode(p,nodeLabel, p->rtSubIdx, ipAddr, port, NULL, kLocalNodeNetFl, 0)) != kOkNetRC )
  575. goto errLabel;
  576. // the last created node is always the first node on the list
  577. p->localNode = p->nodes;
  578. p->bcastAddr = cmMemResizeStr(p->bcastAddr,bcastAddr);
  579. // begin listening on the local port
  580. if( cmUdpEnableListen(p->udpH, true ) != kOkUdpRC )
  581. {
  582. rc = cmErrMsg(&p->err,kUdpPortFailNetRC,"The UDP port failed to enter 'listen' mode.");
  583. goto errLabel;
  584. }
  585. errLabel:
  586. return rc;
  587. }
  588. bool cmRtNetIsInitialized( cmRtNetH_t h )
  589. {
  590. if( cmRtNetIsValid(h) == false )
  591. return false;
  592. cmRtNet_t* p = _cmRtNetHandleToPtr(h);
  593. return p->localNode != NULL && cmTextIsNotEmpty(p->bcastAddr);
  594. }
  595. cmRtNetRC_t cmRtNetRegisterEndPoint( cmRtNetH_t h, const cmChar_t* endPtLabel, unsigned endPtId )
  596. {
  597. cmRtNetRC_t rc = kOkNetRC;
  598. cmRtNet_t* p = _cmRtNetHandleToPtr(h);
  599. if( p->localNode == NULL )
  600. return cmErrMsg(&p->err,kLocalNodeNetRC,"Local endpoints may not be added if a local node has not been defined.");
  601. if((rc = _cmRtNetCreateEndpoint(p, p->localNode,endPtLabel,endPtId )) == kOkNetRC )
  602. p->localNode->endPtCnt += 1;
  603. return rc;
  604. }
  605. cmRtNetRC_t cmRtNetFinalize( cmRtNetH_t h )
  606. {
  607. cmRtNet_t* p = _cmRtNetHandleToPtr(h);
  608. _cmRtNetReleaseNodes(p);
  609. return kOkNetRC;
  610. }
  611. cmRtNetRC_t cmRtNetDoSync( cmRtNetH_t h )
  612. {
  613. cmRtNet_t* p = _cmRtNetHandleToPtr(h);
  614. // broadcast 'node' msg
  615. return _cmRtNetSendSyncMsg( p, p->localNode, kHelloSelNetId, p->localNode->label, p->localNode->endPtCnt, cmInvalidIdx );
  616. }
  617. cmRtNetRC_t cmRtNetReceive( cmRtNetH_t h )
  618. {
  619. cmRtNetRC_t rc = kOkNetRC;
  620. cmRtNet_t* p = _cmRtNetHandleToPtr(h);
  621. // Calling this function results in callbacks to _cmRtNetRecv() (above)
  622. if( cmUdpGetAvailData(p->udpH, NULL, NULL, NULL ) != kOkUdpRC )
  623. {
  624. cmErrMsg(&p->err,kUdpPortFailNetRC,"UDP port query failed.");
  625. goto errLabel;
  626. }
  627. errLabel:
  628. return rc;
  629. }
  630. unsigned cmRtNetAddrToNodeIndex( cmRtNetH_t h, const struct sockaddr_in* a )
  631. {
  632. cmRtNet_t* p = _cmRtNetHandleToPtr(h);
  633. return _cmRtNetAddrToNodeIndex(p,a);
  634. }
  635. cmRtNetRC_t cmRtNetEndpointHandle( cmRtNetH_t h, const cmChar_t* nodeLabel, const cmChar_t* endptLabel, cmRtNetEndptH_t* hp )
  636. {
  637. cmRtNetRC_t rc = kOkNetRC;
  638. cmRtNet_t* p = _cmRtNetHandleToPtr(h);
  639. cmRtNetNode_t* np;
  640. cmRtNetEnd_t* ep;
  641. if(( np = _cmRtNetFindNode(p,nodeLabel)) == NULL )
  642. return cmErrMsg(&p->err,kNodeNotFoundNetRC,"The node '%s' was not found.",cmStringNullGuard(nodeLabel));
  643. if(( ep = _cmRtNetFindNodeEnd(np, endptLabel )) == NULL )
  644. return cmErrMsg(&p->err,kEndNotFoundNetRC,"The endpoint '%s' on '%s' on node was not found.",cmStringNullGuard(endptLabel),cmStringNullGuard(nodeLabel));
  645. hp->h = ep;
  646. return rc;
  647. }
  648. bool cmRtNetEndpointIsValid( cmRtNetEndptH_t endPtH )
  649. { return endPtH.h != NULL; }
  650. unsigned cmRtNetEndpointId( cmRtNetEndptH_t endPtH )
  651. {
  652. if( !cmRtNetEndpointIsValid(endPtH) )
  653. return cmInvalidId;
  654. cmRtNetEnd_t* ep = _cmRtNetEndptHandleToPtr( endPtH );
  655. return ep->id;
  656. }
  657. const cmChar_t* cmRtNetEndpointLabel( cmRtNetEndptH_t endPtH )
  658. {
  659. if( !cmRtNetEndpointIsValid(endPtH) )
  660. return NULL;
  661. cmRtNetEnd_t* ep = _cmRtNetEndptHandleToPtr( endPtH );
  662. return ep->label;
  663. }
  664. cmRtNetRC_t _cmRtNetSend( cmRtNet_t* p, unsigned srcEndPtId, const cmRtNetEnd_t* ep, const void* msg, unsigned msgByteCnt )
  665. {
  666. cmRtNetRC_t rc = kOkNetRC;
  667. unsigned hN = sizeof(cmRtNetMsg_t);
  668. unsigned dN = hN + msgByteCnt;
  669. char data[ dN ];
  670. cmRtNetMsg_t* r = (cmRtNetMsg_t*)data;
  671. r->hdr.rtSubIdx = ep->np->rtSubIdx;
  672. r->hdr.selId = kMsgSelRtId;
  673. r->dstEndPtId = ep->id;
  674. r->srcEndPtId = srcEndPtId;
  675. memcpy(data+hN,msg,msgByteCnt);
  676. // ep->np->sockaddr identifies the node on the receiving cmRtNet.
  677. // cmRtNetMsg_t* r.endptId is then used by the receiving cmRtNet to indicate which endpoint on
  678. // the node the incoming message should be associated with.
  679. if( cmUdpSendTo(p->udpH, data, dN, &ep->np->sockaddr ) != kOkUdpRC )
  680. return cmErrMsg(&p->err,kUdpPortFailNetRC,"Send to node:%s endpt:%s failed.\n",cmStringNullGuard(ep->np->label),cmStringNullGuard(ep->label));
  681. return rc;
  682. }
  683. cmRtNetRC_t cmRtNetSend( cmRtNetH_t h, unsigned srcEndPtId, cmRtNetEndptH_t epH, const void* msg, unsigned msgByteCnt )
  684. {
  685. cmRtNet_t* p = _cmRtNetHandleToPtr(h);
  686. cmRtNetEnd_t* ep = _cmRtNetEndptHandleToPtr(epH);
  687. assert( ep != NULL );
  688. return _cmRtNetSend(p,srcEndPtId,ep,msg,msgByteCnt);
  689. }
  690. cmRtNetRC_t cmRtNetSendByLabels( cmRtNetH_t h, unsigned srcEndPtId, const cmChar_t* nodeLabel, const cmChar_t* endptLabel, const void* msg, unsigned msgByteCnt )
  691. {
  692. cmRtNetRC_t rc = kOkNetRC;
  693. cmRtNetEndptH_t epH = cmRtNetEndptNullHandle;
  694. if((rc = cmRtNetEndpointHandle(h,nodeLabel,endptLabel,&epH)) != kOkNetRC )
  695. return rc;
  696. return cmRtNetSend(h,srcEndPtId,epH,msg,msgByteCnt);
  697. }
  698. cmRtNetRC_t cmRtNetSendByIndex( cmRtNetH_t h, unsigned srcEndPtId, unsigned nodeIdx, unsigned endptIdx, const void* msg, unsigned msgByteCnt )
  699. {
  700. cmRtNet_t* p = _cmRtNetHandleToPtr(h);
  701. const cmRtNetEnd_t* ep;
  702. if((ep = _cmRtNetFindEndpt(p, nodeIdx, endptIdx )) == NULL )
  703. return cmErrMsg(&p->err,kEndNotFoundNetRC,"The endpoint at node index %i endpoint index %i was not found.",nodeIdx,endptIdx);
  704. return _cmRtNetSend( p, srcEndPtId, ep, msg, msgByteCnt );
  705. }
  706. bool cmRtNetReportSyncEnable( cmRtNetH_t h, bool enableFl )
  707. {
  708. cmRtNet_t* p = _cmRtNetHandleToPtr(h);
  709. bool fl = cmIsFlag(p->flags,kReportSyncNetFl);
  710. p->flags = cmEnaFlag(p->flags,kReportSyncNetFl,enableFl);
  711. return fl;
  712. }
  713. bool cmRtNetReportSyncIsEnabled( cmRtNetH_t h )
  714. {
  715. cmRtNet_t* p = _cmRtNetHandleToPtr(h);
  716. return cmIsFlag(p->flags,kReportSyncNetFl);
  717. }
  718. void cmRtNetReport( cmRtNetH_t h )
  719. {
  720. cmRtNet_t* p = _cmRtNetHandleToPtr(h);
  721. cmRpt_t* rpt = p->err.rpt;
  722. cmRtNetNode_t* np = p->nodes;
  723. for(; np!=NULL; np=np->link)
  724. {
  725. cmRptPrintf(rpt,"Node: %s ",np->label);
  726. if( np->addr != NULL )
  727. cmRptPrintf(rpt,"%s ",np->addr );
  728. if( cmIsFlag(np->flags,kLocalNodeNetFl) )
  729. cmRptPrintf(rpt,"LOCAL ");
  730. cmRptPrintf(rpt,"%s ",cmStringNullGuard(cmUdpAddrToString(p->udpH,&np->sockaddr)));
  731. if( np->port != kInvalidUdpPortNumber )
  732. cmRptPrintf(rpt,"%i ",np->port );
  733. cmRptPrintf(rpt,"\n");
  734. cmRtNetEnd_t* ep = np->ends;
  735. for(; ep!=NULL; ep=ep->link)
  736. {
  737. cmRptPrintf(rpt," endpt: %i %s\n",ep->id,cmStringNullGuard(ep->label ));
  738. }
  739. }
  740. }
  741. const cmChar_t* cmRtNetLocalNodeLabel( cmRtNetH_t h )
  742. {
  743. cmRtNet_t* p = _cmRtNetHandleToPtr( h );
  744. return p->localNode->label;
  745. }
  746. unsigned cmRtNetRemoteNodeCount( cmRtNetH_t h )
  747. {
  748. cmRtNet_t* p = _cmRtNetHandleToPtr( h );
  749. const cmRtNetNode_t* np = p->nodes;
  750. unsigned n = 0;
  751. for(; np!=NULL; np=np->link)
  752. if( np != p->localNode )
  753. ++n;
  754. return n;
  755. }
  756. const cmChar_t* cmRtNetRemoteNodeLabel( cmRtNetH_t h, unsigned idx )
  757. {
  758. cmRtNet_t* p = _cmRtNetHandleToPtr( h );
  759. const cmRtNetNode_t* np;
  760. if((np = _cmRtNetIndexToRemoteNode( p, idx )) == NULL )
  761. return NULL;
  762. return np->label;
  763. }
  764. unsigned cmRtNetRemoteNodeEndPointCount( cmRtNetH_t h, unsigned nodeIdx )
  765. {
  766. const cmRtNetNode_t* np;
  767. const cmRtNetEnd_t* ep;
  768. cmRtNet_t* p = _cmRtNetHandleToPtr( h );
  769. unsigned n = 0;
  770. if((np = _cmRtNetIndexToRemoteNode( p, nodeIdx )) == NULL )
  771. return 0;
  772. for(ep=np->ends; ep!=NULL; ep=ep->link)
  773. ++n;
  774. return n;
  775. }
  776. cmRtNetRC_t cmRtNetRemoteNodeEndPoint(
  777. cmRtNetH_t h,
  778. unsigned nodeIdx,
  779. unsigned epIdx,
  780. const cmChar_t** labelRef,
  781. unsigned* idRef,
  782. unsigned* rsiRef )
  783. {
  784. const cmRtNetEnd_t* ep;
  785. cmRtNet_t* p = _cmRtNetHandleToPtr( h );
  786. if(( ep = _cmRtNetFindEndpt(p, nodeIdx, epIdx )) == NULL )
  787. {
  788. *labelRef = NULL;
  789. *idRef = cmInvalidId;
  790. *rsiRef = cmInvalidIdx;
  791. return kEndNotFoundNetRC;
  792. }
  793. *labelRef = ep->label;
  794. *idRef = ep->id;
  795. *rsiRef = ep->np->rtSubIdx;
  796. return kOkNetRC;
  797. }
  798. //==========================================================================
  799. #include "cmThread.h"
  800. typedef struct
  801. {
  802. cmThreadH_t thH;
  803. cmRtNetH_t netH;
  804. unsigned msgVal;
  805. } _cmRtNetTest_t;
  806. // This function is called within the context of cmRtNetReceive().
  807. void _cmRtNetTestRecv( void* cbArg, const char* data, unsigned dataByteCnt, const struct sockaddr_in* fromAddr )
  808. {
  809. //_cmRtNetTest_t* p = (_cmRtNetTest_t*)cbArg;
  810. cmRtNetMsg_t* r = (cmRtNetMsg_t*)data;
  811. unsigned i = *(unsigned*)(data + sizeof(cmRtNetMsg_t));
  812. printf("rtSubIdx:%i endptId:%i %i\n",r->hdr.rtSubIdx,r->dstEndPtId,i);
  813. }
  814. bool _cmRtNetTestThreadFunc(void* param)
  815. {
  816. _cmRtNetTest_t* p = (_cmRtNetTest_t*)param;
  817. if( cmRtNetIsValid(p->netH) )
  818. {
  819. cmRtNetReceive(p->netH);
  820. }
  821. cmSleepMs(40);
  822. return true;
  823. }
  824. void cmRtNetTest( cmCtx_t* ctx, bool mstrFl )
  825. {
  826. char c;
  827. _cmRtNetTest_t t;
  828. const unsigned rtSubIdx = 0;
  829. cmUdpPort_t port = 5876;
  830. _cmRtNetTest_t* p = &t;
  831. cmRtNetRC_t rc = kOkNetRC;
  832. const cmChar_t* localHostStr = mstrFl ? "master" : "slave";
  833. const cmChar_t* localEndpStr = mstrFl ? "master_ep" : "slave_ep";
  834. const cmChar_t* remoteHostStr = !mstrFl ? "master" : "slave";
  835. const cmChar_t* remoteEndpStr = !mstrFl ? "master_ep" : "slave_ep";
  836. const cmChar_t* bcastAddr = "192.168.15.255";
  837. cmRtNetEndptH_t eH = cmRtNetEndptNullHandle;
  838. unsigned srcEndPtId = cmInvalidId;
  839. memset(&t,0,sizeof(t));
  840. if( cmThreadCreate(&p->thH,_cmRtNetTestThreadFunc,p,&ctx->rpt) != kOkThRC )
  841. goto errLabel;
  842. if((rc = cmRtNetAlloc(ctx,&p->netH,rtSubIdx,_cmRtNetTestRecv,p)) != kOkNetRC )
  843. goto errLabel;
  844. cmRtNetReportSyncEnable(p->netH,true); // enable sync. protocol reporting
  845. if((rc = cmRtNetInitialize(p->netH, bcastAddr, localHostStr, NULL, port )) != kOkNetRC)
  846. goto errLabel;
  847. if((rc = cmRtNetRegisterEndPoint(p->netH,localEndpStr, 0 )) != kOkNetRC )
  848. goto errLabel;
  849. if((rc = cmRtNetEndpointHandle(p->netH, localHostStr, localEndpStr, &eH )) != kOkNetRC )
  850. goto errLabel;
  851. if((srcEndPtId = cmRtNetEndpointId(eH)) == cmInvalidIdx )
  852. goto errLabel;
  853. if( cmThreadPause(p->thH,0) != kOkThRC )
  854. goto errLabel;
  855. cmRptPrintf(&ctx->rpt,"%s t=transmit s=sync r=report q=quit\n", localHostStr );
  856. while( (c=getchar()) != 'q' )
  857. {
  858. switch(c)
  859. {
  860. case 'r':
  861. cmRtNetReport(p->netH);
  862. break;
  863. case 's':
  864. cmRtNetDoSync(p->netH);
  865. break;
  866. case 't':
  867. {
  868. if( cmRtNetSendByLabels(p->netH, srcEndPtId, remoteHostStr, remoteEndpStr, &p->msgVal, sizeof(p->msgVal)) == kOkNetRC )
  869. p->msgVal += 1;
  870. }
  871. break;
  872. }
  873. }
  874. errLabel:
  875. cmThreadDestroy(&p->thH);
  876. cmRtNetFree(&p->netH);
  877. return;
  878. }