libcm is a C development framework with an emphasis on audio signal processing applications.
You can not select more than 25 topics Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.

cmRtNet.c 23KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695696697698699700701702703704705706707708709710711712713714715716717718719720721722723724725726727728729730731732733734735736737738739740741742743744745746747748749750751752753754755756757758759760761762763764765766767768769770771772773774775776777778779780781782783784785786787788789790791792793794795796797798799800801802803804805806807808809810811812813814815816817818819820821822823824825826827828829830831832833834835836837838839840841842843844845846847848849850851852853854855856857858859860861862863864865866867868869870871872873874875876877878879880881882883884885886887888889890891892893894895896897898899900901902903904905906907908909910911912913914915916917918
  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 "cmUdpPort.h"
  8. #include "cmRtNet.h"
  9. #include "cmTime.h"
  10. #include "cmRtSysMsg.h"
  11. enum
  12. {
  13. kLocalNetFl = 0x01,
  14. kSockAddrNetFl = 0x02,
  15. kRegNodeNetFl = 0x04,
  16. kRecvNodeNetFl = 0x08
  17. };
  18. typedef enum
  19. {
  20. kSendHelloStNetId = 0,
  21. kWaitHelloAckStNetId,
  22. kSendEndpointStNetId,
  23. kWaitEndpointAckStNetId,
  24. kDoneStNetId,
  25. kErrorStNetId,
  26. kInvalidStNetId,
  27. } cmRtNetNodeState_t;
  28. typedef enum
  29. {
  30. kHelloSelNetId,
  31. kHelloAckSelNetId,
  32. kEndpointSelNetId,
  33. kEndpointAckSelNetId,
  34. kDoneSelNetId
  35. } cmRtNetSelId_t;
  36. typedef struct cmRtNetEnd_str
  37. {
  38. cmChar_t* endPtLabel;
  39. unsigned endPtId;
  40. struct cmRtNetEnd_str* link;
  41. } cmRtNetEnd_t;
  42. typedef struct cmRtNetNode_str
  43. {
  44. cmChar_t* label;
  45. struct sockaddr_in sockaddr;
  46. cmChar_t* addr;
  47. cmUdpPort_t port;
  48. unsigned flags;
  49. cmRtNetNodeState_t state;
  50. unsigned epIdx; // tracks the next endpoint to send during sync-mode
  51. cmTimeSpec_t lastSendTime;
  52. cmRtNetEnd_t* ends;
  53. struct cmRtNetNode_str* link;
  54. } cmRtNetNode_t;
  55. typedef struct
  56. {
  57. cmErr_t err;
  58. cmUdpH_t udpH;
  59. cmUdpCallback_t cbFunc;
  60. void* cbArg;
  61. cmRtNetNode_t* nodes;
  62. cmRtNetNode_t* localNode;
  63. bool syncModeFl;
  64. bool masterFl;
  65. unsigned udpRecvBufByteCnt;
  66. unsigned udpTimeOutMs;
  67. unsigned interSyncSendTimeMs;
  68. } cmRtNet_t;
  69. typedef struct
  70. {
  71. cmRtSysMsgHdr_t hdr;
  72. cmRtNetSelId_t selId;
  73. const cmChar_t* endPtLabel;
  74. unsigned endPtId;
  75. } cmRtNetSyncMsg_t;
  76. cmRtNetH_t cmRtNetNullHandle = cmSTATIC_NULL_HANDLE;
  77. cmRtNet_t* _cmRtNetHandleToPtr( cmRtNetH_t h )
  78. {
  79. cmRtNet_t* p = (cmRtNet_t*)h.h;
  80. assert( p != NULL );
  81. return p;
  82. }
  83. void _cmRtNetVRpt( cmRtNet_t* p, const cmChar_t* fmt, va_list vl )
  84. {
  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( cmIsFlag(np->flags,kSockAddrNetFl) && 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->endPtLabel);
  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. // we should never release the local node via this function
  143. assert( np != p->localNode );
  144. cmRtNetNode_t* cnp = p->nodes;
  145. cmRtNetNode_t* pnp = NULL;
  146. while( cnp != NULL )
  147. {
  148. cmRtNetNode_t* nnp = cnp->link;
  149. if( np == cnp )
  150. {
  151. if( pnp == NULL )
  152. p->nodes = np->link;
  153. else
  154. pnp->link = np->link;
  155. _cmRtNetFreeNode(np);
  156. return kOkNetRC;
  157. }
  158. pnp = np;
  159. cnp = nnp;
  160. }
  161. assert(0);
  162. return cmErrMsg(&p->err,kNodeNotFoundNetRC,"Node to release not found.");
  163. }
  164. cmRtNetRC_t _cmRtNetCreateNode( cmRtNet_t* p, const cmChar_t* label, const cmChar_t* addr, cmUdpPort_t port, const struct sockaddr_in* saddr, unsigned flags )
  165. {
  166. cmRtNetRC_t rc = kOkNetRC;
  167. cmRtNetNode_t* np;
  168. if( label == NULL )
  169. return cmErrMsg(&p->err,kInvalidLabelNetRC,"A null or blank node label was encountered.");
  170. if((np = _cmRtNetFindNode(p,label)) != NULL )
  171. return cmErrMsg(&p->err,kDuplLabelNetRC,"The node label '%s' is already in use.",cmStringNullGuard(label));
  172. bool localNodeFl = addr==NULL && saddr==NULL;
  173. if( localNodeFl && p->localNode != NULL )
  174. return cmErrMsg(&p->err,kDuplLocalNetRC,"The local node '%s' has already been set.",cmStringNullGuard(p->localNode->label));
  175. np = cmMemAllocZ(cmRtNetNode_t,1);
  176. np->label = cmMemAllocStr(label);
  177. np->addr = addr==NULL ? NULL : cmMemAllocStr(addr);
  178. np->port = port;
  179. np->flags = cmEnaFlag(flags,kLocalNetFl,localNodeFl);
  180. np->link = p->nodes;
  181. p->nodes = np;
  182. if( localNodeFl )
  183. p->localNode = np;
  184. if( saddr != NULL )
  185. np->sockaddr = *saddr;
  186. else
  187. {
  188. if( cmUdpInitAddr(p->udpH, np->addr, np->port, &np->sockaddr ) != kOkUdpRC )
  189. {
  190. rc = cmErrMsg(&p->err,kUdpPortFailNetRC,"IP::port to socket address conversion failed.");
  191. goto errLabel;
  192. }
  193. }
  194. np->flags = cmSetFlag(np->flags,kSockAddrNetFl);
  195. errLabel:
  196. return rc;
  197. }
  198. cmRtNetEnd_t* _cmRtNetFindNodeEnd(cmRtNetNode_t* np, const cmChar_t* endPtLabel )
  199. {
  200. cmRtNetEnd_t* ep = np->ends;
  201. for(; ep!=NULL; ep=ep->link)
  202. if( strcmp(ep->endPtLabel,endPtLabel)==0 )
  203. return ep;
  204. return NULL;
  205. }
  206. cmRtNetEnd_t* _cmRtNetIndexToEndpoint( cmRtNet_t* p, cmRtNetNode_t* np, unsigned endIndex )
  207. {
  208. cmRtNetEnd_t* ep = np->ends;
  209. unsigned i;
  210. for(i=0; ep!=NULL; ep=ep->link)
  211. {
  212. if( i == endIndex )
  213. return ep;
  214. ++i;
  215. }
  216. return NULL;
  217. }
  218. cmRtNetRC_t _cmRtNetCreateEndpoint( cmRtNet_t* p, cmRtNetNode_t* np, const cmChar_t* endPtLabel, unsigned endPtId )
  219. {
  220. if( endPtLabel == NULL )
  221. return cmErrMsg(&p->err,kInvalidLabelNetRC,"A null or blank node label was encountered.");
  222. if( _cmRtNetFindNodeEnd( np, endPtLabel) != NULL)
  223. return cmErrMsg(&p->err,kDuplEndNetRC,"A duplicate endpoint ('%s') was encountered on node '%s'.",endPtLabel,np->label);
  224. cmRtNetRC_t rc = kOkNetRC;
  225. cmRtNetEnd_t* ep = cmMemAllocZ(cmRtNetEnd_t,1);
  226. ep->endPtLabel = cmMemAllocStr(endPtLabel);
  227. ep->endPtId = endPtId;
  228. ep->link = np->ends;
  229. np->ends = ep;
  230. return rc;
  231. }
  232. unsigned _cmRtNetSyncMsgSerialByteCount( const cmRtNetSyncMsg_t* m )
  233. { return sizeof(cmRtNetSyncMsg_t) + (m->endPtLabel==NULL ? 1 : strlen(m->endPtLabel) + 1); }
  234. cmRtNetRC_t _cmRtNetSerializeSyncMsg( cmRtNet_t* p, const cmRtNetSyncMsg_t* m, void* buf, unsigned n )
  235. {
  236. unsigned bn = _cmRtNetSyncMsgSerialByteCount(m);
  237. char* b = (char*)buf;
  238. if( bn > n )
  239. return cmErrMsg(&p->err,kBufToSmallNetRC,"Serialize buffer too small.");
  240. memcpy(b,m,sizeof(*m));
  241. strcpy(b + sizeof(*m),m->endPtLabel==NULL ? "" : m->endPtLabel);
  242. return kOkNetRC;
  243. }
  244. cmRtNetRC_t _cmRtNetDeserializeSyncMsg( const void* buf, unsigned n, cmRtNetSyncMsg_t* m )
  245. {
  246. assert( n > sizeof(*m));
  247. memcpy(m,buf,sizeof(*m));
  248. const cmRtNetSyncMsg_t* mp = (const cmRtNetSyncMsg_t*)buf;
  249. const cmChar_t* s = (const cmChar_t*)(mp+1);
  250. m->endPtLabel = cmMemAllocStr(s);
  251. return kOkNetRC;
  252. }
  253. cmRtNetRC_t _cmRtNetSendSyncMsg( cmRtNet_t* p, cmRtNetNode_t* np, cmRtNetSelId_t selId, const cmChar_t* endPtLabel, unsigned endPtId, cmRtNetNodeState_t nextStId )
  254. {
  255. cmRtNetSyncMsg_t m;
  256. cmRtNetRC_t rc = kOkNetRC;
  257. cmUdpRC_t udpRC = kOkUdpRC;
  258. m.hdr.rtSubIdx = cmInvalidIdx;
  259. m.hdr.selId = kNetSyncSelRtId;
  260. m.selId = selId;
  261. m.endPtLabel = endPtLabel;
  262. m.endPtId = endPtId;
  263. // determine size of msg to send
  264. unsigned n = _cmRtNetSyncMsgSerialByteCount(&m);
  265. cmChar_t buf[n];
  266. // serialize msg into buf[]
  267. if((rc = _cmRtNetSerializeSyncMsg(p,&m,buf,n)) != kOkNetRC )
  268. return rc;
  269. // store this nodes current sync state
  270. cmRtNetNodeState_t orgState = np->state;
  271. if( nextStId != kInvalidStNetId )
  272. np->state = nextStId;
  273. // send the msg
  274. if( cmIsFlag(np->flags,kSockAddrNetFl) == false )
  275. udpRC = cmUdpSend2(p->udpH, buf, n, np->addr, np->port );
  276. else
  277. udpRC = cmUdpSendTo(p->udpH, buf, n, &np->sockaddr );
  278. // check for send errors
  279. if( udpRC != kOkUdpRC )
  280. {
  281. rc = cmErrMsg(&p->err,kUdpPortFailNetRC,"Sync msg. send on UDP port failed.");
  282. np->state = orgState; // restore node state so we can try again
  283. }
  284. else
  285. {
  286. // record the last send time
  287. cmTimeGet(&np->lastSendTime);
  288. }
  289. return rc;
  290. }
  291. cmRtNetRC_t _cmRtNetFree( cmRtNet_t* p )
  292. {
  293. cmRtNetRC_t rc = kOkNetRC;
  294. if( cmUdpFree(&p->udpH) != kOkUdpRC )
  295. cmErrMsg(&p->err,kUdpPortFailNetRC,"UDP Port free failed.");
  296. _cmRtNetReleaseNodes(p);
  297. cmMemFree(p);
  298. return rc;
  299. }
  300. cmRtNetRC_t cmRtNetAlloc( cmCtx_t* ctx, cmRtNetH_t* hp, cmUdpCallback_t cbFunc, void* cbArg )
  301. {
  302. cmRtNetRC_t rc;
  303. if((rc = cmRtNetFree(hp)) != kOkNetRC )
  304. return rc;
  305. cmRtNet_t* p = cmMemAllocZ(cmRtNet_t,1);
  306. cmErrSetup(&p->err,&ctx->rpt,"cmRtNet");
  307. // allocate the UDP port
  308. if(cmUdpAlloc(ctx,&p->udpH) != kOkUdpRC )
  309. {
  310. cmErrMsg(&p->err,kUdpPortFailNetRC,"UDP Port allocate failed.");
  311. goto errLabel;
  312. }
  313. p->udpTimeOutMs = 50;
  314. p->udpRecvBufByteCnt = 8192;
  315. p->interSyncSendTimeMs = 10;
  316. p->cbFunc = cbFunc;
  317. p->cbArg = cbArg;
  318. hp->h = p;
  319. errLabel:
  320. if(rc != kOkNetRC )
  321. _cmRtNetFree(p);
  322. return rc;
  323. }
  324. cmRtNetRC_t cmRtNetFree( cmRtNetH_t* hp )
  325. {
  326. cmRtNetRC_t rc = kOkNetRC;
  327. if( hp==NULL || cmRtNetIsValid(*hp)==false )
  328. return rc;
  329. cmRtNet_t* p = _cmRtNetHandleToPtr(*hp);
  330. if((rc = _cmRtNetFree(p)) != kOkNetRC )
  331. return rc;
  332. hp->h = NULL;
  333. return rc;
  334. }
  335. bool cmRtNetIsValid( cmRtNetH_t h )
  336. { return h.h !=NULL; }
  337. cmUdpH_t cmRtNetUdpPortHandle( cmRtNetH_t h )
  338. {
  339. cmRtNet_t* p = _cmRtNetHandleToPtr(h);
  340. return p->udpH;
  341. }
  342. cmRtNetRC_t cmRtNetCreateNode( cmRtNetH_t h, const cmChar_t* nodeLabel, const cmChar_t* ipAddr, cmUdpPort_t port )
  343. {
  344. cmRtNet_t* p = _cmRtNetHandleToPtr(h);
  345. cmRtNetRC_t rc;
  346. // create a node
  347. if((rc = _cmRtNetCreateNode(p,nodeLabel,ipAddr, port, NULL, kRegNodeNetFl)) != kOkNetRC )
  348. return rc;
  349. // if this is not the local node
  350. if( ipAddr != NULL )
  351. return rc;
  352. // if this is the local node then initialze the local socket
  353. if( cmUdpInit(p->udpH,port,kNonBlockingUdpFl,p->cbFunc,p->cbArg,NULL,0,p->udpRecvBufByteCnt,p->udpTimeOutMs) != kOkUdpRC )
  354. {
  355. rc = cmErrMsg(&p->err,kUdpPortFailNetRC,"The UDP port initialization failed.");
  356. goto errLabel;
  357. }
  358. // begin listening on the local port
  359. if( cmUdpEnableListen(p->udpH, true ) != kOkUdpRC )
  360. {
  361. rc = cmErrMsg(&p->err,kUdpPortFailNetRC,"The UDP port failed to enter 'listen' mode.");
  362. goto errLabel;
  363. }
  364. errLabel:
  365. return rc;
  366. }
  367. cmRtNetRC_t cmRtNetRegisterEndPoint( cmRtNetH_t h, const cmChar_t* endPtLabel, unsigned endPtId )
  368. {
  369. cmRtNet_t* p = _cmRtNetHandleToPtr(h);
  370. if( p->localNode == NULL )
  371. return cmErrMsg(&p->err,kLocalNodeNetRC,"Local endpoints may not be added if a local node has not been defined.");
  372. return _cmRtNetCreateEndpoint(p, p->localNode,endPtLabel,endPtId );
  373. }
  374. cmRtNetRC_t cmRtNetClearAll( cmRtNetH_t h )
  375. {
  376. cmRtNet_t* p = _cmRtNetHandleToPtr(h);
  377. _cmRtNetReleaseNodes(p);
  378. return kOkNetRC;
  379. }
  380. cmRtNetRC_t cmRtNetBeginSyncMode( cmRtNetH_t h )
  381. {
  382. cmRtNetRC_t rc = kOkNetRC;
  383. cmRtNet_t* p = _cmRtNetHandleToPtr(h);
  384. p->syncModeFl = true;
  385. p->masterFl = true;
  386. return rc;
  387. }
  388. bool cmRtNetIsInSyncMode( cmRtNetH_t h )
  389. {
  390. cmRtNet_t* p = _cmRtNetHandleToPtr(h);
  391. return p->syncModeFl;
  392. }
  393. // Used by slaves to send the master an 'ack' msg.
  394. cmRtNetRC_t _cmRtNetSendAck( cmRtNet_t* p, cmRtNetSelId_t ackSelId, const struct sockaddr_in* saddr )
  395. {
  396. cmRtNetNode_t* np;
  397. if((np = _cmRtNetFindNodeFromSockAddr(p,saddr)) == NULL )
  398. return cmErrMsg(&p->err,kNodeNotFoundNetRC,"The net node associated with an ack cmd was not found. Ack not sent.");
  399. return _cmRtNetSendSyncMsg(p,np,ackSelId,NULL,cmInvalidId,kInvalidStNetId);
  400. }
  401. // Used by master to update state upon receipt of 'ack' msg
  402. cmRtNetRC_t _cmRtNetRecvAck( cmRtNet_t* p, const struct sockaddr_in* fromAddr, cmRtNetNodeState_t expectedState, cmRtNetNodeState_t nextState )
  403. {
  404. cmRtNetNode_t* np;
  405. cmRtNetRC_t rc = kOkNetRC;
  406. if((np = _cmRtNetFindNodeFromSockAddr(p,fromAddr)) == NULL )
  407. {
  408. rc = cmErrMsg(&p->err,kNodeNotFoundNetRC,"The net node associated with a ack receive was not found.");
  409. goto errLabel;
  410. }
  411. if( np->state != expectedState )
  412. {
  413. rc = cmErrMsg(&p->err,kNodeStateErrNetRC,"Node '%s' expected in state %i was in state %i.",cmStringNullGuard(np->label),kWaitHelloAckStNetId,np->state);
  414. np->state = kErrorStNetId;
  415. goto errLabel;
  416. }
  417. np->state = nextState;
  418. // if we are about to send another endpoint - incr the endpoint index
  419. if( nextState == kSendEndpointStNetId )
  420. np->epIdx += 1;
  421. errLabel:
  422. return rc;
  423. }
  424. cmRtNetRC_t cmRtNetSyncModeRecv( cmRtNetH_t h, const char* data, unsigned dataByteCnt, const struct sockaddr_in* fromAddr )
  425. {
  426. cmRtNet_t* p = _cmRtNetHandleToPtr(h);
  427. cmRtNetRC_t rc = kOkNetRC;
  428. cmRtNetNode_t* np = NULL;
  429. cmRtNetSyncMsg_t m;
  430. m.endPtLabel = NULL;
  431. assert( cmRtNetIsSyncModeMsg(data,dataByteCnt));
  432. if( _cmRtNetDeserializeSyncMsg(data,dataByteCnt,&m) != kOkNetRC )
  433. {
  434. rc = cmErrMsg(&p->err,rc,"Net sync. receive failed due to deserialize fail.");
  435. goto errLabel;
  436. }
  437. assert( m.hdr.selId == kNetSyncSelRtId );
  438. switch( m.selId )
  439. {
  440. case kHelloSelNetId: // slave response
  441. {
  442. _cmRtNetRpt(p,"rcv hello\n");
  443. // attempt to locate the remote node which sent the endpoint
  444. if((np = _cmRtNetFindNodeFromSockAddr(p,fromAddr)) != NULL )
  445. {
  446. // delete the existing node because we are about to get new info. about it.
  447. if((rc = _cmRtNetReleaseNode(p,np )) != kOkNetRC )
  448. goto errLabel;
  449. }
  450. // create a node proxy to represent the remote node
  451. if(( rc = _cmRtNetCreateNode(p,m.endPtLabel,NULL,0,fromAddr,kRecvNodeNetFl)) != kOkNetRC )
  452. goto errLabel;
  453. // send an ackknowledgement of the 'hello' msg
  454. rc = _cmRtNetSendAck(p,kHelloAckSelNetId,fromAddr);
  455. }
  456. break;
  457. case kEndpointSelNetId: // slave response
  458. {
  459. cmRtNetEnd_t* ep;
  460. _cmRtNetRpt(p,"rcv endpoint\n");
  461. // locate the remote node which sent the endpoint
  462. if((np = _cmRtNetFindNodeFromSockAddr(p,fromAddr)) == NULL )
  463. {
  464. rc = cmErrMsg(&p->err,kNodeNotFoundNetRC,"The net node associated with an endpoint receive was not found.");
  465. goto errLabel;
  466. }
  467. // attempt to find the end point
  468. if((ep = _cmRtNetFindNodeEnd(np,m.endPtLabel)) != NULL )
  469. ep->endPtId = m.endPtId; // the endpoint was found update the endPtId
  470. else
  471. {
  472. // create a local proxy for the endpoint
  473. if((rc = _cmRtNetCreateEndpoint(p,np,m.endPtLabel,m.endPtId)) != kOkNetRC )
  474. goto errLabel;
  475. }
  476. // ack. the endpoint msg
  477. rc = _cmRtNetSendAck(p,kEndpointAckSelNetId,fromAddr);
  478. }
  479. break;
  480. case kDoneSelNetId:
  481. {
  482. _cmRtNetRpt(p,"rcv done\n");
  483. if( p->masterFl==false )
  484. p->syncModeFl = true;
  485. }
  486. break;
  487. case kHelloAckSelNetId: // master response
  488. {
  489. assert( p->syncModeFl );
  490. _cmRtNetRpt(p,"rcv hello ack\n");
  491. rc = _cmRtNetRecvAck(p,fromAddr,kWaitHelloAckStNetId,kSendEndpointStNetId);
  492. }
  493. break;
  494. case kEndpointAckSelNetId: // master response
  495. {
  496. assert( p->syncModeFl );
  497. _cmRtNetRpt(p,"rcv endpoint ack\n");
  498. rc = _cmRtNetRecvAck(p,fromAddr,kWaitEndpointAckStNetId,kSendEndpointStNetId);
  499. }
  500. break;
  501. default:
  502. break;
  503. }
  504. errLabel:
  505. cmMemFree((cmChar_t*)m.endPtLabel);
  506. return rc;
  507. }
  508. cmRtNetRC_t _cmRtNetSendNodeSync( cmRtNet_t* p, cmRtNetNode_t* np )
  509. {
  510. cmRtNetRC_t rc = kOkNetRC;
  511. switch( np->state )
  512. {
  513. case kSendHelloStNetId:
  514. {
  515. np->epIdx = -1;
  516. // send a 'hello' to this remote node
  517. if((rc = _cmRtNetSendSyncMsg(p,np,kHelloSelNetId,p->localNode->label, cmInvalidId, kWaitHelloAckStNetId )) != kOkNetRC )
  518. rc = cmErrMsg(&p->err,rc,"Send 'hello' to %s:%s:%i failed.",cmStringNullGuard(np->label),cmStringNullGuard(np->addr),np->port);
  519. else
  520. _cmRtNetRpt(p,"%s sent hello\n",cmStringNullGuard(np->label));
  521. }
  522. break;
  523. case kSendEndpointStNetId:
  524. {
  525. cmRtNetEnd_t* ep;
  526. // if all of the endpoints have been sent to this node ...
  527. if((ep = _cmRtNetIndexToEndpoint(p,p->localNode,np->epIdx)) == NULL )
  528. {
  529. // notify the remote node that all endpoints have been sent
  530. if((rc = _cmRtNetSendSyncMsg(p,np,kDoneSelNetId,p->localNode->label,cmInvalidId, kDoneStNetId )) != kOkNetRC )
  531. rc = cmErrMsg(&p->err,rc,"Send 'done' to %s:%s:%i failed.",cmStringNullGuard(np->label),cmStringNullGuard(np->addr),np->port);
  532. else
  533. _cmRtNetRpt(p,"Node %s done.\n",cmStringNullGuard(np->label));
  534. }
  535. else
  536. {
  537. // send an endpoint to this node
  538. if((rc = _cmRtNetSendSyncMsg(p,np,kEndpointSelNetId,ep->endPtLabel, ep->endPtId, kWaitEndpointAckStNetId )) != kOkNetRC )
  539. rc = cmErrMsg(&p->err,rc,"Endpoint (%s index:%i) transmission to %s:%s:%i failed.",cmStringNullGuard(ep->endPtLabel),cmStringNullGuard(np->label),cmStringNullGuard(np->addr),np->port);
  540. else
  541. _cmRtNetRpt(p,"%s sent endpoint %s\n",cmStringNullGuard(np->label),cmStringNullGuard(ep->endPtLabel));
  542. }
  543. }
  544. break;
  545. case kWaitHelloAckStNetId:
  546. case kWaitEndpointAckStNetId:
  547. {
  548. cmTimeSpec_t t;
  549. cmTimeGet(&t);
  550. unsigned twentySecs = 20000000;
  551. if( cmTimeElapsedMicros(&np->lastSendTime,&t) > twentySecs)
  552. {
  553. const cmChar_t* ackStr = np->state==kWaitHelloAckStNetId ? "hello" : "endpoint";
  554. rc = cmErrMsg(&p->err,kTimeOutErrNetRC,"The node %s:%s:%i did not give a '%s' acknowledge.",cmStringNullGuard(np->label),cmStringNullGuard(np->addr),np->port,ackStr);
  555. }
  556. }
  557. break;
  558. default:
  559. break;
  560. }
  561. // if an error occurred put the node into an error state
  562. if( rc != kOkNetRC )
  563. np->state = kErrorStNetId;
  564. return rc;
  565. }
  566. cmRtNetRC_t cmRtNetSyncModeSend( cmRtNetH_t h )
  567. {
  568. cmRtNetRC_t rc = kOkNetRC;
  569. cmRtNet_t* p = _cmRtNetHandleToPtr(h);
  570. if( p->syncModeFl == false )
  571. return rc;
  572. unsigned activeCnt = 0;
  573. cmRtNetNode_t* np = p->nodes;
  574. for(; np != NULL; np=np->link )
  575. {
  576. bool fl = (p->masterFl && cmIsFlag(np->flags,kRegNodeNetFl)) || (p->masterFl==false && cmIsFlag(np->flags,kRecvNodeNetFl));
  577. if( fl && np != p->localNode && np->state != kDoneStNetId && np->state != kErrorStNetId )
  578. {
  579. _cmRtNetSendNodeSync(p,np);
  580. activeCnt += 1;
  581. }
  582. }
  583. if( activeCnt == 0 )
  584. {
  585. p->syncModeFl = false;
  586. _cmRtNetRpt(p,"sync mode complete.\n");
  587. }
  588. return rc;
  589. }
  590. cmRtNetRC_t cmRtNetReceive( cmRtNetH_t h )
  591. {
  592. cmRtNetRC_t rc = kOkNetRC;
  593. cmRtNet_t* p = _cmRtNetHandleToPtr(h);
  594. if( cmUdpGetAvailData(p->udpH, NULL, NULL, NULL ) != kOkUdpRC )
  595. {
  596. cmErrMsg(&p->err,kUdpPortFailNetRC,"UDP port query failed.");
  597. goto errLabel;
  598. }
  599. errLabel:
  600. return rc;
  601. }
  602. bool cmRtNetIsSyncModeMsg( const void* data, unsigned dataByteCnt )
  603. {
  604. cmRtNetSyncMsg_t* m = (cmRtNetSyncMsg_t*)data;
  605. return dataByteCnt >= sizeof(cmRtNetSyncMsg_t) && m->hdr.selId == kNetSyncSelRtId;
  606. }
  607. unsigned cmRtNetEndPointIndex( cmRtNetH_t h, const cmChar_t* nodeLabel, const cmChar_t* endPtLabel )
  608. {
  609. //cmRtNet_t* p = _cmRtNetHandleToPtr(h);
  610. return cmInvalidIdx;
  611. }
  612. cmRtNetRC_t cmRtNetSend( cmRtNetH_t h, unsigned endPointIndex, const void* msg, unsigned msgByteCnt )
  613. {
  614. cmRtNetRC_t rc = kOkNetRC;
  615. //cmRtNet_t* p = _cmRtNetHandleToPtr(h);
  616. return rc;
  617. }
  618. void cmRtNetReport( cmRtNetH_t h )
  619. {
  620. cmRtNet_t* p = _cmRtNetHandleToPtr(h);
  621. cmRpt_t* rpt = p->err.rpt;
  622. cmRptPrintf(rpt,"Sync Mode:%s\n",p->syncModeFl ? "ON" : "OFF");
  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,kLocalNetFl) )
  630. cmRptPrintf(rpt,"LOCAL ");
  631. if( cmIsFlag(np->flags,kSockAddrNetFl) )
  632. cmRptPrintf(rpt,"%s ",cmStringNullGuard(cmUdpAddrToString(p->udpH,&np->sockaddr)));
  633. if( np->port != cmInvalidId )
  634. cmRptPrintf(rpt,"%i ",np->port );
  635. cmRptPrintf(rpt,"\n");
  636. cmRtNetEnd_t* ep = np->ends;
  637. for(; ep!=NULL; ep=ep->link)
  638. {
  639. cmRptPrintf(rpt," endpt: %i %s\n",ep->endPtId,cmStringNullGuard(ep->endPtLabel));
  640. }
  641. }
  642. }
  643. //==========================================================================
  644. #include "cmThread.h"
  645. typedef struct
  646. {
  647. cmThreadH_t thH;
  648. cmRtNetH_t netH;
  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. if( cmRtNetIsSyncModeMsg(data,dataByteCnt))
  654. cmRtNetSyncModeRecv(p->netH,data,dataByteCnt,fromAddr);
  655. }
  656. bool _cmRtNetTestThreadFunc(void* param)
  657. {
  658. _cmRtNetTest_t* p = (_cmRtNetTest_t*)param;
  659. if( cmRtNetIsValid(p->netH) )
  660. {
  661. if( cmRtNetIsInSyncMode(p->netH) )
  662. cmRtNetSyncModeSend(p->netH);
  663. cmRtNetReceive(p->netH);
  664. }
  665. cmSleepMs(40);
  666. return true;
  667. }
  668. void cmRtNetTest( cmCtx_t* ctx, bool mstrFl )
  669. {
  670. char c;
  671. _cmRtNetTest_t t;
  672. cmUdpPort_t port = 5876;
  673. _cmRtNetTest_t* p = &t;
  674. cmRtNetRC_t rc = kOkNetRC;
  675. unsigned hn = cmUdpHostNameMaxCharCount();
  676. cmChar_t hostNameStr[ hn ];
  677. memset(&t,0,sizeof(t));
  678. if( cmUdpHostName(hostNameStr,hn) != kOkUdpRC )
  679. goto errLabel;
  680. if( cmThreadCreate(&p->thH,_cmRtNetTestThreadFunc,p,&ctx->rpt) != kOkThRC )
  681. goto errLabel;
  682. if((rc = cmRtNetAlloc(ctx,&p->netH,_cmRtNetTestRecv,p)) != kOkNetRC )
  683. goto errLabel;
  684. if((rc = cmRtNetCreateNode(p->netH, hostNameStr, NULL, port )) != kOkNetRC)
  685. goto errLabel;
  686. if( mstrFl )
  687. {
  688. if((rc = cmRtNetCreateNode(p->netH,"whirl", "192.168.15.109", port )) != kOkNetRC )
  689. goto errLabel;
  690. if((rc = cmRtNetRegisterEndPoint(p->netH,"thunk_ep0", 0 )) != kOkNetRC )
  691. goto errLabel;
  692. if(( rc = cmRtNetBeginSyncMode(p->netH)) != kOkNetRC )
  693. goto errLabel;
  694. }
  695. else
  696. {
  697. if((rc = cmRtNetRegisterEndPoint(p->netH,"whirl_ep0", 0 )) != kOkNetRC )
  698. goto errLabel;
  699. }
  700. if( cmThreadPause(p->thH,0) != kOkThRC )
  701. goto errLabel;
  702. cmRptPrintf(&ctx->rpt,"%s q=quit\n", mstrFl ? "Master: " : "Slave: ");
  703. while( (c=getchar()) != 'q' )
  704. {
  705. switch(c)
  706. {
  707. case 'r':
  708. cmRtNetReport(p->netH);
  709. break;
  710. }
  711. }
  712. errLabel:
  713. cmThreadDestroy(&p->thH);
  714. cmRtNetFree(&p->netH);
  715. return;
  716. }