libcm/cmRtNet.c

#include "cmGlobal.h"
#include "cmRpt.h"
#include "cmErr.h"
#include "cmCtx.h"
#include "cmMem.h"
#include "cmMallocDebug.h"
#include "cmUdpPort.h"
#include "cmRtNet.h"
#include "cmTime.h"
#include "cmRtSysMsg.h"

enum
{
  kLocalNetFl    = 0x01,
  kSockAddrNetFl = 0x02,
  kRegNodeNetFl  = 0x04,
  kRecvNodeNetFl = 0x08
};

typedef enum
{
  kSendHelloStNetId = 0,
  kWaitHelloAckStNetId,
  kSendEndpointStNetId,
  kWaitEndpointAckStNetId,
  kDoneStNetId,
  kErrorStNetId,
  kInvalidStNetId,
} cmRtNetNodeState_t;

typedef enum
{
  kHelloSelNetId,
  kHelloAckSelNetId,
  kEndpointSelNetId,
  kEndpointAckSelNetId,
  kDoneSelNetId
} cmRtNetSelId_t;

typedef struct cmRtNetEnd_str
{
  cmChar_t*              endPtLabel;
  unsigned               endPtId;
  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;
  cmRtNetNodeState_t      state;
  unsigned                epIdx;         // tracks the next endpoint to send during sync-mode
  cmTimeSpec_t            lastSendTime;
  cmRtNetEnd_t*           ends;
  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;
  bool            syncModeFl;
  bool            masterFl;
  unsigned        udpRecvBufByteCnt;
  unsigned        udpTimeOutMs;
  unsigned        interSyncSendTimeMs;
} cmRtNet_t;


typedef struct
{
  cmRtSysMsgHdr_t hdr;
  cmRtNetSelId_t  selId;
  const cmChar_t* endPtLabel;
  unsigned        endPtId;
} cmRtNetSyncMsg_t;

cmRtNetH_t cmRtNetNullHandle = cmSTATIC_NULL_HANDLE;

cmRtNet_t* _cmRtNetHandleToPtr( cmRtNetH_t h )
{
  cmRtNet_t* p = (cmRtNet_t*)h.h;
  assert( p != NULL );
  return p;
}

void _cmRtNetVRpt( cmRtNet_t* p, const cmChar_t* fmt, va_list vl )
{
  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 )
{
  if( label == NULL )
    return NULL;

  cmRtNetNode_t* np = p->nodes;
  for(; np!=NULL; np=np->link)
    if( strcmp(label,np->label)==0)
      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( cmIsFlag(np->flags,kSockAddrNetFl) && 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->endPtLabel);
    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 )
{
  // we should never release the local node via this function
  assert( np != p->localNode );

  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, const cmChar_t* addr, cmUdpPort_t port, const struct sockaddr_in* saddr, unsigned flags )
{
  cmRtNetRC_t rc = kOkNetRC;
  cmRtNetNode_t* np;

  if( label == NULL )
    return cmErrMsg(&p->err,kInvalidLabelNetRC,"A null or blank node label was encountered.");

  if((np = _cmRtNetFindNode(p,label)) != NULL )
    return cmErrMsg(&p->err,kDuplLabelNetRC,"The node label '%s' is already in use.",cmStringNullGuard(label));

  bool localNodeFl = addr==NULL && saddr==NULL;
  
  if( localNodeFl && p->localNode != NULL )
    return cmErrMsg(&p->err,kDuplLocalNetRC,"The local node '%s' has already been set.",cmStringNullGuard(p->localNode->label));

  np = cmMemAllocZ(cmRtNetNode_t,1);
  np->label = cmMemAllocStr(label);
  np->addr  = addr==NULL ? NULL : cmMemAllocStr(addr);
  np->port  = port;
  np->flags = cmEnaFlag(flags,kLocalNetFl,localNodeFl);
  np->link  = p->nodes;
  p->nodes  = np;

  if( localNodeFl )
    p->localNode = np;

  if( saddr != NULL )
    np->sockaddr = *saddr;
  else
  {
    if( cmUdpInitAddr(p->udpH, np->addr, np->port, &np->sockaddr ) != kOkUdpRC )
    {
      rc = cmErrMsg(&p->err,kUdpPortFailNetRC,"IP::port to socket address conversion failed.");
      goto errLabel;
    }
  }

  np->flags = cmSetFlag(np->flags,kSockAddrNetFl);

 errLabel:
  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->endPtLabel,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->endPtLabel = cmMemAllocStr(endPtLabel);
  ep->endPtId    = endPtId;
  ep->link      = np->ends;
  np->ends      = ep;

  return rc;
}

unsigned _cmRtNetSyncMsgSerialByteCount( const cmRtNetSyncMsg_t* m )
{ return sizeof(cmRtNetSyncMsg_t) + (m->endPtLabel==NULL ? 1 : strlen(m->endPtLabel) + 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->endPtLabel==NULL ? "" : m->endPtLabel);
  return kOkNetRC;
}

cmRtNetRC_t _cmRtNetDeserializeSyncMsg( const void* buf, unsigned n, cmRtNetSyncMsg_t* m )
{
  assert( n > sizeof(*m));
  memcpy(m,buf,sizeof(*m));
  const cmRtNetSyncMsg_t* mp = (const cmRtNetSyncMsg_t*)buf;
  const cmChar_t*   s  = (const cmChar_t*)(mp+1);
  m->endPtLabel = cmMemAllocStr(s);
  return kOkNetRC;
}

cmRtNetRC_t _cmRtNetSendSyncMsg( cmRtNet_t* p, cmRtNetNode_t* np, cmRtNetSelId_t selId, const cmChar_t* endPtLabel, unsigned endPtId, cmRtNetNodeState_t nextStId )
{
  cmRtNetSyncMsg_t m;
  cmRtNetRC_t      rc = kOkNetRC;
  cmUdpRC_t        udpRC = kOkUdpRC;

  m.hdr.rtSubIdx = cmInvalidIdx;
  m.hdr.selId    = kNetSyncSelRtId;
  m.selId        = selId;
  m.endPtLabel    = endPtLabel;
  m.endPtId       = endPtId;

  // 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;

  // store this nodes current sync state
  cmRtNetNodeState_t orgState = np->state;

  if( nextStId != kInvalidStNetId )
    np->state = nextStId;

 
  // send the msg
  if( cmIsFlag(np->flags,kSockAddrNetFl) == false )
    udpRC = cmUdpSend2(p->udpH, buf, n, np->addr, 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.");
    np->state = orgState;  // restore node state so we can try again
  }
  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);
  return rc;
}


cmRtNetRC_t cmRtNetAlloc( cmCtx_t* ctx, cmRtNetH_t* hp, 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->udpTimeOutMs        = 50;
  p->udpRecvBufByteCnt   = 8192;
  p->interSyncSendTimeMs = 10;
  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 cmRtNetCreateNode( cmRtNetH_t h, const cmChar_t* nodeLabel, const cmChar_t* ipAddr, cmUdpPort_t port )
{
  cmRtNet_t*  p = _cmRtNetHandleToPtr(h);
  cmRtNetRC_t rc;

  // create a node
  if((rc = _cmRtNetCreateNode(p,nodeLabel,ipAddr, port, NULL, kRegNodeNetFl)) != kOkNetRC )
    return rc;

  // if this is not the local node
  if( ipAddr != NULL )
    return rc;

  // if this is the local node then initialze the local socket
  if( cmUdpInit(p->udpH,port,kNonBlockingUdpFl,p->cbFunc,p->cbArg,NULL,0,p->udpRecvBufByteCnt,p->udpTimeOutMs) != kOkUdpRC )
  {
    rc = cmErrMsg(&p->err,kUdpPortFailNetRC,"The UDP port initialization failed.");
    goto errLabel;
  }

  // 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;
}

cmRtNetRC_t cmRtNetRegisterEndPoint( cmRtNetH_t h, const cmChar_t* endPtLabel, unsigned endPtId )
{
  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.");

  return _cmRtNetCreateEndpoint(p, p->localNode,endPtLabel,endPtId );

}

cmRtNetRC_t cmRtNetClearAll( cmRtNetH_t h )
{
  cmRtNet_t* p = _cmRtNetHandleToPtr(h);
  _cmRtNetReleaseNodes(p);
  return kOkNetRC;
}

cmRtNetRC_t cmRtNetBeginSyncMode( cmRtNetH_t h )
{
  cmRtNetRC_t rc = kOkNetRC;
  
  cmRtNet_t* p = _cmRtNetHandleToPtr(h);

  p->syncModeFl = true;
  p->masterFl   = true;
  return rc;
}

bool      cmRtNetIsInSyncMode(  cmRtNetH_t h )
{
  cmRtNet_t* p = _cmRtNetHandleToPtr(h);

  return p->syncModeFl;
}


// Used by slaves to send the master an 'ack' msg.
cmRtNetRC_t _cmRtNetSendAck( cmRtNet_t* p, cmRtNetSelId_t ackSelId, const struct sockaddr_in* saddr )
{
  cmRtNetNode_t* np;

  if((np = _cmRtNetFindNodeFromSockAddr(p,saddr)) == NULL )
    return cmErrMsg(&p->err,kNodeNotFoundNetRC,"The net node associated with an ack cmd was not found. Ack not sent.");

  return _cmRtNetSendSyncMsg(p,np,ackSelId,NULL,cmInvalidId,kInvalidStNetId);
}

// Used by master to update state upon receipt of 'ack' msg
cmRtNetRC_t _cmRtNetRecvAck( cmRtNet_t* p, const struct sockaddr_in* fromAddr, cmRtNetNodeState_t expectedState, cmRtNetNodeState_t nextState )
{
  cmRtNetNode_t* np;
  cmRtNetRC_t rc = kOkNetRC;

  if((np = _cmRtNetFindNodeFromSockAddr(p,fromAddr)) == NULL )
  {
    rc = cmErrMsg(&p->err,kNodeNotFoundNetRC,"The net node associated with a  ack receive was not found.");
    goto errLabel;
  }

  if( np->state != expectedState )
  {
    rc = cmErrMsg(&p->err,kNodeStateErrNetRC,"Node '%s' expected in state %i was in state %i.",cmStringNullGuard(np->label),kWaitHelloAckStNetId,np->state);
    np->state = kErrorStNetId;
    goto errLabel;
  }

  np->state = nextState;

  // if we are about to send another endpoint - incr the endpoint index
  if( nextState == kSendEndpointStNetId )
    np->epIdx += 1;

 errLabel:
  return rc;
}


cmRtNetRC_t  cmRtNetSyncModeRecv( cmRtNetH_t h, const char* data, unsigned dataByteCnt, const struct sockaddr_in* fromAddr )
{
  cmRtNet_t*       p  = _cmRtNetHandleToPtr(h);
  cmRtNetRC_t      rc = kOkNetRC;
  cmRtNetNode_t*   np = NULL;
  cmRtNetSyncMsg_t m;

  m.endPtLabel = 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;
  }

  assert( m.hdr.selId == kNetSyncSelRtId );

  switch( m.selId )
  {
    
    case kHelloSelNetId: // slave response
      {
        _cmRtNetRpt(p,"rcv hello\n");

        // attempt to locate the remote node which sent the endpoint 
        if((np = _cmRtNetFindNodeFromSockAddr(p,fromAddr)) != NULL )
        {
          // delete the existing node 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
        if(( rc = _cmRtNetCreateNode(p,m.endPtLabel,NULL,0,fromAddr,kRecvNodeNetFl)) != kOkNetRC )
          goto errLabel;

        // send an ackknowledgement of the 'hello' msg
        rc = _cmRtNetSendAck(p,kHelloAckSelNetId,fromAddr);

      }
      break;

    
    case kEndpointSelNetId: // slave response
      {
        cmRtNetEnd_t* ep;

        _cmRtNetRpt(p,"rcv endpoint\n");

        // locate the remote node which sent the endpoint
        if((np = _cmRtNetFindNodeFromSockAddr(p,fromAddr)) == 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.endPtLabel)) != NULL )
          ep->endPtId = m.endPtId; // the endpoint was found update the endPtId
        else
        {
          // create a local proxy for the endpoint
          if((rc = _cmRtNetCreateEndpoint(p,np,m.endPtLabel,m.endPtId)) != kOkNetRC )
            goto errLabel;
        }

        // ack. the endpoint msg
        rc = _cmRtNetSendAck(p,kEndpointAckSelNetId,fromAddr);
      }
      break;

    case kDoneSelNetId:
      {
        _cmRtNetRpt(p,"rcv done\n");

        if( p->masterFl==false  )
          p->syncModeFl = true;
      }
      break;

    case kHelloAckSelNetId: // master response
      {
        assert( p->syncModeFl );
        _cmRtNetRpt(p,"rcv hello ack\n");
        rc = _cmRtNetRecvAck(p,fromAddr,kWaitHelloAckStNetId,kSendEndpointStNetId);
      }
      break;

    case kEndpointAckSelNetId: // master response
      {
        assert( p->syncModeFl );  
        _cmRtNetRpt(p,"rcv endpoint ack\n");
        rc = _cmRtNetRecvAck(p,fromAddr,kWaitEndpointAckStNetId,kSendEndpointStNetId);
      }
      break;

    default:
      break;
  }

 errLabel:

  cmMemFree((cmChar_t*)m.endPtLabel);
  return rc;
}


cmRtNetRC_t _cmRtNetSendNodeSync( cmRtNet_t* p, cmRtNetNode_t* np )
{
  cmRtNetRC_t rc = kOkNetRC;

  switch( np->state )
  {
    case kSendHelloStNetId:
      {
        np->epIdx = -1;

        // send a 'hello' to this remote node
        if((rc = _cmRtNetSendSyncMsg(p,np,kHelloSelNetId,p->localNode->label, cmInvalidId, kWaitHelloAckStNetId )) != kOkNetRC )
          rc = cmErrMsg(&p->err,rc,"Send 'hello' to %s:%s:%i failed.",cmStringNullGuard(np->label),cmStringNullGuard(np->addr),np->port);
        else
          _cmRtNetRpt(p,"%s sent hello\n",cmStringNullGuard(np->label));
      }
      break;

    case kSendEndpointStNetId:
      {
        cmRtNetEnd_t* ep;

        // if all of the endpoints have been sent to this node ...
        if((ep = _cmRtNetIndexToEndpoint(p,p->localNode,np->epIdx)) == NULL )
        {
          // notify the remote node that all endpoints have been sent
          if((rc = _cmRtNetSendSyncMsg(p,np,kDoneSelNetId,p->localNode->label,cmInvalidId, kDoneStNetId )) != kOkNetRC )
            rc = cmErrMsg(&p->err,rc,"Send 'done' to %s:%s:%i failed.",cmStringNullGuard(np->label),cmStringNullGuard(np->addr),np->port);
          else
            _cmRtNetRpt(p,"Node %s done.\n",cmStringNullGuard(np->label));
        }
        else
        {
          // send an endpoint to this node 
          if((rc = _cmRtNetSendSyncMsg(p,np,kEndpointSelNetId,ep->endPtLabel, ep->endPtId, kWaitEndpointAckStNetId )) != kOkNetRC )
            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);
          else
            _cmRtNetRpt(p,"%s sent endpoint %s\n",cmStringNullGuard(np->label),cmStringNullGuard(ep->endPtLabel));

        }
      }
      break;

    case kWaitHelloAckStNetId:
    case kWaitEndpointAckStNetId:
      {
        cmTimeSpec_t t;
        cmTimeGet(&t);
        unsigned twentySecs = 20000000;
        if( cmTimeElapsedMicros(&np->lastSendTime,&t) > twentySecs)
        {
          const cmChar_t* ackStr = np->state==kWaitHelloAckStNetId ? "hello" : "endpoint";
          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);         
        }
      }
      break;

    default:
      break;
  }
      
  // if an error occurred put the node into an error state
  if( rc != kOkNetRC )
    np->state = kErrorStNetId;
      
  return rc;
}



cmRtNetRC_t cmRtNetSyncModeSend( cmRtNetH_t h )
{
  cmRtNetRC_t rc = kOkNetRC;
  cmRtNet_t*  p  = _cmRtNetHandleToPtr(h);

  if( p->syncModeFl == false )
    return rc;
  
  unsigned     activeCnt = 0;
  cmRtNetNode_t* np        = p->nodes;
  for(; np != NULL; np=np->link )
  {
    bool fl = (p->masterFl && cmIsFlag(np->flags,kRegNodeNetFl)) || (p->masterFl==false && cmIsFlag(np->flags,kRecvNodeNetFl));

    if( fl && np != p->localNode && np->state != kDoneStNetId && np->state != kErrorStNetId  )
    {
      _cmRtNetSendNodeSync(p,np);
      activeCnt += 1;
    }
  }
 
  if( activeCnt == 0 )
  {
    p->syncModeFl = false;
    _cmRtNetRpt(p,"sync mode complete.\n");
  }

  return rc;
}


cmRtNetRC_t cmRtNetReceive( cmRtNetH_t h )
{
  cmRtNetRC_t rc = kOkNetRC;
  cmRtNet_t*  p  = _cmRtNetHandleToPtr(h);

  if( cmUdpGetAvailData(p->udpH, NULL, NULL, NULL ) != kOkUdpRC )
  {
    cmErrMsg(&p->err,kUdpPortFailNetRC,"UDP port query failed.");
    goto errLabel;
  }
 errLabel:
  return rc;
}

bool    cmRtNetIsSyncModeMsg( const void* data, unsigned dataByteCnt )
{
  cmRtNetSyncMsg_t* m = (cmRtNetSyncMsg_t*)data;
  return dataByteCnt >= sizeof(cmRtNetSyncMsg_t) && m->hdr.selId == kNetSyncSelRtId;
}


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 )
{
  cmRtNetRC_t rc = kOkNetRC;
  //cmRtNet_t* p = _cmRtNetHandleToPtr(h);
  return rc;
}

void   cmRtNetReport( cmRtNetH_t h )
{
  cmRtNet_t* p = _cmRtNetHandleToPtr(h);
  cmRpt_t* rpt = p->err.rpt;

  cmRptPrintf(rpt,"Sync Mode:%s\n",p->syncModeFl ? "ON" : "OFF");

  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,kLocalNetFl) )
      cmRptPrintf(rpt,"LOCAL ");

    if( cmIsFlag(np->flags,kSockAddrNetFl) )
      cmRptPrintf(rpt,"%s ",cmStringNullGuard(cmUdpAddrToString(p->udpH,&np->sockaddr)));

    if( np->port != cmInvalidId )
      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->endPtId,cmStringNullGuard(ep->endPtLabel));
    }
  }
}


//==========================================================================
#include "cmThread.h"

typedef struct
{
  cmThreadH_t thH;
  cmRtNetH_t netH;
} _cmRtNetTest_t;

void _cmRtNetTestRecv( void* cbArg, const char* data, unsigned dataByteCnt, const struct sockaddr_in* fromAddr )
{
  _cmRtNetTest_t* p = (_cmRtNetTest_t*)cbArg;

  if( cmRtNetIsSyncModeMsg(data,dataByteCnt))
    cmRtNetSyncModeRecv(p->netH,data,dataByteCnt,fromAddr);

}


bool _cmRtNetTestThreadFunc(void* param)
{
  _cmRtNetTest_t* p = (_cmRtNetTest_t*)param;

  if( cmRtNetIsValid(p->netH) )
  {
    if( cmRtNetIsInSyncMode(p->netH) )
      cmRtNetSyncModeSend(p->netH);

    cmRtNetReceive(p->netH);
  }

  cmSleepMs(40);

  return true;
}

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;
  memset(&t,0,sizeof(t));


  if( cmThreadCreate(&p->thH,_cmRtNetTestThreadFunc,p,&ctx->rpt) != kOkThRC )
    goto errLabel;

  if((rc = cmRtNetAlloc(ctx,&p->netH,_cmRtNetTestRecv,p)) != kOkNetRC )
    goto errLabel;

  hostNameStr = cmRtNetLocalHostName(p->netH);
  if( hostNameStr == NULL )
    hostNameStr = "<no-host-name>";

  if((rc = cmRtNetCreateNode(p->netH, hostNameStr, NULL, port )) != kOkNetRC)
    goto errLabel;

  if( mstrFl )
  {
    if((rc = cmRtNetCreateNode(p->netH,"whirl", "192.168.15.109", port )) != kOkNetRC )
      goto errLabel;

    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( cmThreadPause(p->thH,0) != kOkThRC )
    goto errLabel;

  cmRptPrintf(&ctx->rpt,"%s q=quit\n", mstrFl ? "Master: " : "Slave: ");
  while( (c=getchar()) != 'q' )
  {
    switch(c)
    {
      case 'r':
        cmRtNetReport(p->netH);
        break;
    }
    
  }

 errLabel:

  cmThreadDestroy(&p->thH);

  cmRtNetFree(&p->netH);


  return;

}