cml/libcm
1
0
Fork 0
Code Issues Pull Requests Releases Wiki Activity

cmRtNet.h/c: Many changes. Completed testing of cmRtNetTest().

Browse Source
This commit is contained in:
kpl 2013-04-28 15:20:33 -07:00
parent 9e65734d66
commit 63bba3ffa7
2 changed files with 297 additions and 201 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

358
cmRtNet.c
View File

@ -9,66 +9,78 @@
#include "cmTime.h"
#include "cmRtSysMsg.h"
// flags for cmRtNetNode_t.flags;
enum
{
kLocalNodeNetFl = 0x01,
kValidNodeNetFl = 0x02
};
// flags for cmRtNet_t.flags
enum
{
kReportSyncNetFl = 0x01
};
typedef enum
{
kHelloSelNetId,
kNodeSelNetId,
kEndpointSelNetId,
kEndpointAckSelNetId,
kDoneSelNetId,
kHelloSelNetId, // broadcast msg (label=node label, id=endpt cnt)
kNodeSelNetId, // define remote node (label=remote node label, id=endpt cnt)
kEndpointSelNetId, // define remote endpt (label=remote endpt label, id=endpt id)
kDoneSelNetId, // declare all endpts sent
kInvalidSelNetId
} cmRtNetSelId_t;
struct cmRtNetNode_str;
typedef struct cmRtNetEnd_str
{
cmChar_t* label;
unsigned id;
struct cmRtNetEnd_str* link;
cmChar_t* label;
unsigned id;
struct cmRtNetNode_str* np; // Owner node.
struct cmRtNetEnd_str* link;
} cmRtNetEnd_t;
typedef struct cmRtNetNode_str
{
cmChar_t* label;
struct sockaddr_in sockaddr;
cmChar_t* addr;
cmUdpPort_t port;
unsigned flags;
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;
cmRtNetEnd_t* ends;
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
{
cmErr_t err;
cmUdpH_t udpH;
cmUdpCallback_t cbFunc;
void* cbArg;
cmRtNetNode_t* nodes;
cmRtNetNode_t* localNode;
unsigned udpRecvBufByteCnt;
unsigned udpTimeOutMs;
unsigned interSyncSendTimeMs;
cmErr_t err; // Error state object
unsigned flags; // See kXXXNetFl above.
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;
// Network synchronization message format
typedef struct
{
cmRtSysMsgHdr_t hdr;
cmRtNetSelId_t selId;
cmRtSysMsgHdr_t hdr; // standard cmRtSys msg header
cmRtNetSelId_t selId; // message selector id (See kXXXSelNetId above)
const cmChar_t* label; // node or endpoint label
unsigned id; // endptCnt or endpoint id
} cmRtNetSyncMsg_t;
cmRtNetH_t cmRtNetNullHandle = cmSTATIC_NULL_HANDLE;
cmRtNetH_t cmRtNetNullHandle = cmSTATIC_NULL_HANDLE;
cmRtNetEndptH_t cmRtNetEndptNullHandle = cmSTATIC_NULL_HANDLE;
cmRtNet_t* _cmRtNetHandleToPtr( cmRtNetH_t h )
{
@ -79,7 +91,8 @@ cmRtNet_t* _cmRtNetHandleToPtr( cmRtNetH_t h )
void _cmRtNetVRpt( cmRtNet_t* p, const cmChar_t* fmt, va_list vl )
{
cmRptVPrintf(p->err.rpt,fmt,vl);
if( cmIsFlag(p->flags,kReportSyncNetFl) )
cmRptVPrintf(p->err.rpt,fmt,vl);
}
void _cmRtNetRpt( cmRtNet_t* p, const cmChar_t* fmt, ... )
@ -143,7 +156,7 @@ void _cmRtNetReleaseNodes( cmRtNet_t* p )
np = nnp;
}
p->nodes = NULL;
p->nodes = NULL;
p->localNode = NULL;
}
@ -188,7 +201,10 @@ cmRtNetRC_t _cmRtNetCreateNode( cmRtNet_t* p, const cmChar_t* label, const cmCha
np = cmMemAllocZ(cmRtNetNode_t,1);
np->label = cmMemAllocStr(label);
np->sockaddr = *saddr;
if( saddr != NULL )
np->sockaddr = *saddr;
np->addr = addr==NULL ? NULL : cmMemAllocStr(addr);
np->port = port;
np->flags = flags;
@ -235,12 +251,22 @@ cmRtNetRC_t _cmRtNetCreateEndpoint( cmRtNet_t* p, cmRtNetNode_t* np, const cmCha
ep->label = cmMemAllocStr(endPtLabel);
ep->id = endPtId;
ep->link = np->ends;
np->ends = ep;
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); }
@ -287,10 +313,9 @@ cmRtNetRC_t _cmRtNetSendSyncMsg( cmRtNet_t* p, cmRtNetNode_t* np, cmRtNetSelId_t
if((rc = _cmRtNetSerializeSyncMsg(p,&m,buf,n)) != kOkNetRC )
return rc;
// send the msg
if( np==p->localNode )
udpRC = cmUdpSend2(p->udpH, buf, n, "255.255.255.255", np->port );
if( selId == kHelloSelNetId )
udpRC = cmUdpSend2(p->udpH, buf, n, p->bcastAddr, np->port );
else
udpRC = cmUdpSendTo(p->udpH, buf, n, &np->sockaddr );
@ -317,11 +342,12 @@ cmRtNetRC_t _cmRtNetFree( cmRtNet_t* p )
_cmRtNetReleaseNodes(p);
cmMemFree(p->bcastAddr);
cmMemFree(p);
return rc;
}
cmRtNetRC_t cmRtNetAlloc( cmCtx_t* ctx, cmRtNetH_t* hp, cmUdpCallback_t cbFunc, void* cbArg )
{
cmRtNetRC_t rc;
@ -340,7 +366,6 @@ cmRtNetRC_t cmRtNetAlloc( cmCtx_t* ctx, cmRtNetH_t* hp, cmUdpCallback_t cbFunc,
p->udpTimeOutMs = 50;
p->udpRecvBufByteCnt = 8192;
p->interSyncSendTimeMs = 10;
p->cbFunc = cbFunc;
p->cbArg = cbArg;
@ -386,7 +411,7 @@ cmUdpH_t cmRtNetUdpPortHandle( cmRtNetH_t h )
return p->udpH;
}
cmRtNetRC_t _cmRtNetSendEndpointReplyMsg( cmRtNet_t* p, cmRtNetNode_t* np )
cmRtNetRC_t _cmRtNetSendEndpointReplyMsg( cmRtNet_t* p, cmRtNetNode_t* np, cmRtNetSelId_t srcSelId )
{
cmRtNetRC_t rc = kOkNetRC;
cmRtNetEnd_t* ep;
@ -398,37 +423,58 @@ cmRtNetRC_t _cmRtNetSendEndpointReplyMsg( cmRtNet_t* p, cmRtNetNode_t* np )
if( np == NULL )
return cmErrMsg(&p->err,kNodeNotFoundNetRC,"The net node associated with an endpoint reply was not found.");
// if all of the endpoints have been sent to this node ...
if((ep = _cmRtNetIndexToEndpoint(p,p->localNode,np->endPtIdx)) == NULL )
// if we got here by receiving a 'done' msg from the remote node ...
if( srcSelId == kDoneSelNetId )
{
if( np->endPtIdx == p->localNode->endPtCnt )
{
selId = kDoneSelNetId;
rptLabel = "done";
}
// ... 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
{
selId = kEndpointAckSelNetId;
rptLabel = "ep ack";
}
np->flags = cmSetFlag(np->flags,kValidNodeNetFl);
}
else
// attempt to get the next local endpoint to send ...
if((ep = _cmRtNetIndexToEndpoint(p,p->localNode,np->endPtIdx)) != NULL )
{
msgLabel = ep->label;
msgLabel = ep->label; // ... send next local endpoint
msgId = ep->id;
}
else // .... all local endpoints have been sent
{
selId = kInvalidSelNetId;
rptLabel = "done";
// notify the remote node that all endpoints have been sent
if((rc = _cmRtNetSendSyncMsg(p,np,selId,msgLabel,msgId )) != 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));
// 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;
}
}
np->endPtIdx += 1;
// selId is set to kInvalidSelNetId when we encounter the (stop) criteria
if( selId != kInvalidSelNetId )
{
if((rc = _cmRtNetSendSyncMsg(p,np,selId,msgLabel,msgId )) != 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(cmRtNetSyncMsg_t) && m->hdr.selId == kNetSyncSelRtId;
}
// When the network message recieve function (See cmRtNetAlloc() 'cbFunc')
@ -438,10 +484,10 @@ cmRtNetRC_t _cmRtNetSendEndpointReplyMsg( cmRtNet_t* p, cmRtNetNode_t* np )
cmRtNetRC_t _cmRtNetSyncModeRecv( cmRtNet_t* p, const char* data, unsigned dataByteCnt, const struct sockaddr_in* fromAddr )
{
cmRtNetRC_t rc = kOkNetRC;
cmRtNetSyncMsg_t m;
cmRtNetSyncMsg_t m;
m.label = NULL;
assert( cmRtNetIsSyncModeMsg(data,dataByteCnt));
assert( _cmRtNetIsSyncModeMsg(data,dataByteCnt));
if( _cmRtNetDeserializeSyncMsg(data,dataByteCnt,&m) != kOkNetRC )
{
@ -449,14 +495,32 @@ cmRtNetRC_t _cmRtNetSyncModeRecv( cmRtNet_t* p, const char* data, unsigned data
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 && 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 ...
@ -472,17 +536,19 @@ cmRtNetRC_t _cmRtNetSyncModeRecv( cmRtNet_t* p, const char* data, unsigned data
if(( rc = _cmRtNetCreateNode(p,m.label,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, NULL, p->localNode->endPtCnt );
rc = _cmRtNetSendSyncMsg( p, np, kNodeSelNetId, p->localNode->label, p->localNode->endPtCnt );
break;
case kNodeSelNetId:
_cmRtNetRpt(p,"rcv node\n");
_cmRtNetSendEndpointReplyMsg( p, np ); // reply with first endpoint
_cmRtNetSendEndpointReplyMsg( p, np, m.selId ); // reply with first endpoint
break;
default:
@ -492,17 +558,15 @@ cmRtNetRC_t _cmRtNetSyncModeRecv( cmRtNet_t* p, const char* data, unsigned data
}
break;
case kEndpointAckSelNetId:
case kDoneSelNetId:
rc = _cmRtNetSendEndpointReplyMsg(p,np);
//case kEndpointAckSelNetId:
rc = _cmRtNetSendEndpointReplyMsg(p,np,m.selId);
break;
case kEndpointSelNetId:
{
cmRtNetEnd_t* ep;
// verify the remote node exists.
if( np == NULL )
{
@ -521,7 +585,7 @@ cmRtNetRC_t _cmRtNetSyncModeRecv( cmRtNet_t* p, const char* data, unsigned data
}
// reply with a local endpoint or 'done' msg
rc = _cmRtNetSendEndpointReplyMsg( p, np );
rc = _cmRtNetSendEndpointReplyMsg( p, np, m.selId );
}
break;
@ -540,27 +604,21 @@ void _cmRtNetRecv( void* cbArg, const char* data, unsigned dataByteCnt, const st
{
cmRtNet_t* p = (cmRtNet_t*)cbArg;
if( cmRtNetIsSyncModeMsg(data,dataByteCnt))
if( _cmRtNetIsSyncModeMsg(data,dataByteCnt))
_cmRtNetSyncModeRecv(p,data,dataByteCnt,fromAddr);
else
p->cbFunc(p->cbArg,data,dataByteCnt,fromAddr);
}
cmRtNetRC_t cmRtNetRegisterLocalNode( cmRtNetH_t h, const cmChar_t* nodeLabel, const cmChar_t* ipAddr, cmUdpPort_t port )
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;
struct sockaddr_in sockaddr;
// release the local node
if( p->localNode != NULL )
{
_cmRtNetReleaseNode(p,p->localNode);
p->localNode = NULL;
}
if((rc = cmRtNetFinalize(h)) != kOkNetRC )
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 )
@ -569,19 +627,13 @@ cmRtNetRC_t cmRtNetRegisterLocalNode( cmRtNetH_t h, const cmChar_t* nodeLabel, c
goto errLabel;
}
// get the socket address
if( cmUdpInitAddr(p->udpH, ipAddr, port, &sockaddr ) != kOkUdpRC )
{
rc = cmErrMsg(&p->err,kUdpPortFailNetRC,"IP::port to socket address conversion failed.");
goto errLabel;
}
// create the local node
if((rc = _cmRtNetCreateNode(p,nodeLabel, ipAddr, port, &sockaddr, kLocalNodeNetFl, 0)) != kOkNetRC )
if((rc = _cmRtNetCreateNode(p,nodeLabel, 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 )
@ -609,23 +661,22 @@ cmRtNetRC_t cmRtNetRegisterEndPoint( cmRtNetH_t h, const cmChar_t* endPtLabel, u
return rc;
}
cmRtNetRC_t cmRtNetClearAll( cmRtNetH_t h )
cmRtNetRC_t cmRtNetFinalize( cmRtNetH_t h )
{
cmRtNet_t* p = _cmRtNetHandleToPtr(h);
_cmRtNetReleaseNodes(p);
return kOkNetRC;
}
cmRtNetRC_t cmRtNetBeginSyncMode( cmRtNetH_t h )
cmRtNetRC_t cmRtNetDoSync( cmRtNetH_t h )
{
cmRtNet_t* p = _cmRtNetHandleToPtr(h);
// broadcast 'node' msg
return _cmRtNetSendSyncMsg( p, p->localNode, kHelloSelNetId, NULL, p->localNode->endPtCnt );
return _cmRtNetSendSyncMsg( p, p->localNode, kHelloSelNetId, p->localNode->label, p->localNode->endPtCnt );
}
cmRtNetRC_t cmRtNetReceive( cmRtNetH_t h )
{
cmRtNetRC_t rc = kOkNetRC;
@ -640,24 +691,70 @@ cmRtNetRC_t cmRtNetReceive( cmRtNetH_t h )
return rc;
}
bool cmRtNetIsSyncModeMsg( const void* data, unsigned dataByteCnt )
cmRtNetRC_t cmRtNetEndpointHandle( cmRtNetH_t h, const cmChar_t* nodeLabel, const cmChar_t* endptLabel, cmRtNetEndptH_t* hp )
{
cmRtNetSyncMsg_t* m = (cmRtNetSyncMsg_t*)data;
return dataByteCnt >= sizeof(cmRtNetSyncMsg_t) && m->hdr.selId == kNetSyncSelRtId;
cmRtNetRC_t rc = kOkNetRC;
cmRtNet_t* p = _cmRtNetHandleToPtr(h);
cmRtNetNode_t* np;
cmRtNetEnd_t* ep;
if(( np = _cmRtNetFindNode(p,nodeLabel)) == 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;
}
cmRtNetRC_t cmRtNetSend( cmRtNetH_t h, cmRtNetEndptH_t epH, const void* msg, unsigned msgByteCnt )
{
cmRtNetRC_t rc = kOkNetRC;
cmRtNet_t* p = _cmRtNetHandleToPtr(h);
cmRtNetEnd_t* ep = (cmRtNetEnd_t*)epH.h;
assert( ep != NULL );
unsigned dN = sizeof(unsigned) + msgByteCnt;
char data[ dN ];
unsigned *hdr = (unsigned*)data;
hdr[0] = ep->id;
memcpy(hdr+1,msg,msgByteCnt);
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 cmRtNetSendByLabels( cmRtNetH_t h, 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,epH,msg,msgByteCnt);
}
unsigned cmRtNetEndPointIndex( cmRtNetH_t h, const cmChar_t* nodeLabel, const cmChar_t* endPtLabel )
{
//cmRtNet_t* p = _cmRtNetHandleToPtr(h);
return cmInvalidIdx;
}
cmRtNetRC_t cmRtNetSend( cmRtNetH_t h, unsigned endPointIndex, const void* msg, unsigned msgByteCnt )
bool cmRtNetReportSyncEnable( cmRtNetH_t h, bool enableFl )
{
cmRtNetRC_t rc = kOkNetRC;
//cmRtNet_t* p = _cmRtNetHandleToPtr(h);
return rc;
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 )
@ -699,17 +796,20 @@ void cmRtNetReport( cmRtNetH_t h )
typedef struct
{
cmThreadH_t thH;
cmRtNetH_t netH;
cmRtNetH_t netH;
unsigned msgVal;
} _cmRtNetTest_t;
void _cmRtNetTestRecv( void* cbArg, const char* data, unsigned dataByteCnt, const struct sockaddr_in* fromAddr )
{
//_cmRtNetTest_t* p = (_cmRtNetTest_t*)cbArg;
unsigned* hdr = (unsigned*)data;
printf("%i %i\n",hdr[0],hdr[1]);
}
bool _cmRtNetTestThreadFunc(void* param)
{
_cmRtNetTest_t* p = (_cmRtNetTest_t*)param;
@ -728,10 +828,15 @@ void cmRtNetTest( cmCtx_t* ctx, bool mstrFl )
{
char c;
_cmRtNetTest_t t;
const cmChar_t* hostNameStr;
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";
memset(&t,0,sizeof(t));
if( cmThreadCreate(&p->thH,_cmRtNetTestThreadFunc,p,&ctx->rpt) != kOkThRC )
@ -740,32 +845,19 @@ void cmRtNetTest( cmCtx_t* ctx, bool mstrFl )
if((rc = cmRtNetAlloc(ctx,&p->netH,_cmRtNetTestRecv,p)) != kOkNetRC )
goto errLabel;
hostNameStr = cmRtNetLocalHostName(p->netH);
if( hostNameStr == NULL )
hostNameStr = "<no-host-name>";
cmRtNetReportSyncEnable(p->netH,true); // enable sync. protocol reporting
if((rc = cmRtNetRegisterLocalNode(p->netH, hostNameStr, NULL, port )) != kOkNetRC)
if((rc = cmRtNetInitialize(p->netH, bcastAddr, localHostStr, NULL, port )) != kOkNetRC)
goto errLabel;
if( mstrFl )
{
if((rc = cmRtNetRegisterEndPoint(p->netH,"thunk_ep0", 0 )) != kOkNetRC )
goto errLabel;
if(( rc = cmRtNetBeginSyncMode(p->netH)) != kOkNetRC )
goto errLabel;
}
else
{
if((rc = cmRtNetRegisterEndPoint(p->netH,"whirl_ep0", 0 )) != kOkNetRC )
goto errLabel;
}
if((rc = cmRtNetRegisterEndPoint(p->netH,localEndpStr, 0 )) != kOkNetRC )
goto errLabel;
if( cmThreadPause(p->thH,0) != kOkThRC )
goto errLabel;
cmRptPrintf(&ctx->rpt,"%s q=quit\n", mstrFl ? "Master: " : "Slave: ");
cmRptPrintf(&ctx->rpt,"%s t=transmit s=sync r=report q=quit\n", localHostStr );
while( (c=getchar()) != 'q' )
{
switch(c)
@ -773,6 +865,18 @@ void cmRtNetTest( cmCtx_t* ctx, bool mstrFl )
case 'r':
cmRtNetReport(p->netH);
break;
case 's':
cmRtNetDoSync(p->netH);
break;
case 't':
{
if( cmRtNetSendByLabels(p->netH, remoteHostStr, remoteEndpStr, &p->msgVal, sizeof(p->msgVal)) == kOkNetRC )
p->msgVal += 1;
}
break;
}
}

140
cmRtNet.h
View File

@ -11,21 +11,23 @@ extern "C" {
kUdpPortFailNetRC,
kInvalidLabelNetRC,
kDuplLabelNetRC,
kDuplLocalNetRC,
kDuplEndNetRC,
kDuplLocalNetRC,
kThreadFailNetRC,
kBufToSmallNetRC,
kNodeNotFoundNetRC,
kNodeStateErrNetRC,
kTimeOutErrNetRC,
kEndNotFoundNetRC,
kLocalNodeNetRC,
kSyncFailNetRC,
kNodeEndCntErrNetRC
};
typedef cmRC_t cmRtNetRC_t;
typedef cmHandle_t cmRtNetH_t;
typedef cmHandle_t cmRtNetEndptH_t;
extern cmRtNetH_t cmRtNetNullHandle;
extern cmRtNetH_t cmRtNetNullHandle;
extern cmRtNetEndptH_t cmRtNetEndptNullHandle;
// 'cbFunc' will be called within the context of cmRtNetReceive() to receive
// incoming network messages.
@ -34,99 +36,89 @@ extern "C" {
bool cmRtNetIsValid( cmRtNetH_t h );
// Get the local host name for this machine. This function
// is synonomous with gethostname().
const cmChar_t* cmRtNetLocalHostName( cmRtNetH_t h );
// Create a network node.
// The 'nodeLabel' refers to a network device cfg. (see cmDevCfg).
// Set 'ipAddr' to NULL if this is the local node.
// During sync mode this node will attempt to sync with all
// nodes in the node list.
cmRtNetRC_t cmRtNetRegisterLocalNode( cmRtNetH_t h, const cmChar_t* nodeLabel, const cmChar_t* ipAddr, cmUdpPort_t ipPort );
// Initialize the local network node.
// 'bcastAddr' is the network broadcast address (e.g. 192.168.15.255).
// 'nodeLabel' is the local network node label
// 'ipAddr' may be set to NULL to use any available IP address.
// 'ipPort' refers to the socket port (which may need to be made available
// by the machine firewall cfg.)
cmRtNetRC_t cmRtNetInitialize( cmRtNetH_t h, const cmChar_t* bcastAddr, const cmChar_t* nodeLabel, const cmChar_t* ipAddr, cmUdpPort_t ipPort );
// Register the local endpoints.
// Endpoints may only be registered once the network is initialized via
// cmRtNetInitialize().
// Remote nodes will be able to send messages to these endpoints by
// referring to (nodeLabel/endPtLabel)
cmRtNetRC_t cmRtNetRegisterEndPoint( cmRtNetH_t h, const cmChar_t* endPtLabel, unsigned endPtId );
// Delete all nodes and endpoints.
cmRtNetRC_t cmRtNetClearAll( cmRtNetH_t h );
// Go into 'sync' node.
// When a node enters sync mode it systematically transmits all of it's
// local endpoint information to each registered remote node. Prior to
// entering sync mode a node must therefore have been setup with a list
// of remote nodes (via cmRtNetCreateNode()) and a list of local endpoints
// (cmRtNetRegisterEndpoint()). During sync mode a node sends it's local
// endpoint list to each registered remote node. When a remote node receives
// an endpoint it updates it's own remote node/endpoint
// list.
cmRtNetRC_t cmRtNetBeginSyncMode( cmRtNetH_t h );
cmRtNetRC_t cmRtNetFinalize( cmRtNetH_t h );
// Broadcast the 'hello' to all machines listening on the
// broadcast addresss. This starts the synchronization sequence
cmRtNetRC_t cmRtNetDoSync( cmRtNetH_t h );
// This function must be polled to receive incoming network messages
// via the callback funcion 'cbFunc' as passed to cmRtNetAlloc()
cmRtNetRC_t cmRtNetReceive( cmRtNetH_t h );
bool cmRtNetIsSyncModeMsg( const void* data, unsigned dataByteCnt );
// Get an end point handle for use with cmRtNetSend.
cmRtNetRC_t cmRtNetEndpointHandle( cmRtNetH_t h, const cmChar_t* nodeLabel, const cmChar_t* endptLabel, cmRtNetEndptH_t* hp );
unsigned cmRtNetEndPointIndex( cmRtNetH_t h, const cmChar_t* nodeLabel, const cmChar_t* endPtLabel );
// Send a message to a remote endpoint.
cmRtNetRC_t cmRtNetSend( cmRtNetH_t h, cmRtNetEndptH_t epH, const void* msg, unsigned msgByteCnt );
cmRtNetRC_t cmRtNetSend( cmRtNetH_t h, unsigned endPointIndex, const void* msg, unsigned msgByteCnt );
// Send a message to a remote endpoint. This function is a composite
// of cmRtNetEndpointHandle() and cmRtNetSend().
cmRtNetRC_t cmRtNetSendByLabels( cmRtNetH_t h, const cmChar_t* nodeLabel, const cmChar_t* endptLabel, const void* msg, unsigned msgByteCnt );
void cmRtNetReport( cmRtNetH_t h );
// Enable/disable synchronization protocol reporting.
// Return the previous state of the report sync. flag.
bool cmRtNetReportSyncEnable( cmRtNetH_t h, bool enableFl );
bool cmRtNetReportSyncIsEnabled( cmRtNetH_t h );
void cmRtNetReport( cmRtNetH_t h );
void cmRtNetTest( cmCtx_t* ctx, bool mstrFl );
void cmRtNetTest( cmCtx_t* ctx, bool mstrFl );
/*
Master:
cmRtNetBeginSyncMode().
while( cmRtNetIsSyncMode())
{
// Give the master an oppurtunity to advance it's sync mode state.
// When the master is has sync'd with all remote nodes in it's
// remote node list then it will automatically exit sync mode.
cmRtNetSyncModeSend()
}
_myNetRecv(dataV,dataN,addr)
{
if( cmRtNetIsSyncModeMsg(dataV,dataN) )
cmRtNetSyncModeRecv(dataV,dataN,addr)
}
Synchronization Protocol:
Machine A Machine B
================================== ====================================
broadcast 'hello' --------------------> create node-A w/ ei=0 -------+
|
+<-- create node-B w/ ei=0 <----------- send 'node' <----------------+
|
+--> switch(ei,m_t)
| ei < en : send endpt[ei++] ---> create endpt[] on node-A -->+
| |
| ei == en : ++ei,send 'done' -------------------------------->+ |
| |
| m_t!='done' : send 'done' -------------------------------->+ |
| |
| (stop) : |
| |
| v
| switch(ei,m_t)
+<-- create endpt[] on node-B <--------- send endpt[ei++] : ei < en
|
+<--------------------------------------- send 'done',++ei : ei == en
|
+<--------------------------------------- send 'done' : m_t!= 'done'
: (stop)
The 'master' is the machine which cmRtNetBeginSyncMode() is called on.
1) 'master' sends local endpoints to all registered remote nodes.
2) When a 'slave' receives the kDoneSelNetId msg it transmits
it's own local endpoints back to the master.
a. Each node in the node list has a type id:
1. local
2. registered - remote node that was explicitely registered on a master
3. received - remote node that was received from a master
b.
1. All nodes are created in the 'send-hello' state.
2. If a master machine is in 'sync-mode' then it systematically sends
each of it's local endpoints to all 'registered' nodes.
3. When a slave machine recives a 'hello' it creates a
'received' node.
4. When a slave machine recieves a 'done' it enters sync mode
and systematically sends each of its local endpoints to
the 'done' source.
Protocol:
1. A: on init bcast 'hello'
2. B: on 'hello' - create node-A w/ ei=0 - send 'node'
3. A: on 'node' - create node-B w/ ei=0 - send first 'endpt'
4. B: on 'endpt' - create endpt on node-A - ei!=en ? send 'endpt' or send 'done'
5. A: on 'endpt' - create endpt on node-B - ei!=en ? send 'endpt' or send 'done'
6. B: on 'done' - mark node-A as 'valid'
7. A: on 'done' - mark node-B as 'valid'.
Notes:
1) 'ei' is the index of the next local end point to transmit.
2) 'en' is the count of local endpoints.
3) 'm_t' is the msg type (i.e.'hello','node','endpoint','done')
of the incoming message.
*/