123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695696697698699700701702703704705706707708709710711712713714715716717718719720721722723724725726727728729730731732733734735736737738739740741742743744745746747748749750751752753754755756757758759760761762763764765766767768769770771772773774775776777778779780781782783784785786787788789790791792793794795796797798799800801802803804805806807808809810811812813814815816817818819820821822823824825826827828829830831832833834835836837838839840841842843844845846847848849850851852853854855856857858859860861862863864865866867868869870871872873874875876877878879880881882883884885886887888889890891892893894895896897898899900901902903904905906907908909910911912913914915916917918919920921922923924925926927928929930931932933934935936937938939940941942943944945946947948949950951952953954955956957958959960961962963964965966967968969970971972973974975976977978979980981982983984985986987988989990991992993994995996997998999100010011002100310041005100610071008100910101011101210131014101510161017101810191020102110221023102410251026102710281029103010311032103310341035103610371038103910401041104210431044104510461047104810491050105110521053105410551056105710581059106010611062106310641065106610671068106910701071107210731074107510761077107810791080108110821083108410851086108710881089109010911092109310941095109610971098109911001101110211031104110511061107110811091110111111121113111411151116111711181119112011211122112311241125112611271128112911301131113211331134113511361137113811391140114111421143114411451146114711481149115011511152115311541155 |
- #include "cmGlobal.h"
- #include "cmRpt.h"
- #include "cmErr.h"
- #include "cmCtx.h"
- #include "cmMem.h"
- #include "cmMallocDebug.h"
- #include "cmLinkedHeap.h"
- #include "cmUdpPort.h"
- #include "cmRtSysMsg.h"
- #include "cmRtNet.h"
- #include "cmTime.h"
- #include "cmText.h"
-
- // flags for cmRtNetNode_t.flags;
- enum
- {
- kLocalNodeNetFl = 0x01,
- kValidNodeNetFl = 0x02
- };
-
- // flags for cmRtNet_t.flags
- enum
- {
- kReportSyncNetFl = 0x01
- };
-
- struct cmRtNetNode_str;
-
- typedef struct cmRtNetEnd_str
- {
- cmChar_t* label;
- unsigned id;
- struct cmRtNetNode_str* np; // Owner node.
- struct cmRtNetEnd_str* link;
- } cmRtNetEnd_t;
-
- struct cmRtNet_str;
-
- typedef struct cmRtNetNode_str
- {
- unsigned rtSubIdx; // rtSubIdx of the sub-system which owns this node
- cmChar_t* label; // Node label.
- struct sockaddr_in sockaddr; // Socket address
- cmChar_t* addr; // IP Address (human readable)
- cmUdpPort_t port; // Socket port
- unsigned flags; // See kXXXNodeNetFl flags above.
- unsigned endPtIdx; // tracks the next endpoint to send during sync-mode
- unsigned endPtCnt; // local-node=actual cnt of endpt's remote-node:expected cnt of endpt's
- cmTimeSpec_t lastSendTime; // Time of last message sent
- cmRtNetEnd_t* ends; // End point list for this node
- struct cmRtNetNode_str* link;
- } cmRtNetNode_t;
-
- typedef struct cmRtNet_str
- {
- cmErr_t err; // Error state object
- unsigned flags; // See kXXXNetFl above.
- unsigned rtSubIdx; // rtSubIdx of the owning sub-system
- cmUdpH_t udpH; // UDP port handle
- cmUdpCallback_t cbFunc; // Client callback to receive incoming messages from network.
- void* cbArg; //
- cmRtNetNode_t* nodes; // Node list.
- cmRtNetNode_t* localNode; // Pointer to local node (which is also in node list)
- unsigned udpRecvBufByteCnt; // UDP port receive buffer size.
- unsigned udpTimeOutMs; // UDP time-out period
- cmChar_t* bcastAddr; // Network broadcast address
- } cmRtNet_t;
-
- cmRtNetH_t cmRtNetNullHandle = cmSTATIC_NULL_HANDLE;
- cmRtNetEndptH_t cmRtNetEndptNullHandle = cmSTATIC_NULL_HANDLE;
-
- cmRtNet_t* _cmRtNetHandleToPtr( cmRtNetH_t h )
- {
- cmRtNet_t* p = (cmRtNet_t*)h.h;
- assert( p != NULL );
- return p;
- }
-
- cmRtNetEnd_t* _cmRtNetEndptHandleToPtr( cmRtNetEndptH_t h )
- {
- cmRtNetEnd_t* p = (cmRtNetEnd_t*)h.h;
- assert( p != NULL );
- return p;
- }
-
- void _cmRtNetVRpt( cmRtNet_t* p, const cmChar_t* fmt, va_list vl )
- {
- if( cmIsFlag(p->flags,kReportSyncNetFl) )
- cmRptVPrintf(p->err.rpt,fmt,vl);
- }
-
- void _cmRtNetRpt( cmRtNet_t* p, const cmChar_t* fmt, ... )
- {
- va_list vl;
- va_start(vl,fmt);
- _cmRtNetVRpt(p,fmt,vl);
- va_end(vl);
- }
-
- cmRtNetNode_t* _cmRtNetFindNode( cmRtNet_t* p, const cmChar_t* label, unsigned* idxRef )
- {
- if( label == NULL )
- return NULL;
-
- if( idxRef != NULL )
- *idxRef = cmInvalidIdx;
-
- cmRtNetNode_t* np = p->nodes;
- unsigned i;
- for(i=0; np!=NULL; np=np->link,++i)
- if( strcmp(label,np->label)==0)
- {
- if( idxRef != NULL )
- *idxRef = i;
-
- return np;
- }
-
- return NULL;
- }
-
- cmRtNetNode_t* _cmRtNetFindNodeFromSockAddr( cmRtNet_t* p, const struct sockaddr_in* saddr )
- {
- if( saddr == NULL )
- return NULL;
-
- cmRtNetNode_t* np = p->nodes;
- for(; np!=NULL; np=np->link)
- if( np->sockaddr.sin_addr.s_addr == saddr->sin_addr.s_addr && np->sockaddr.sin_port == saddr->sin_port )
- return np;
-
- return NULL;
- }
-
- void _cmRtNetFreeNode( cmRtNetNode_t* np )
- {
- cmRtNetEnd_t* ep = np->ends;
- while( ep != NULL )
- {
- cmRtNetEnd_t* nep = ep->link;
- cmMemFree(ep->label);
- cmMemFree(ep);
- ep = nep;
- }
-
- cmMemFree(np->label);
- cmMemFree(np->addr);
- cmMemFree(np);
- }
-
- void _cmRtNetReleaseNodes( cmRtNet_t* p )
- {
- cmRtNetNode_t* np = p->nodes;
- while( np != NULL )
- {
- cmRtNetNode_t* nnp = np->link;
-
- _cmRtNetFreeNode(np);
-
- np = nnp;
- }
- p->nodes = NULL;
- p->localNode = NULL;
- }
-
- cmRtNetRC_t _cmRtNetReleaseNode( cmRtNet_t* p, cmRtNetNode_t* np )
- {
- cmRtNetNode_t* cnp = p->nodes;
- cmRtNetNode_t* pnp = NULL;
-
- while( cnp != NULL )
- {
- cmRtNetNode_t* nnp = cnp->link;
- if( np == cnp )
- {
- if( pnp == NULL )
- p->nodes = np->link;
- else
- pnp->link = np->link;
-
- _cmRtNetFreeNode(np);
-
- return kOkNetRC;
- }
-
- pnp = np;
- cnp = nnp;
- }
-
- assert(0);
- return cmErrMsg(&p->err,kNodeNotFoundNetRC,"Node to release not found.");
- }
-
- cmRtNetRC_t _cmRtNetCreateNode( cmRtNet_t* p, const cmChar_t* label, unsigned rtSubIdx, const cmChar_t* addr, cmUdpPort_t port, const struct sockaddr_in* saddr, unsigned flags, unsigned endPtCnt )
- {
- cmRtNetRC_t rc = kOkNetRC;
- cmRtNetNode_t* np;
-
- if( cmTextIsEmpty(label) )
- return cmErrMsg(&p->err,kInvalidLabelNetRC,"A null or blank node label was encountered.");
-
- if((np = _cmRtNetFindNode(p,label,NULL)) != NULL )
- cmErrWarnMsg(&p->err,kOkNetRC/*kDuplLabelNetRC*/,"The node label '%s' is being reused.",cmStringNullGuard(label));
- else
- {
- np = cmMemAllocZ(cmRtNetNode_t,1);
- np->label = cmMemAllocStr(label);
- }
-
- if( saddr != NULL )
- np->sockaddr = *saddr;
-
- np->rtSubIdx = rtSubIdx;
- np->addr = addr==NULL ? NULL : cmMemResizeStr(np->addr,addr);
- np->port = port;
- np->flags = flags;
- np->endPtCnt = endPtCnt;
- np->link = p->nodes;
- p->nodes = np;
-
- return rc;
- }
-
- cmRtNetEnd_t* _cmRtNetFindNodeEnd(cmRtNetNode_t* np, const cmChar_t* endPtLabel )
- {
- cmRtNetEnd_t* ep = np->ends;
- for(; ep!=NULL; ep=ep->link)
- if( strcmp(ep->label,endPtLabel)==0 )
- return ep;
- return NULL;
- }
-
- cmRtNetEnd_t* _cmRtNetIndexToEndpoint( cmRtNet_t* p, cmRtNetNode_t* np, unsigned endIndex )
- {
- cmRtNetEnd_t* ep = np->ends;
- unsigned i;
- for(i=0; ep!=NULL; ep=ep->link)
- {
- if( i == endIndex )
- return ep;
- ++i;
- }
-
- return NULL;
- }
-
- cmRtNetRC_t _cmRtNetCreateEndpoint( cmRtNet_t* p, cmRtNetNode_t* np, const cmChar_t* endPtLabel, unsigned endPtId )
- {
- if( endPtLabel == NULL )
- return cmErrMsg(&p->err,kInvalidLabelNetRC,"A null or blank node label was encountered.");
-
- if( _cmRtNetFindNodeEnd( np, endPtLabel) != NULL)
- return cmErrMsg(&p->err,kDuplEndNetRC,"A duplicate endpoint ('%s') was encountered on node '%s'.",endPtLabel,np->label);
-
- cmRtNetRC_t rc = kOkNetRC;
- cmRtNetEnd_t* ep = cmMemAllocZ(cmRtNetEnd_t,1);
-
- ep->label = cmMemAllocStr(endPtLabel);
- ep->id = endPtId;
- ep->np = np;
- ep->link = np->ends;
- np->ends = ep;
-
- return rc;
- }
-
- unsigned _cmRtNetNodeEndpointCount( cmRtNetNode_t* np )
- {
- cmRtNetEnd_t* ep = np->ends;
- unsigned n = 0;
- for(; ep!=NULL; ep=ep->link)
- ++n;
- return n;
- }
-
- unsigned _cmRtNetSyncMsgSerialByteCount( const cmRtNetSyncMsg_t* m )
- {
- return sizeof(cmRtNetSyncMsg_t) + (m->label==NULL ? 1 : strlen(m->label) + 1);
- }
-
- cmRtNetRC_t _cmRtNetSerializeSyncMsg( cmRtNet_t* p, const cmRtNetSyncMsg_t* m, void* buf, unsigned n )
- {
- unsigned bn = _cmRtNetSyncMsgSerialByteCount(m);
- char* b = (char*)buf;
-
- if( bn > n )
- return cmErrMsg(&p->err,kBufToSmallNetRC,"Serialize buffer too small.");
-
- memcpy(b,m,sizeof(*m));
- strcpy(b + sizeof(*m),m->label==NULL ? "" : m->label);
- return kOkNetRC;
- }
-
- cmRtNetRC_t _cmRtNetDeserializeSyncMsg( const void* buf, unsigned n, cmRtNetSyncMsg_t* m )
- {
- const cmRtNetSyncMsg_t* mp = (const cmRtNetSyncMsg_t*)buf;
-
- m->hdr.rtSubIdx = mp->hdr.rtSubIdx;
- m->hdr.selId = mp->hdr.selId;
- m->selId = mp->selId;
- m->hdrByteCnt = sizeof(cmRtNetSyncMsg_t);
- m->rtSubIdx = mp->rtSubIdx;
- m->id = mp->id;
-
- const cmChar_t* s = ((const cmChar_t*)(mp)) + mp->hdrByteCnt;
- m->label = cmMemAllocStr(s);
- return kOkNetRC;
- }
-
- cmRtNetRC_t _cmRtNetSendSyncMsg( cmRtNet_t* p, cmRtNetNode_t* np, cmRtNetSelId_t selId, const cmChar_t* msgLabel, unsigned msgId, unsigned msgRtSubIdx )
- {
- cmRtNetSyncMsg_t m;
- cmRtNetRC_t rc = kOkNetRC;
- cmUdpRC_t udpRC = kOkUdpRC;
-
- m.hdr.rtSubIdx = cmInvalidIdx;
- m.hdr.selId = kNetSyncSelRtId;
- m.selId = selId;
- m.hdrByteCnt = sizeof(cmRtNetSyncMsg_t);
- m.label = msgLabel;
- m.id = msgId;
- m.rtSubIdx = msgRtSubIdx;
-
- // determine size of msg to send
- unsigned n = _cmRtNetSyncMsgSerialByteCount(&m);
- cmChar_t buf[n];
-
- // serialize msg into buf[]
- if((rc = _cmRtNetSerializeSyncMsg(p,&m,buf,n)) != kOkNetRC )
- return rc;
-
- // send the msg
- if( selId == kHelloSelNetId )
- udpRC = cmUdpSend2(p->udpH, buf, n, p->bcastAddr, np->port );
- else
- udpRC = cmUdpSendTo(p->udpH, buf, n, &np->sockaddr );
-
- // check for send errors
- if( udpRC != kOkUdpRC )
- {
- rc = cmErrMsg(&p->err,kUdpPortFailNetRC,"Sync msg. send on UDP port failed.");
- }
- else
- {
- // record the last send time
- cmTimeGet(&np->lastSendTime);
- }
-
- return rc;
- }
-
- cmRtNetRC_t _cmRtNetFree( cmRtNet_t* p )
- {
- cmRtNetRC_t rc = kOkNetRC;
-
- if( cmUdpFree(&p->udpH) != kOkUdpRC )
- cmErrMsg(&p->err,kUdpPortFailNetRC,"UDP Port free failed.");
-
- _cmRtNetReleaseNodes(p);
-
- cmMemFree(p->bcastAddr);
-
- cmMemFree(p);
- return rc;
- }
-
- const cmRtNetNode_t* _cmRtNetIndexToRemoteNode( cmRtNet_t* p, unsigned idx )
- {
- const cmRtNetNode_t* np = p->nodes;
- unsigned i = 0;
-
- for(; np!=NULL; np=np->link)
- if( np != p->localNode )
- {
- if( i == idx )
- return np;
- ++i;
- }
-
- return NULL;
- }
-
- const cmRtNetEnd_t* _cmRtNetIndexToEndpt( const cmRtNetNode_t* np, unsigned endIdx )
- {
- const cmRtNetEnd_t* ep = np->ends;
- unsigned i = 0;
-
- for(; ep!=NULL; ep=ep->link,++i)
- if( i==endIdx )
- return ep;
-
- return NULL;
- }
-
- const cmRtNetEnd_t* _cmRtNetFindEndpt( cmRtNet_t* p, unsigned nodeIdx, unsigned epIdx )
- {
- const cmRtNetNode_t* np;
- const cmRtNetEnd_t* ep;
-
- if((np = _cmRtNetIndexToRemoteNode( p, nodeIdx )) == NULL )
- return NULL;
-
- if((ep = _cmRtNetIndexToEndpt( np, epIdx )) == NULL )
- return NULL;
-
- return ep;
- }
-
-
- const cmChar_t* cmRtNetSyncMsgLabel( const cmRtNetSyncMsg_t* m )
- {
- if( m->selId==kNodeSelNetId || m->selId==kEndpointSelNetId )
- return (const cmChar_t*)(m+1);
-
- return "";
- }
-
- cmRtNetRC_t cmRtNetAlloc( cmCtx_t* ctx, cmRtNetH_t* hp, unsigned rtSubIdx, cmUdpCallback_t cbFunc, void* cbArg )
- {
- cmRtNetRC_t rc;
- if((rc = cmRtNetFree(hp)) != kOkNetRC )
- return rc;
-
- cmRtNet_t* p = cmMemAllocZ(cmRtNet_t,1);
- cmErrSetup(&p->err,&ctx->rpt,"cmRtNet");
-
- // allocate the UDP port
- if(cmUdpAlloc(ctx,&p->udpH) != kOkUdpRC )
- {
- cmErrMsg(&p->err,kUdpPortFailNetRC,"UDP Port allocate failed.");
- goto errLabel;
- }
-
- p->rtSubIdx = rtSubIdx;
- p->udpTimeOutMs = 50;
- p->udpRecvBufByteCnt = 8192;
- p->cbFunc = cbFunc;
- p->cbArg = cbArg;
-
- hp->h = p;
-
- errLabel:
- if(rc != kOkNetRC )
- _cmRtNetFree(p);
-
- return rc;
- }
-
- cmRtNetRC_t cmRtNetFree( cmRtNetH_t* hp )
- {
- cmRtNetRC_t rc = kOkNetRC;
-
- if( hp==NULL || cmRtNetIsValid(*hp)==false )
- return rc;
-
- cmRtNet_t* p = _cmRtNetHandleToPtr(*hp);
-
- if((rc = _cmRtNetFree(p)) != kOkNetRC )
- return rc;
-
- hp->h = NULL;
-
- return rc;
- }
-
- const cmChar_t* cmRtNetLocalHostName( cmRtNetH_t h )
- {
- cmRtNet_t* p = _cmRtNetHandleToPtr(h);
- return cmUdpHostName(p->udpH);
- }
-
-
- bool cmRtNetIsValid( cmRtNetH_t h )
- { return h.h !=NULL; }
-
- cmUdpH_t cmRtNetUdpPortHandle( cmRtNetH_t h )
- {
- cmRtNet_t* p = _cmRtNetHandleToPtr(h);
- return p->udpH;
- }
-
- cmRtNetRC_t _cmRtNetSendEndpointReplyMsg( cmRtNet_t* p, cmRtNetNode_t* np, cmRtNetSelId_t srcSelId )
- {
- cmRtNetRC_t rc = kOkNetRC;
- cmRtNetEnd_t* ep = NULL;
- const cmChar_t* msgLabel = NULL;
- unsigned msgId = cmInvalidId;
- unsigned msgRtSubIdx = cmInvalidIdx;
- cmRtNetSelId_t selId = kEndpointSelNetId;
- const cmChar_t* rptLabel = "endpoint";
-
- if( np == NULL )
- return cmErrMsg(&p->err,kNodeNotFoundNetRC,"The net node associated with an endpoint reply was not found.");
-
- // if we got here by receiving a 'done' msg from the remote node ...
- if( srcSelId == kDoneSelNetId )
- {
- // ... then mark the remote node as having recieved all endpoints
- unsigned n;
- if((n = _cmRtNetNodeEndpointCount(np)) != np->endPtCnt )
- rc = cmErrMsg(&p->err,kNodeEndCntErrNetRC,"The node '%s' expected %i endpoints but received %i.",cmStringNullGuard(np->label),np->endPtCnt,n);
- else
- np->flags = cmSetFlag(np->flags,kValidNodeNetFl);
- }
-
- // attempt to get the next local endpoint to send ...
- if((ep = _cmRtNetIndexToEndpoint(p,p->localNode,np->endPtIdx)) != NULL )
- {
- msgLabel = ep->label; // ... send next local endpoint
- msgId = ep->id;
- msgRtSubIdx = ep->np->rtSubIdx;
- }
- else // .... all local endpoints have been sent
- {
- selId = kInvalidSelNetId;
- rptLabel = "done";
-
- // verify that no endpoints are available
- if( np->endPtIdx < p->localNode->endPtCnt )
- rc = cmErrMsg(&p->err,kSyncFailNetRC,"More endpoints are available to send but are not reachable.");
- else
- {
- // if the remote node still has endpts to send then continue
- // sending 'done' messages.
- if( np->endPtIdx==p->localNode->endPtCnt || srcSelId != kDoneSelNetId )
- selId = kDoneSelNetId;
- }
- }
-
- // selId is set to kInvalidSelNetId when we encounter the (stop) criteria
- if( selId != kInvalidSelNetId )
- {
- if((rc = _cmRtNetSendSyncMsg(p,np,selId,msgLabel,msgId,msgRtSubIdx)) != kOkNetRC )
- rc = cmErrMsg(&p->err,rc,"Send '%s' to %s:%s:%i failed.",rptLabel,cmStringNullGuard(np->label),cmStringNullGuard(np->addr),np->port);
- else
- _cmRtNetRpt(p,"Sent %s.\n",cmStringNullGuard(rptLabel));
-
- np->endPtIdx += 1;
- }
-
- return rc;
- }
-
- bool _cmRtNetIsSyncModeMsg( const void* data, unsigned dataByteCnt )
- {
- cmRtNetSyncMsg_t* m = (cmRtNetSyncMsg_t*)data;
- return dataByteCnt >= sizeof(cmRtSysMsgHdr_t) && m->hdr.selId == kNetSyncSelRtId;
- }
-
- // When the network message recieve function (See cmRtNetAlloc() 'cbFunc')
- // receives a message with the cmRtSysMsgHdr_t.selId == kNetSyncSelRtId
- // it should call this function to update the current sync state of the
- // cmRtNet.
- cmRtNetRC_t _cmRtNetSyncModeRecv( cmRtNet_t* p, const char* data, unsigned dataByteCnt, const struct sockaddr_in* fromAddr )
- {
- cmRtNetRC_t rc = kOkNetRC;
- cmRtNetSyncMsg_t m;
- m.label = NULL;
-
- assert( _cmRtNetIsSyncModeMsg(data,dataByteCnt));
-
- if( _cmRtNetDeserializeSyncMsg(data,dataByteCnt,&m) != kOkNetRC )
- {
- rc = cmErrMsg(&p->err,rc,"Net sync. receive failed due to deserialize fail.");
- goto errLabel;
- }
-
- _cmRtNetRpt(p,"recv from:%s\n",cmUdpAddrToString(p->udpH, fromAddr ));
-
-
- assert( m.hdr.selId == kNetSyncSelRtId );
-
- // attempt to locate the remote node which sent the msg
- cmRtNetNode_t* np = _cmRtNetFindNodeFromSockAddr(p,fromAddr);
-
-
- switch( m.selId )
- {
- case kHelloSelNetId:
- // if this is a response to a broadcast from the local node then ignore it
- if(m.label!=NULL && p->localNode->label!=NULL && (np = _cmRtNetFindNode(p,m.label,NULL)) != NULL && strcmp(p->localNode->label,m.label)==0 )
- {
- const cmChar_t* fromAddrStr = cmUdpAddrToString(p->udpH,fromAddr);
- const cmChar_t* localAddrStr = cmUdpAddrToString(p->udpH,cmUdpLocalAddr(p->udpH));
-
- if( fromAddrStr!=NULL && localAddrStr!=NULL && strcmp(fromAddrStr,localAddrStr)!=0)
- cmErrMsg(&p->err,kDuplLocalNetRC,"The node label '%s' appears to be duplicated at address %s and locally.",cmStringNullGuard(m.label),fromAddrStr);
-
- np->sockaddr = *fromAddr;
- goto errLabel;
- }
- // fall through
-
- case kNodeSelNetId:
- {
- // if the node already exists ...
- if( np != NULL )
- {
- // ... delete it because we are about to get new info. about it.
- if((rc = _cmRtNetReleaseNode(p,np )) != kOkNetRC )
- goto errLabel;
- }
-
- // create a node proxy to represent the remote node
- // (Note:m.id == remote node endpoint count (i.e. the count of endpoints expected for the remote node.))
- if(( rc = _cmRtNetCreateNode(p,m.label,m.rtSubIdx,NULL,0,fromAddr,0,m.id)) != kOkNetRC )
- goto errLabel;
-
- np = p->nodes; // newest node is always the first node
-
- // send response
- switch( m.selId )
- {
- case kHelloSelNetId:
- _cmRtNetRpt(p,"rcv hello\n"); // reply with local node
- rc = _cmRtNetSendSyncMsg( p, np, kNodeSelNetId, p->localNode->label, p->localNode->endPtCnt, p->localNode->rtSubIdx );
- break;
-
- case kNodeSelNetId:
- _cmRtNetRpt(p,"rcv node\n");
- _cmRtNetSendEndpointReplyMsg( p, np, m.selId ); // reply with first endpoint
- break;
-
- default:
- assert(0);
- }
-
- }
- break;
-
- case kDoneSelNetId:
- //case kEndpointAckSelNetId:
- rc = _cmRtNetSendEndpointReplyMsg(p,np,m.selId);
- break;
-
- case kEndpointSelNetId:
- {
- cmRtNetEnd_t* ep;
-
- // verify the remote node exists.
- if( np == NULL )
- {
- rc = cmErrMsg(&p->err,kNodeNotFoundNetRC,"The net node associated with an endpoint receive was not found.");
- goto errLabel;
- }
-
- // attempt to find the end point
- if((ep = _cmRtNetFindNodeEnd(np, m.label)) != NULL )
- ep->id = m.id; // the endpoint was found update the endPtId
- else
- {
- // create a local proxy for the endpoint
- if((rc = _cmRtNetCreateEndpoint(p,np,m.label,m.id)) != kOkNetRC )
- goto errLabel;
- }
-
- // reply with a local endpoint or 'done' msg
- rc = _cmRtNetSendEndpointReplyMsg( p, np, m.selId );
- }
- break;
-
- default:
- assert(0);
- break;
- }
-
- errLabel:
-
- cmMemFree((cmChar_t*)m.label);
- return rc;
- }
-
- unsigned _cmRtNetAddrToNodeIndex( cmRtNet_t* p, const struct sockaddr_in* addr )
- {
- unsigned i;
- cmRtNetNode_t* np = p->nodes;
- for(i=0; np!=NULL; np=np->link,++i)
- if( cmUdpAddrIsEqual( addr, &np->sockaddr ) )
- return i;
-
- return cmInvalidIdx;
- }
-
-
- // This is called in the context of cmRtNetReceive().
- void _cmRtNetRecv( void* cbArg, const char* data, unsigned dataByteCnt, const struct sockaddr_in* fromAddr )
- {
- cmRtNet_t* p = (cmRtNet_t*)cbArg;
-
- // if this is a sync msg - then handle it here
- if( _cmRtNetIsSyncModeMsg(data,dataByteCnt))
- _cmRtNetSyncModeRecv(p,data,dataByteCnt,fromAddr);
- else
- {
- // All non-sync messages arriving here are prefixed by a cmRtNetMsg_t header - fill in the source addr here.
- // NOTE: the source addr could be filled in by the sender but this would increase the size
- // of the msg. Instead we choose the more time consuming method of looking up the
- // soure node here - hmmmm????.
-
- cmRtNetMsg_t* hdr = (cmRtNetMsg_t*)(data);
- hdr->srcNodeIdx = _cmRtNetAddrToNodeIndex(p,fromAddr);
- }
-
- p->cbFunc(p->cbArg,data,dataByteCnt,fromAddr);
-
- }
-
- cmRtNetRC_t cmRtNetInitialize( cmRtNetH_t h, const cmChar_t* bcastAddr, const cmChar_t* nodeLabel, const cmChar_t* ipAddr, cmUdpPort_t port )
- {
- cmRtNet_t* p = _cmRtNetHandleToPtr(h);
- cmRtNetRC_t rc;
-
- // release the local node
- if((rc = cmRtNetFinalize(h)) != kOkNetRC )
- goto errLabel;
-
- if( cmTextIsEmpty(bcastAddr) )
- {
- rc = cmErrMsg(&p->err,kInvalidArgNetRC,"The 'broadcast address' is not valid.");
- goto errLabel;
- }
-
- // if this is the local node then initialze the local socket
- if( cmUdpInit(p->udpH,port,kNonBlockingUdpFl | kBroadcastUdpFl,_cmRtNetRecv,p,NULL,0,p->udpRecvBufByteCnt,p->udpTimeOutMs) != kOkUdpRC )
- {
- rc = cmErrMsg(&p->err,kUdpPortFailNetRC,"The UDP port initialization failed.");
- goto errLabel;
- }
-
- // create the local node
- if((rc = _cmRtNetCreateNode(p,nodeLabel, p->rtSubIdx, ipAddr, port, NULL, kLocalNodeNetFl, 0)) != kOkNetRC )
- goto errLabel;
-
- // the last created node is always the first node on the list
- p->localNode = p->nodes;
- p->bcastAddr = cmMemResizeStr(p->bcastAddr,bcastAddr);
-
- // begin listening on the local port
- if( cmUdpEnableListen(p->udpH, true ) != kOkUdpRC )
- {
- rc = cmErrMsg(&p->err,kUdpPortFailNetRC,"The UDP port failed to enter 'listen' mode.");
- goto errLabel;
- }
-
-
- errLabel:
- return rc;
- }
-
-
- bool cmRtNetIsInitialized( cmRtNetH_t h )
- {
- if( cmRtNetIsValid(h) == false )
- return false;
-
- cmRtNet_t* p = _cmRtNetHandleToPtr(h);
- return p->localNode != NULL && cmTextIsNotEmpty(p->bcastAddr);
- }
-
-
- cmRtNetRC_t cmRtNetRegisterEndPoint( cmRtNetH_t h, const cmChar_t* endPtLabel, unsigned endPtId )
- {
- cmRtNetRC_t rc = kOkNetRC;
- cmRtNet_t* p = _cmRtNetHandleToPtr(h);
-
- if( p->localNode == NULL )
- return cmErrMsg(&p->err,kLocalNodeNetRC,"Local endpoints may not be added if a local node has not been defined.");
-
- if((rc = _cmRtNetCreateEndpoint(p, p->localNode,endPtLabel,endPtId )) == kOkNetRC )
- p->localNode->endPtCnt += 1;
-
- return rc;
- }
-
- cmRtNetRC_t cmRtNetFinalize( cmRtNetH_t h )
- {
- cmRtNet_t* p = _cmRtNetHandleToPtr(h);
- _cmRtNetReleaseNodes(p);
-
- return kOkNetRC;
- }
-
- cmRtNetRC_t cmRtNetDoSync( cmRtNetH_t h )
- {
- cmRtNet_t* p = _cmRtNetHandleToPtr(h);
-
- // broadcast 'node' msg
- return _cmRtNetSendSyncMsg( p, p->localNode, kHelloSelNetId, p->localNode->label, p->localNode->endPtCnt, cmInvalidIdx );
- }
-
- cmRtNetRC_t cmRtNetReceive( cmRtNetH_t h )
- {
- cmRtNetRC_t rc = kOkNetRC;
- cmRtNet_t* p = _cmRtNetHandleToPtr(h);
-
- // Calling this function results in callbacks to _cmRtNetRecv() (above)
- if( cmUdpGetAvailData(p->udpH, NULL, NULL, NULL ) != kOkUdpRC )
- {
- cmErrMsg(&p->err,kUdpPortFailNetRC,"UDP port query failed.");
- goto errLabel;
- }
- errLabel:
- return rc;
- }
-
-
- cmRtNetRC_t cmRtNetEndpointHandle( cmRtNetH_t h, const cmChar_t* nodeLabel, const cmChar_t* endptLabel, cmRtNetEndptH_t* hp )
- {
- cmRtNetRC_t rc = kOkNetRC;
- cmRtNet_t* p = _cmRtNetHandleToPtr(h);
- cmRtNetNode_t* np;
- cmRtNetEnd_t* ep;
-
- if(( np = _cmRtNetFindNode(p,nodeLabel,NULL)) == NULL )
- return cmErrMsg(&p->err,kNodeNotFoundNetRC,"The node '%s' was not found.",cmStringNullGuard(nodeLabel));
-
-
- if(( ep = _cmRtNetFindNodeEnd(np, endptLabel )) == NULL )
- return cmErrMsg(&p->err,kEndNotFoundNetRC,"The endpoint '%s' on '%s' on node was not found.",cmStringNullGuard(endptLabel),cmStringNullGuard(nodeLabel));
-
- hp->h = ep;
-
- return rc;
- }
-
- bool cmRtNetEndpointIsValid( cmRtNetEndptH_t endPtH )
- { return endPtH.h != NULL; }
-
- unsigned cmRtNetEndpointId( cmRtNetEndptH_t endPtH )
- {
- if( !cmRtNetEndpointIsValid(endPtH) )
- return cmInvalidId;
-
- cmRtNetEnd_t* ep = _cmRtNetEndptHandleToPtr( endPtH );
- return ep->id;
- }
-
- const cmChar_t* cmRtNetEndpointLabel( cmRtNetEndptH_t endPtH )
- {
- if( !cmRtNetEndpointIsValid(endPtH) )
- return NULL;
-
- cmRtNetEnd_t* ep = _cmRtNetEndptHandleToPtr( endPtH );
- return ep->label;
- }
-
-
- cmRtNetRC_t _cmRtNetSend( cmRtNet_t* p, unsigned srcEndPtId, const cmRtNetEnd_t* ep, const void* msg, unsigned msgByteCnt )
- {
- cmRtNetRC_t rc = kOkNetRC;
-
- unsigned hN = sizeof(cmRtNetMsg_t);
- unsigned dN = hN + msgByteCnt;
- char data[ dN ];
-
- cmRtNetMsg_t* r = (cmRtNetMsg_t*)data;
- r->hdr.rtSubIdx = ep->np->rtSubIdx;
- r->hdr.selId = kMsgSelRtId;
- r->dstEndPtId = ep->id;
- r->srcEndPtId = srcEndPtId;
- memcpy(data+hN,msg,msgByteCnt);
-
- // ep->np->sockaddr identifies the node on the receiving cmRtNet.
- // cmRtNetMsg_t* r.endptId is then used by the receiving cmRtNet to indicate which endpoint on
- // the node the incoming message should be associated with.
-
- if( cmUdpSendTo(p->udpH, data, dN, &ep->np->sockaddr ) != kOkUdpRC )
- return cmErrMsg(&p->err,kUdpPortFailNetRC,"Send to node:%s endpt:%s failed.\n",cmStringNullGuard(ep->np->label),cmStringNullGuard(ep->label));
-
- return rc;
- }
-
- cmRtNetRC_t cmRtNetSend( cmRtNetH_t h, unsigned srcEndPtId, cmRtNetEndptH_t epH, const void* msg, unsigned msgByteCnt )
- {
- cmRtNet_t* p = _cmRtNetHandleToPtr(h);
- cmRtNetEnd_t* ep = _cmRtNetEndptHandleToPtr(epH);
-
- assert( ep != NULL );
- return _cmRtNetSend(p,srcEndPtId,ep,msg,msgByteCnt);
- }
-
-
- cmRtNetRC_t cmRtNetSendByLabels( cmRtNetH_t h, unsigned srcEndPtId, const cmChar_t* nodeLabel, const cmChar_t* endptLabel, const void* msg, unsigned msgByteCnt )
- {
- cmRtNetRC_t rc = kOkNetRC;
- cmRtNetEndptH_t epH = cmRtNetEndptNullHandle;
-
- if((rc = cmRtNetEndpointHandle(h,nodeLabel,endptLabel,&epH)) != kOkNetRC )
- return rc;
-
- return cmRtNetSend(h,srcEndPtId,epH,msg,msgByteCnt);
- }
-
- cmRtNetRC_t cmRtNetSendByIndex( cmRtNetH_t h, unsigned srcEndPtId, unsigned nodeIdx, unsigned endptIdx, const void* msg, unsigned msgByteCnt )
- {
- cmRtNet_t* p = _cmRtNetHandleToPtr(h);
-
- const cmRtNetEnd_t* ep;
-
- if((ep = _cmRtNetFindEndpt(p, nodeIdx, endptIdx )) == NULL )
- return cmErrMsg(&p->err,kEndNotFoundNetRC,"The endpoint at node index %i endpoint index %i was not found.",nodeIdx,endptIdx);
-
- return _cmRtNetSend( p, srcEndPtId, ep, msg, msgByteCnt );
- }
-
-
- bool cmRtNetReportSyncEnable( cmRtNetH_t h, bool enableFl )
- {
- cmRtNet_t* p = _cmRtNetHandleToPtr(h);
- bool fl = cmIsFlag(p->flags,kReportSyncNetFl);
- p->flags = cmEnaFlag(p->flags,kReportSyncNetFl,enableFl);
- return fl;
- }
-
- bool cmRtNetReportSyncIsEnabled( cmRtNetH_t h )
- {
- cmRtNet_t* p = _cmRtNetHandleToPtr(h);
- return cmIsFlag(p->flags,kReportSyncNetFl);
- }
-
- void cmRtNetReport( cmRtNetH_t h )
- {
- cmRtNet_t* p = _cmRtNetHandleToPtr(h);
- cmRpt_t* rpt = p->err.rpt;
-
-
- cmRtNetNode_t* np = p->nodes;
- for(; np!=NULL; np=np->link)
- {
- cmRptPrintf(rpt,"Node: %s ",np->label);
-
- if( np->addr != NULL )
- cmRptPrintf(rpt,"%s ",np->addr );
-
- if( cmIsFlag(np->flags,kLocalNodeNetFl) )
- cmRptPrintf(rpt,"LOCAL ");
-
- cmRptPrintf(rpt,"%s ",cmStringNullGuard(cmUdpAddrToString(p->udpH,&np->sockaddr)));
-
- if( np->port != kInvalidUdpPortNumber )
- cmRptPrintf(rpt,"%i ",np->port );
-
- cmRptPrintf(rpt,"\n");
-
- cmRtNetEnd_t* ep = np->ends;
- for(; ep!=NULL; ep=ep->link)
- {
- cmRptPrintf(rpt," endpt: %i %s\n",ep->id,cmStringNullGuard(ep->label ));
- }
- }
- }
-
- const cmChar_t* cmRtNetLocalNodeLabel( cmRtNetH_t h )
- {
- cmRtNet_t* p = _cmRtNetHandleToPtr( h );
- return p->localNode->label;
- }
-
- unsigned cmRtNetRemoteNodeCount( cmRtNetH_t h )
- {
- cmRtNet_t* p = _cmRtNetHandleToPtr( h );
- const cmRtNetNode_t* np = p->nodes;
- unsigned n = 0;
-
- for(; np!=NULL; np=np->link)
- if( np != p->localNode )
- ++n;
-
- return n;
- }
-
- unsigned cmRtNetAddrToNodeIndex( cmRtNetH_t h, const struct sockaddr_in* a )
- {
- cmRtNet_t* p = _cmRtNetHandleToPtr(h);
- return _cmRtNetAddrToNodeIndex(p,a);
- }
-
- unsigned cmRtNetRemoteNodeIndex( cmRtNetH_t h, const cmChar_t* label )
- {
- cmRtNet_t* p = _cmRtNetHandleToPtr( h );
- unsigned idx = cmInvalidIdx;
-
- _cmRtNetFindNode(p,label,&idx);
-
- return idx;
- }
-
- const cmChar_t* cmRtNetRemoteNodeLabel( cmRtNetH_t h, unsigned idx )
- {
- cmRtNet_t* p = _cmRtNetHandleToPtr( h );
- const cmRtNetNode_t* np;
-
- if((np = _cmRtNetIndexToRemoteNode( p, idx )) == NULL )
- return NULL;
-
- return np->label;
- }
-
- unsigned cmRtNetRemoteNodeEndPointCount( cmRtNetH_t h, unsigned nodeIdx )
- {
- const cmRtNetNode_t* np;
- const cmRtNetEnd_t* ep;
- cmRtNet_t* p = _cmRtNetHandleToPtr( h );
- unsigned n = 0;
-
- if((np = _cmRtNetIndexToRemoteNode( p, nodeIdx )) == NULL )
- return 0;
-
- for(ep=np->ends; ep!=NULL; ep=ep->link)
- ++n;
-
- return n;
- }
-
- cmRtNetRC_t cmRtNetRemoteNodeEndPoint(
- cmRtNetH_t h,
- unsigned nodeIdx,
- unsigned epIdx,
- const cmChar_t** labelRef,
- unsigned* idRef,
- unsigned* rsiRef )
- {
- const cmRtNetEnd_t* ep;
- cmRtNet_t* p = _cmRtNetHandleToPtr( h );
-
- if(( ep = _cmRtNetFindEndpt(p, nodeIdx, epIdx )) == NULL )
- {
- *labelRef = NULL;
- *idRef = cmInvalidId;
- *rsiRef = cmInvalidIdx;
- return kEndNotFoundNetRC;
- }
-
- *labelRef = ep->label;
- *idRef = ep->id;
- *rsiRef = ep->np->rtSubIdx;
-
- return kOkNetRC;
- }
-
-
-
- //==========================================================================
- #include "cmThread.h"
-
-
- typedef struct
- {
- cmThreadH_t thH;
- cmRtNetH_t netH;
- unsigned msgVal;
- } _cmRtNetTest_t;
-
- // This function is called within the context of cmRtNetReceive().
- void _cmRtNetTestRecv( void* cbArg, const char* data, unsigned dataByteCnt, const struct sockaddr_in* fromAddr )
- {
- //_cmRtNetTest_t* p = (_cmRtNetTest_t*)cbArg;
-
- cmRtNetMsg_t* r = (cmRtNetMsg_t*)data;
- unsigned i = *(unsigned*)(data + sizeof(cmRtNetMsg_t));
- printf("rtSubIdx:%i endptId:%i %i\n",r->hdr.rtSubIdx,r->dstEndPtId,i);
-
- }
-
-
- bool _cmRtNetTestThreadFunc(void* param)
- {
- _cmRtNetTest_t* p = (_cmRtNetTest_t*)param;
-
- if( cmRtNetIsValid(p->netH) )
- {
- cmRtNetReceive(p->netH);
- }
-
- cmSleepMs(40);
-
- return true;
- }
-
- void cmRtNetTest( cmCtx_t* ctx, bool mstrFl )
- {
- char c;
- _cmRtNetTest_t t;
- const unsigned rtSubIdx = 0;
- cmUdpPort_t port = 5876;
- _cmRtNetTest_t* p = &t;
- cmRtNetRC_t rc = kOkNetRC;
- const cmChar_t* localHostStr = mstrFl ? "master" : "slave";
- const cmChar_t* localEndpStr = mstrFl ? "master_ep" : "slave_ep";
- const cmChar_t* remoteHostStr = !mstrFl ? "master" : "slave";
- const cmChar_t* remoteEndpStr = !mstrFl ? "master_ep" : "slave_ep";
- const cmChar_t* bcastAddr = "192.168.15.255";
- cmRtNetEndptH_t eH = cmRtNetEndptNullHandle;
- unsigned srcEndPtId = cmInvalidId;
-
- memset(&t,0,sizeof(t));
-
- if( cmThreadCreate(&p->thH,_cmRtNetTestThreadFunc,p,&ctx->rpt) != kOkThRC )
- goto errLabel;
-
- if((rc = cmRtNetAlloc(ctx,&p->netH,rtSubIdx,_cmRtNetTestRecv,p)) != kOkNetRC )
- goto errLabel;
-
- cmRtNetReportSyncEnable(p->netH,true); // enable sync. protocol reporting
-
- if((rc = cmRtNetInitialize(p->netH, bcastAddr, localHostStr, NULL, port )) != kOkNetRC)
- goto errLabel;
-
- if((rc = cmRtNetRegisterEndPoint(p->netH,localEndpStr, 0 )) != kOkNetRC )
- goto errLabel;
-
- if((rc = cmRtNetEndpointHandle(p->netH, localHostStr, localEndpStr, &eH )) != kOkNetRC )
- goto errLabel;
-
- if((srcEndPtId = cmRtNetEndpointId(eH)) == cmInvalidIdx )
- goto errLabel;
-
- if( cmThreadPause(p->thH,0) != kOkThRC )
- goto errLabel;
-
- cmRptPrintf(&ctx->rpt,"%s t=transmit s=sync r=report q=quit\n", localHostStr );
-
- while( (c=getchar()) != 'q' )
- {
- switch(c)
- {
- case 'r':
- cmRtNetReport(p->netH);
- break;
-
- case 's':
- cmRtNetDoSync(p->netH);
- break;
-
- case 't':
- {
- if( cmRtNetSendByLabels(p->netH, srcEndPtId, remoteHostStr, remoteEndpStr, &p->msgVal, sizeof(p->msgVal)) == kOkNetRC )
- p->msgVal += 1;
-
- }
- break;
- }
-
- }
-
- errLabel:
-
- cmThreadDestroy(&p->thH);
-
- cmRtNetFree(&p->netH);
-
-
- return;
-
- }
|