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 20KB

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