libcm is a C development framework with an emphasis on audio signal processing applications.
Você não pode selecionar mais de 25 tópicos Os tópicos devem começar com uma letra ou um número, podem incluir traços ('-') e podem ter até 35 caracteres.

cmRtNet.c 24KB

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