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

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695696697698699700701702703704705706707708709710711712713714715716717718719720721722723724725726727728729730731732733734735736737738739740741742743744745746747748749750751752753754755756757758759760761762763764765766767768769770771772773774775776777778779780781782783784785786787788789790791792793794795796797798799800801802803804805806807808809810811812813814815816817818819820821822823824825826827828829830831832833834835836837838839840841842843844845846847848849850851852853854855856857858
  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( cmRtNetIsSyncModeMsg(data,dataByteCnt));
  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. assert( p->syncModeFl );
  464. _cmRtNetRpt(p,"rcv hello ack\n");
  465. rc = _cmRtNetRecvAck(p,fromAddr,kWaitHelloAckStNetId,kSendEndpointStNetId);
  466. break;
  467. case kEndpointAckSelNetId: // master response
  468. assert( p->syncModeFl );
  469. _cmRtNetRpt(p,"rcv endpoint ack\n");
  470. rc = _cmRtNetRecvAck(p,fromAddr,kWaitEndpointAckStNetId,kSendEndpointStNetId);
  471. break;
  472. default:
  473. break;
  474. }
  475. errLabel:
  476. return rc;
  477. }
  478. cmRtNetRC_t _cmRtNetSendNodeSync( cmRtNet_t* p, cmRtNetNode_t* np )
  479. {
  480. cmRtNetRC_t rc = kOkNetRC;
  481. switch( np->state )
  482. {
  483. case kSendHelloStNetId:
  484. // send a 'hello' to this remote node
  485. if((rc = _cmRtNetSendSyncMsg(p,np,kHelloSelNetId,p->localNode->label, cmInvalidId, kWaitHelloAckStNetId )) != kOkNetRC )
  486. rc = cmErrMsg(&p->err,rc,"Send 'hello' to %s:%s:%i failed.",cmStringNullGuard(np->label),cmStringNullGuard(np->addr),np->port);
  487. else
  488. _cmRtNetRpt(p,"send hello\n");
  489. break;
  490. case kSendEndpointStNetId:
  491. {
  492. cmRtNetEnd_t* ep;
  493. // if all of the endpoints have been sent to this node ...
  494. if((ep = _cmRtNetIndexToEndpoint(p,p->localNode,np->epIdx)) == NULL )
  495. np->state = kDoneStNetId; // ... we are done
  496. else
  497. {
  498. // send an endpoint to this node
  499. if((rc = _cmRtNetSendSyncMsg(p,np,kHelloSelNetId,ep->endPtLabel, ep->endPtId, kWaitEndpointAckStNetId )) != kOkNetRC )
  500. 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);
  501. else
  502. _cmRtNetRpt(p,"send endpoint\n");
  503. }
  504. }
  505. break;
  506. case kWaitHelloAckStNetId:
  507. case kWaitEndpointAckStNetId:
  508. {
  509. cmTimeSpec_t t;
  510. cmTimeGet(&t);
  511. unsigned twentySecs = 20000000;
  512. if( cmTimeElapsedMicros(&np->lastSendTime,&t) > twentySecs)
  513. {
  514. const cmChar_t* ackStr = np->state==kWaitHelloAckStNetId ? "hello" : "endpoint";
  515. 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);
  516. }
  517. }
  518. break;
  519. default:
  520. break;
  521. }
  522. // if an error occurred put the node into an error state
  523. if( rc != kOkNetRC )
  524. np->state = kErrorStNetId;
  525. return rc;
  526. }
  527. cmRtNetRC_t cmRtNetSyncModeSend( cmRtNetH_t h )
  528. {
  529. cmRtNetRC_t rc = kOkNetRC;
  530. cmRtNet_t* p = _cmRtNetHandleToPtr(h);
  531. if( p->syncModeFl == false )
  532. return rc;
  533. unsigned activeCnt = 0;
  534. cmRtNetNode_t* np = p->nodes;
  535. for(; np != NULL; np=np->link )
  536. if( np != p->localNode && np->state != kDoneStNetId && np->state != kErrorStNetId )
  537. {
  538. _cmRtNetSendNodeSync(p,np);
  539. activeCnt += 1;
  540. }
  541. if( activeCnt == 0 )
  542. p->syncModeFl = false;
  543. return rc;
  544. }
  545. cmRtNetRC_t cmRtNetReceive( cmRtNetH_t h )
  546. {
  547. cmRtNetRC_t rc = kOkNetRC;
  548. cmRtNet_t* p = _cmRtNetHandleToPtr(h);
  549. if( cmUdpGetAvailData(p->udpH, NULL, NULL, NULL ) != kOkUdpRC )
  550. {
  551. cmErrMsg(&p->err,kUdpPortFailNetRC,"UDP port query failed.");
  552. goto errLabel;
  553. }
  554. errLabel:
  555. return rc;
  556. }
  557. bool cmRtNetIsSyncModeMsg( const void* data, unsigned dataByteCnt )
  558. {
  559. cmRtNetSyncMsg_t* m = (cmRtNetSyncMsg_t*)data;
  560. return dataByteCnt >= sizeof(cmRtNetSyncMsg_t) && m->hdr.selId == kNetSyncSelRtId;
  561. }
  562. unsigned cmRtNetEndPointIndex( cmRtNetH_t h, const cmChar_t* nodeLabel, const cmChar_t* endPtLabel )
  563. {
  564. //cmRtNet_t* p = _cmRtNetHandleToPtr(h);
  565. return cmInvalidIdx;
  566. }
  567. cmRtNetRC_t cmRtNetSend( cmRtNetH_t h, unsigned endPointIndex, const void* msg, unsigned msgByteCnt )
  568. {
  569. cmRtNetRC_t rc = kOkNetRC;
  570. //cmRtNet_t* p = _cmRtNetHandleToPtr(h);
  571. return rc;
  572. }
  573. void cmRtNetReport( cmRtNetH_t h )
  574. {
  575. cmRtNet_t* p = _cmRtNetHandleToPtr(h);
  576. cmRpt_t* rpt = p->err.rpt;
  577. cmRptPrintf(rpt,"Sync Mode:%s\n",p->syncModeFl ? "ON" : "OFF");
  578. cmRtNetNode_t* np = p->nodes;
  579. for(; np!=NULL; np=np->link)
  580. {
  581. cmRptPrintf(rpt,"Node: %s ",np->label);
  582. if( np->addr != NULL )
  583. cmRptPrintf(rpt,"%s ",np->addr );
  584. if( cmIsFlag(np->flags,kLocalNetFl) )
  585. cmRptPrintf(rpt,"LOCAL ");
  586. if( cmIsFlag(np->flags,kSockAddrNetFl) )
  587. cmRptPrintf(rpt,"%s ",cmStringNullGuard(cmUdpAddrToString(p->udpH,&np->sockaddr)));
  588. if( np->port != cmInvalidId )
  589. cmRptPrintf(rpt,"%i ",np->port );
  590. cmRptPrintf(rpt,"\n");
  591. cmRtNetEnd_t* ep = np->ends;
  592. for(; ep!=NULL; ep=ep->link)
  593. {
  594. cmRptPrintf(rpt," endpt: %i %s\n",ep->endPtId,cmStringNullGuard(ep->endPtLabel));
  595. }
  596. }
  597. }
  598. //==========================================================================
  599. #include "cmThread.h"
  600. typedef struct
  601. {
  602. cmThreadH_t thH;
  603. cmRtNetH_t netH;
  604. } _cmRtNetTest_t;
  605. void _cmRtNetTestRecv( void* cbArg, const char* data, unsigned dataByteCnt, const struct sockaddr_in* fromAddr )
  606. {
  607. _cmRtNetTest_t* p = (_cmRtNetTest_t*)cbArg;
  608. if( cmRtNetIsSyncModeMsg(data,dataByteCnt))
  609. cmRtNetSyncModeRecv(p->netH,data,dataByteCnt,fromAddr);
  610. }
  611. bool _cmRtNetTestThreadFunc(void* param)
  612. {
  613. _cmRtNetTest_t* p = (_cmRtNetTest_t*)param;
  614. if( cmRtNetIsValid(p->netH) )
  615. {
  616. if( cmRtNetIsInSyncMode(p->netH) )
  617. cmRtNetSyncModeSend(p->netH);
  618. cmRtNetReceive(p->netH);
  619. }
  620. cmSleepMs(40);
  621. return true;
  622. }
  623. void cmRtNetTest( cmCtx_t* ctx, bool mstrFl )
  624. {
  625. char c;
  626. _cmRtNetTest_t t;
  627. cmUdpPort_t port = 5876;
  628. _cmRtNetTest_t* p = &t;
  629. cmRtNetRC_t rc = kOkNetRC;
  630. unsigned hn = cmUdpHostNameMaxCharCount();
  631. cmChar_t hostNameStr[ hn ];
  632. memset(&t,0,sizeof(t));
  633. if( cmUdpHostName(hostNameStr,hn) != kOkUdpRC )
  634. goto errLabel;
  635. if( cmThreadCreate(&p->thH,_cmRtNetTestThreadFunc,p,&ctx->rpt) != kOkThRC )
  636. goto errLabel;
  637. if((rc = cmRtNetAlloc(ctx,&p->netH,_cmRtNetTestRecv,p)) != kOkNetRC )
  638. goto errLabel;
  639. if((rc = cmRtNetCreateNode(p->netH, hostNameStr, NULL, port )) != kOkNetRC)
  640. goto errLabel;
  641. if( mstrFl )
  642. {
  643. if((rc = cmRtNetCreateNode(p->netH,"whirl", "192.168.15.109", port )) != kOkNetRC )
  644. goto errLabel;
  645. if((rc = cmRtNetRegisterEndPoint(p->netH,"thunk_ep0", 0 )) != kOkNetRC )
  646. goto errLabel;
  647. if(( rc = cmRtNetBeginSyncMode(p->netH)) != kOkNetRC )
  648. goto errLabel;
  649. }
  650. else
  651. {
  652. if((rc = cmRtNetRegisterEndPoint(p->netH,"whirl_ep0", 0 )) != kOkNetRC )
  653. goto errLabel;
  654. }
  655. if( cmThreadPause(p->thH,0) != kOkThRC )
  656. goto errLabel;
  657. cmRptPrintf(&ctx->rpt,"%s q=quit\n", mstrFl ? "Master: " : "Slave: ");
  658. while( (c=getchar()) != 'q' )
  659. {
  660. }
  661. errLabel:
  662. cmThreadDestroy(&p->thH);
  663. cmRtNetFree(&p->netH);
  664. return;
  665. }