开通博客赚积分
发布资源赚积分
分类

源码开发语言/平台
当前位置：首页 > 源码/资料 > 嵌入式/单片机编程 > VxWorks > 查看源码
distNodeLib.c
资源名称：vxwork_src.rar　[点击查看]
上传用户：nvosite88
上传日期：2007-01-17
资源大小：4983k
文件大小：87k
源码类别：

VxWorks

开发平台：

C/C++

distNodeLib.c：源码内容
  1. /* distNodeLib - node library (VxFusion option) */
  3. /* Copyright 1999 - 2002 Wind River Systems, Inc. */
  5. /*
  6. modification history
  7. --------------------
  8. 01q,15apr02,jws  fix SPR74878 (not locking node database properly)
  9. 01p,23oct01,jws  fix compiler warnings (SPR 71117)
  10. 01o,04oct01,jws  final fixes for SPR 34770
  11. 01n,27sep01,p_r  Fixes for SPR#34770
  12. 01m,11jun99,drm  Changing unicast piggy-backing default to FALSE.
  13. 01l,24may99,drm  added vxfusion prefix to VxFusion related includes
  14. 01k,11sep98,drm  added #include to pick up distPanic()
  15. 01j,12aug98,drm  added #include stmt for distObjTypeP.h
  16. 01i,08aug98,drm  added code to set broadcast flag when sending broadcast msgs
  17. 01h,08apr98,ur   made `piggy backing' switchable with distNodeCtl()
  18. 01g,13mar98,ur   added support for multiple crashed and operational hooks
  19. 01f,29jan98,ur   added support for sending negative acknowledgments
  20. 01e,23jan98,ur   bug fixed in the delivery of reassembled packets
  21. 01d,20jan98,ur   splited distNodeInit() in two parts
  22. 01c,19jan98,ur   bug fixed in distNodePktAck()
  23. 01b,16jan98,ur   XACK contains state of sending node
  24. 01a,10jun97,ur   written.
  25. */
  27. /*
  28. DESCRIPTION
  29. This library contains the node database, and routines to handle it.
  30. For every node in the system, one entry should be placed in the database.
  31. The database knows about the state of the node (as it looks from the local
  32. node) and about communication between the local node and the remote one.
  34. INTERNAL
  35. The semaphore <distNodeDbSem> is used to lock the node database.  The
  36. database is a hash table of nodes in the system.  Packets chained for
  37. input/output are part of each node's entry.  The table must be locked
  38. when entering an entry.  It is also locked when finding and entry, although
  39. it is not clear that this is necessary because entries are never removed.
  41. The DIST_NODE_COMM structures are the packet queues.  The database must
  42. be locked when dealing with these structures.  It is believed that this
  43. is now true.  Because of the way the communications routines are called,
  44. <distNodeDbSem> is now a mutex, and is sometimes called recursively.
  46. AVAILABILITY
  47. This module is distributed as a component of the unbundled distributed
  48. message queues option, VxFusion.
  49. */
  51. #include "vxWorks.h"
  53. #undef  DIST_DIAGNOSTIC   /* defining this seems to break VxFusion! */
  54. #undef  DIST_DIAGNOSTIC_SHOW
  56. #if defined (DIST_NODE_REPORT) 
  57.  || defined (DIST_DIAGNOSTIC) 
  58.  || defined (DIST_DIAGNOSTIC_SHOW)
  60. #include "stdio.h"
  62. #endif
  64. #include "stdlib.h"
  65. #include "hashLib.h"
  66. #include "semLib.h"
  67. #include "taskLib.h"
  68. #include "string.h"
  69. #include "errnoLib.h"
  70. #include "netinet/in.h"
  71. #include "private/distObjTypeP.h"
  72. #include "vxfusion/distLib.h"
  73. #include "vxfusion/distIfLib.h"
  74. #include "vxfusion/distNodeLib.h"
  75. #include "vxfusion/distStatLib.h"
  76. #include "vxfusion/private/distLibP.h"
  77. #include "vxfusion/private/distNodeLibP.h"
  78. #include "vxfusion/private/distTBufLibP.h"
  79. #include "vxfusion/private/distPktLibP.h"
  81. /* defines */
  83. /* make two macros look like macros */
  85. #define DIST_NODE_DB_LOCK    distNodeDbLock()
  86. #define DIST_NODE_DB_UNLOCK  distNodeDbUnlock()
  89. #define UNUSED_ARG(x)  if(sizeof(x)) {} /* to suppress compiler warnings */
  91. /* test if b is within [a,c) */
  92. #define winWithin(a, b, c) 
  93.     (((a <= b) && (b < c)) || ((c < a) && (a <= b)) || ((b < c) && (c < a)))
  95. /* add a to b and fold it back */
  96. #define winAdd(a, b) 
  97.     (((a) + (b)) & (DIST_IF_RNG_BUF_SZ - 1))
  99. /* sub a by b and fold it back */
  100. #define winSub(a, b) 
  101.     (((a) - (b)) & (DIST_IF_RNG_BUF_SZ - 1))
  103. #define KEY_ARG              65537        /* used by hash function */
  104. #define KEY_CMP_ARG          0            /* unused */
  106. #define DIST_NODE_MAX_HOOKS  8
  108. /* global variables */
  110. DIST_NODE_ID distNodeLocalId;   /* windSh needs this global */
  112. /*
  113.  * The semaphore used to be binary, but is now initialized as mutex.
  114.  * It cannot be changed back, because it is sometimes taken recursively,
  115.  * and will block if not mutex.
  116.  */
  117. SEM_ID       distNodeDbSem;     /* Should be taken before DIST_TBUF_FREEM */
  119. /* local variables */
  121. LOCAL HASH_ID     distNodeDbId;
  122. LOCAL BOOL        distNodeLibInstalled = FALSE;
  124. LOCAL int        distNodeNumNodesAll = 0;
  125. LOCAL int        distNodeNumNodesAlive = 0;
  126. LOCAL int        distNodeNumNodesOperational = 0;
  128. LOCAL int        distNodeLocalState;
  130. LOCAL DIST_NODE_ID    distNodeGodfatherId;
  132. LOCAL FUNCPTR        distNodeOperationalHook[DIST_NODE_MAX_HOOKS];
  133. LOCAL FUNCPTR        distNodeCrashedHook[DIST_NODE_MAX_HOOKS];
  135. LOCAL int        distNodeMaxRetries = DIST_NODE_MAX_RETRIES;
  136. LOCAL int        distNodeRetryTimeout = DIST_NODE_RETRY_TIMEOUT;
  138. LOCAL BOOL        distNodeSupportNACK = TRUE;   /* negative acknowlege */
  139. LOCAL BOOL        distNodeSupportPBB = FALSE;   /* piggy backing broadcast */
  140. LOCAL BOOL        distNodeSupportPBU = FALSE;   /* piggy backing unicast */
  142. /* local prototypes */
  144. LOCAL DIST_NODE_DB_NODE * distNodeFindById (DIST_NODE_ID nodeId);
  146. LOCAL BOOL   distNodeHCmp (DIST_NODE_DB_NODE *pMatchNode,
  147.                            DIST_NODE_DB_NODE *pHNode,
  148.                            int keyArg);
  149. LOCAL BOOL   distNodeHFunc (int elements, DIST_NODE_DB_NODE *pHNode,
  150.                             int keyArg);
  151. LOCAL STATUS distNodePktResend (DIST_NODE_COMM *pComm,
  152.                                 DIST_TBUF_HDR *pTBufHdr);
  153. LOCAL void   distNodeDBTimerTask (void);
  154. LOCAL void   distNodeDBTimer (void);
  155. LOCAL BOOL   distNodeDBNodeTimer (DIST_NODE_DB_NODE *pNode,
  156.                                   DIST_NODE_BTIMO *pBtimo);
  157. LOCAL void   distNodeDBCommTimer (DIST_NODE_DB_NODE *pNode,
  158.                                   DIST_NODE_BTIMO *pBtimo,
  159.                                   BOOL isBroadcastComm);
  160. LOCAL STATUS distNodeSendNegAck (DIST_NODE_DB_NODE *pNode,
  161.                                  short id,
  162.                                  short seq);
  163. LOCAL STATUS distNodeSendAck (DIST_NODE_DB_NODE *pNode,
  164.                               int ackBroadcast,
  165.                               int options);
  166. LOCAL void   distNodeCleanup (DIST_NODE_DB_NODE *pNode);
  167. LOCAL void   distNodeSetState (DIST_NODE_DB_NODE *pNode,
  168.                                int state);
  169. LOCAL STATUS distNodeSendBootstrap (DIST_NODE_ID dest,
  170.                                     int type,
  171.                                     int timeout);
  173. #if defined(DIST_DIAGNOSTIC) || defined(DIST_DIAGNOSTIC_SHOW)
  174. LOCAL BOOL         distNodeNodeShow (DIST_NODE_DB_NODE *pNode,
  175.                                      int dummy);
  176. LOCAL const char * distNodeStateToName (int state);
  177. #endif
  179. /***************************************************************************
  180. *
  181. * distNodeLibInit - initialize this module (VxFusion option)
  182. *
  183. * This routine currently does nothing.
  184. *
  185. * AVAILABILITY
  186. * This routine is distributed as a component of the unbundled distributed
  187. * message queues option, VxFusion.
  188. *
  189. * RETURNS: N/A
  190. * NOMANUAL
  191. */
  193. void distNodeLibInit (void)
  194.     {
  195.     }
  197. /***************************************************************************
  198. *
  199. * distNodeInit - initializes the node library (VxFusion option)
  200. *
  201. * This routine initializes the node database. The database can handle
  202. * up to 2^<sizeLog2> entries.
  203. *
  204. * AVAILABILITY
  205. * This routine is distributed as a component of the unbundled distributed
  206. * message queues option, VxFusion.
  207. *
  208. * RETURNS: OK, if successful.
  209. * NOMANUAL
  210. */
  212. STATUS distNodeInit
  213.     (
  214.     int sizeLog2   /* init database with 2^^sizeLog2 entries */
  215.     )
  216.     {
  217.     int    hashTblSizeLog2;
  218.     int    i, tid;
  220.     if (sizeLog2 < 1)
  221.         return (ERROR);
  223.     if (distNodeLibInstalled == TRUE)
  224.         return (OK);
  226.     for (i = 0; i < DIST_NODE_MAX_HOOKS; i++)
  227.         {
  228.         distNodeOperationalHook[i] = NULL;
  229.         distNodeCrashedHook[i] = NULL;
  230.         }
  232.     distNodeLocalState = DIST_NODE_STATE_BOOT;
  234.     /* Initialize the node database. */
  236.     if (hashLibInit () == ERROR)
  237.         return (ERROR);    /* hashLibInit() failed */
  239.     hashTblSizeLog2 = sizeLog2 - 1;
  240.     distNodeDbId = hashTblCreate (hashTblSizeLog2, distNodeHCmp,
  241.             distNodeHFunc, KEY_ARG);
  242.     if (distNodeDbId == NULL)
  243.         return (ERROR);    /* hashTblCreate() failed */
  245. #ifdef UNDEFINED
  246.     distNodeDbSem = semBCreate (SEM_Q_PRIORITY, SEM_FULL) ; 
  247. #else
  248.     distNodeDbSem = semMCreate (SEM_Q_PRIORITY + SEM_INVERSION_SAFE) ; 
  249. #endif
  251.     /* Get the node database manager running. */
  253.     tid = taskSpawn ("tDiNodeMgr",
  254.                      DIST_NODE_MGR_PRIO,
  255.                      VX_SUPERVISOR_MODE | VX_UNBREAKABLE,
  256.                      DIST_NODE_MGR_STACK_SZ,
  257.                      (FUNCPTR) distNodeDBTimerTask,
  258.                      0, 0, 0, 0, 0, 0, 0, 0, 0, 0);
  259.     if (tid == ERROR)
  260.         return (ERROR);    /* taskSpawn() failed */
  262.     distNodeGodfatherId = DIST_IF_BROADCAST_ADDR;
  264.     /*
  265.      * Add a virtual broadcast node.
  266.      * State information (sliding windows, etc.) for outgoing broadcasts
  267.      * must be handled at a centralized place. This is the virtual
  268.      * broadcast node.
  269.      * State information for incoming broadcasts is managed within
  270.      * the broadcast communication substructure for the corresponding
  271.      * node. When a note enters the incorporation phase, all other
  272.      * nodes in the system get aware of it and update their node
  273.      * databases.
  274.      */
  276.     if (distNodeCreate (DIST_IF_BROADCAST_ADDR, DIST_NODE_STATE_OPERATIONAL)
  277.          ==
  278.         NULL)
  279.         {
  280.         return (ERROR);
  281.         }
  282.         
  283.     distNodeLibInstalled = TRUE;
  285.     return (OK);
  286.     }
  288. /***************************************************************************
  289. *
  290. * distNodeHCmp - compare function for hashing in node DB (VxFusion option)
  291. *
  292. * This routine is the hash compare function for node IDs.
  293. *
  294. * AVAILABILITY
  295. * This routine is distributed as a component of the unbundled distributed
  296. * message queues option, VxFusion.
  297. *
  298. * RETURNS: True, if node IDs match.
  299. * NOMANUAL
  300. */
  302. LOCAL BOOL distNodeHCmp
  303.     (
  304.     DIST_NODE_DB_NODE * pMatchNode, /* first node */
  305.     DIST_NODE_DB_NODE * pHNode,     /* second node */
  306.     int                 keyArg      /* unused arg */
  307.     )
  308.     {
  309.     DIST_NODE_ID  distNodeId1 = pMatchNode->nodeId;
  310.     DIST_NODE_ID  distNodeId2 = pHNode->nodeId;
  311.     
  312.     UNUSED_ARG(keyArg);
  314.     return (distNodeId1 == distNodeId2);
  315.     }
  317. /***************************************************************************
  318. *
  319. * distNodeHFunc - hash function for node DB (VxFusion option)
  320. *
  321. * This is the hash function for node IDs.
  322. *
  323. * AVAILABILITY
  324. * This routine is distributed as a component of the unbundled distributed
  325. * message queues option, VxFusion.
  326. *
  327. * RETURNS: A hash index.
  328. * NOMANUAL
  329. */
  331. LOCAL BOOL distNodeHFunc
  332.     (
  333.     int                 elements,  /* size of hash table */
  334.     DIST_NODE_DB_NODE * pHNode,    /* node whose ID to hash */
  335.     int                 divisor    /* used by hash computation */
  336.     )
  337.     {
  338.     return ((pHNode->nodeId % divisor) & (elements - 1));
  339.     }
  341. /***************************************************************************
  342. *
  343. * distNodeCtl - control function for node database (VxFusion option)
  344. *
  345. * This routine performs a control function on the node database.
  346. * The following functions are accepted:
  347. * is
  348. * i DIST_CTL_RETRY_TIMEOUT
  349. * Set send-timeout in ticks to start with. When no ACK is received
  350. * within the timeout, the packet is resent. The timeout for the
  351. * <n>th sending is: <n> * DIST_CTL_RETRY_TIMEOUT .
  352. * i DIST_CTL_MAX_RETRIES
  353. * Set a limit for number of retries, when sending fails.
  354. * i DIST_CTL_GET_LOCAL_ID
  355. * Get local node id.
  356. * i DIST_CTL_GET_LOCAL_STATE
  357. * Get state of local node.
  358. * i DIST_CTL_NACK_SUPPORT
  359. * Negative acknowledges (NACKs) are used for requesting a resend of a single
  360. * missing fragment from a packet. NACKs are sent immediately after a
  361. * fragment is found to be missing.
  362. * If <arg> is FALSE (0), the sending of negative acknowledges is disabled.
  363. * If <arg> is TRUE (1), sending of NACKs is enabled. This is the default.
  364. * i DIST_CTL_PGGYBAK_UNICST_SUPPORT
  365. * If this is enabled, the system waits a version dependent time until it
  366. * sends an acknowledge for a previously received packet.
  367. * If a data packet is sent to
  368. * the acknowledge awaiting host in meantime, the acknowlege is delivered
  369. * in that packet. This switch turns on/off piggy backing for unicast
  370. * communication only.
  371. * If <arg> is FALSE (0), piggy backing is disabled.
  372. * If <arg> is TRUE (1), piggy backing is enabled.
  373. * Piggy backing is enabled for unicast communication by default.
  374. * i DIST_CTL_PGGYBAK_BRDCST_SUPPORT
  375. * If this is enabled, the system waits
  376. * a version dependent time until it sends an
  377. * acknowledge for a previously received packet. If a data packet is sent to
  378. * the acknowledge awaiting host in meantime, the acknowlege is delivered
  379. * in that packet. This switch turns on/off piggy backing for broadcast
  380. * communication only.
  381. * If <arg> is FALSE (0), piggy backing is disabled.
  382. * If <arg> is TRUE (1), piggy backing is enabled.
  383. * Piggy backing is disabled for broadcast communication by default.
  384. * i DIST_CTL_OPERATIONAL_HOOK
  385. * Set a function to be called, each time a node shifts to operational state.
  386. * i DIST_CTL_CRASHED_HOOK
  387. * Set a function to be called, each time a node shifts to crashed state.
  388. * ie
  389. *
  390. * The prototype of the function hooked to the database, should look
  391. * like this:
  392. * cs
  393. *    void fnc (DIST_NODE_ID nodeStateChanged);
  394. * ce
  395. * 
  396. * RETURNS: OK, the value requested, or ERROR if <function> is unknown.
  397. *
  398. * ERRNO:
  399. * S_distLib_UNKNOWN_REQUEST
  400. *
  401. * AVAILABILITY
  402. * This routine is distributed as a component of the unbundled distributed
  403. * message queues option, VxFusion.
  404. * NOMANUAL
  405. */
  407. int distNodeCtl
  408.     (
  409.     int function,        /* function code      */
  410.     int arg              /* arbitrary argument */
  411.     )
  412.     {
  413.     int i;
  415.     switch (function)
  416.         {
  417.         case DIST_CTL_MAX_RETRIES:
  418.             if (arg < 0)
  419.                 return (ERROR);
  420.             distNodeMaxRetries = arg;
  421.             return (OK);
  423.         case DIST_CTL_RETRY_TIMEOUT:
  424.             if (arg < 0)
  425.                 return (ERROR);
  426.             /*
  427.              * <arg> is in ticks;
  428.              * <distNodeRetryTimeout> is in manager wakeups;
  429.              */
  431.             distNodeRetryTimeout = (1000 * arg / sysClkRateGet() +
  432.                     DIST_NODE_MGR_WAKEUP_MSEC - 1) / DIST_NODE_MGR_WAKEUP_MSEC;
  434.             return (OK);
  436.         case DIST_CTL_OPERATIONAL_HOOK:
  437.             {
  438.             for (i = 0; i < DIST_NODE_MAX_HOOKS; i++)
  439.                 {
  440.                 if (distNodeOperationalHook[i] == NULL)
  441.                     {
  442.                     distNodeOperationalHook[i] = (FUNCPTR) arg;
  443.                     return (OK);
  444.                     }
  445.                 }
  446.             return (ERROR);
  447.             }
  449.         case DIST_CTL_CRASHED_HOOK:
  450.             {
  451.             for (i = 0; i < DIST_NODE_MAX_HOOKS; i++)
  452.                 {
  453.                 if (distNodeCrashedHook[i] == NULL)
  454.                     {
  455.                     distNodeCrashedHook[i] = (FUNCPTR) arg;
  456.                     return (OK);
  457.                     }
  458.                 }
  459.             return (ERROR);
  460.             }
  462.         case DIST_CTL_GET_LOCAL_ID:
  463.             return (distNodeLocalId);
  465.         case DIST_CTL_GET_LOCAL_STATE:
  466.             return (distNodeLocalState);
  468.         case DIST_CTL_NACK_SUPPORT:
  469.             if ((BOOL) arg != TRUE && (BOOL) arg != FALSE)
  470.                 return (ERROR);
  471.             distNodeSupportNACK = (BOOL) arg;
  472.             return (OK);
  474.         case DIST_CTL_PGGYBAK_BRDCST_SUPPORT:
  475.             if ((BOOL) arg != TRUE && (BOOL) arg != FALSE)
  476.                 return (ERROR);
  477.             distNodeSupportPBB = (BOOL) arg;
  478.             return (OK);
  480.         case DIST_CTL_PGGYBAK_UNICST_SUPPORT:
  481.             if ((BOOL) arg != TRUE && (BOOL) arg != FALSE)
  482.                 return (ERROR);
  483.             distNodeSupportPBU = (BOOL) arg;
  484.             return (OK);
  486.         default:
  487.             errnoSet (S_distLib_UNKNOWN_REQUEST);
  488.             return (ERROR);
  489.         }
  490.     }
  492. /***************************************************************************
  493. *
  494. * distNodeSendBootstrap - send a bootstrap packet (VxFusion option)
  495. *
  496. * This routine sends a bootstrap packet.
  497. *
  498. * AVAILABILITY
  499. * This routine is distributed as a component of the unbundled distributed
  500. * message queues option, VxFusion.
  501. *
  502. * RETURNS: OK, if packet sent.
  503. * NOMANUAL
  504. */
  506. LOCAL STATUS distNodeSendBootstrap
  507.     (
  508.     DIST_NODE_ID dest,     /* destination node */
  509.     int          type,     /* bootstrap type */
  510.     int          timeout   /* xmit timeout */
  511.     )
  512.     {
  513.     DIST_TBUF_HDR * pTBufHdr;
  514.     DIST_PKT_BOOT * pPkt;
  515.     DIST_TBUF *     pTBuf;
  516.     
  517.     if ((pTBufHdr = DIST_TBUF_HDR_ALLOC ()) == NULL)
  518.         return (ERROR);             /* out of TBufs */
  520.     pTBufHdr->tBufHdrPrio = DIST_BOOTSTRAP_PRIO;
  521.     pTBufHdr->tBufHdrDest = dest;
  522.     pTBufHdr->tBufHdrOverall = sizeof (*pPkt);
  523.     pTBufHdr->tBufHdrTimo = timeout;
  524.     
  525.     
  526.     if ((pTBuf = DIST_TBUF_ALLOC ()) == NULL)
  527.         {
  528.         DIST_TBUF_FREEM (pTBufHdr);
  529.         return (ERROR);                /* out of TBufs */
  530.         }
  531.         
  532.     pTBuf->tBufFlags = DIST_TBUF_FLAG_HDR | DIST_TBUF_FLAG_BROADCAST;
  533.     pTBuf->tBufType = DIST_TBUF_TTYPE_BOOTSTRAP;
  534.     pTBuf->tBufSeq = 0;
  535.     pTBuf->tBufNBytes = sizeof (*pPkt);
  537.     pPkt = (DIST_PKT_BOOT *) pTBuf->pTBufData;
  538.     
  540.     DIST_TBUF_ENQUEUE (pTBufHdr, pTBuf);
  542.     pPkt->pktBootType = (UINT8) type;
  543.     return (distNodePktSend (pTBufHdr));
  544.     }
  546. /***************************************************************************
  547. *
  548. * distNodeBootstrap - broadcast a bootstrap packet (VxFusion option)
  549. *
  550. * This routine broadcasts a bootstrap packet.
  551. *
  552. * AVAILABILITY
  553. * This routine is distributed as a component of the unbundled distributed
  554. * message queues option, VxFusion.
  555. *
  556. * RETURNS: OK, if broadcast succeeds.
  557. * NOMANUAL
  558. */
  560. STATUS distNodeBootstrap
  561.     (
  562.     int    timeout   /* xmit timeout */
  563.     )
  564.     {
  565.     STATUS status;
  566.     status = distNodeSendBootstrap (DIST_IF_BROADCAST_ADDR,
  567.                                     DIST_BOOTING_REQ, timeout);
  568.     return (status);
  569.     }
  571. /***************************************************************************
  572. *
  573. * distNodeFindById - find node in node database (VxFusion option)
  574. *
  575. * This routine tries to find the information node with node id <nodeId>
  576. * in the node database.
  577. *
  578. * This routine takes <distNodeDbLock>.  Usually, it will be called with
  579. * the lock already taken, but not always, so the lock is taken here
  580. * to be safe.
  581. *
  582. * RETURNS: Pointer to the node entry, or NULL if not found.
  583. *
  584. * AVAILABILITY
  585. * This routine is distributed as a component of the unbundled distributed
  586. * message queues option, VxFusion.
  587. *
  588. * NOMANUAL
  589. */
  591. LOCAL DIST_NODE_DB_NODE * distNodeFindById
  592.     (
  593.     DIST_NODE_ID    nodeId   /* ID of node to find */
  594.     )
  595.     {
  596.     DIST_NODE_DB_NODE   distNodeMatch;
  597.     DIST_NODE_DB_NODE * pDistNodeFound;
  599.     distNodeMatch.nodeId = nodeId;
  600.     
  601.     DIST_NODE_DB_LOCK;
  603.     pDistNodeFound =
  604.        (DIST_NODE_DB_NODE *) hashTblFind (distNodeDbId,
  605.                                           (HASH_NODE *) &distNodeMatch,
  606.                                           KEY_CMP_ARG);
  607.     DIST_NODE_DB_UNLOCK;
  609.     return (pDistNodeFound);
  610.     }
  612. /***************************************************************************
  613. *
  614. * distNodeCreate - create a new node (VxFusion option)
  615. *
  616. * This routine creates a new node with id <distNodeId>.
  617. * distNodeCreate() returns a pointer to the DB entry.
  618. *
  619. * This function takes the DB node database lock before inserting the
  620. * node into the database.
  621. *
  622. * AVAILABILITY
  623. * This routine is distributed as a component of the unbundled distributed
  624. * message queues option, VxFusion.
  625. *
  626. * RETURNS: Pointer to node, or NULL.
  627. * NOMANUAL
  628. */
  630. DIST_NODE_DB_NODE * distNodeCreate
  631.     (
  632.     DIST_NODE_ID    nodeId,   /* ID for new node */
  633.     int             state     /* state of new node */
  634.     )
  635.     {
  636.     DIST_NODE_DB_NODE * pNodeNew;
  637.     DIST_NODE_BFLD *    pBFld;
  638.     int                 sz;
  640.     /* This node is new to the database. */
  642. #ifdef DIST_NODE_REPORT
  643.     distLog("distNodeCreate: Creating node 0x%lx with state %dn",
  644.             nodeId, state);
  645. #endif
  647.     /* allocate needed space; quit if we can't get it */
  648.     
  649.     pNodeNew = (DIST_NODE_DB_NODE *) malloc (sizeof (DIST_NODE_DB_NODE));
  650.     if (pNodeNew == NULL)
  651.         {
  652.         distStat.memShortage++;
  653.         return (NULL);
  654.         }
  656.     sz = DIST_NODE_BFLD_SIZEOF (DIST_IF_RNG_BUF_SZ);
  657.     pBFld = (DIST_NODE_BFLD *) malloc (4 * sz);
  658.     if (pBFld == NULL)
  659.         {
  660.         free(pNodeNew);          /* no memory leaks here */
  661.         distStat.memShortage++;
  662.         return (NULL);
  663.         }
  665.     /* now fill in DIST_NODE_DB_NODE info */
  667.     bzero ((char *) pNodeNew, sizeof (DIST_NODE_DB_NODE));
  668.     pNodeNew->nodeId = nodeId;
  669.     pNodeNew->nodeBroadcast.commAckLastRecvd = (INT16) winSub (0, 1);
  670.     pNodeNew->nodeState = state;
  672.     distNodeNumNodesAll++;
  674.     if (DIST_NODE_STATE_IS_ALIVE (state))
  675.         distNodeNumNodesAlive++;
  677.     if (state == DIST_NODE_STATE_OPERATIONAL)
  678.         distNodeNumNodesOperational++;
  680.     pNodeNew->nodeState = state;
  683.     bzero ((char *) pBFld, 2 * sz);
  684.     pNodeNew->nodeUnicast.pCommCompleted = pBFld;
  685.     pNodeNew->nodeBroadcast.pCommCompleted = ((char *) pBFld) + sz;
  686.     
  687.     /* now link the node into the data base */
  688.     
  689.     DIST_NODE_DB_LOCK;
  691.     hashTblPut (distNodeDbId, (HASH_NODE *)pNodeNew);
  693.     DIST_NODE_DB_UNLOCK;
  695.     return (pNodeNew);
  696.     }
  698. /***************************************************************************
  699. *
  700. * distNodeSetState - set state of a node (VxFusion option)
  701. *
  702. * This routine is used to change the state of a node.  It is only called
  703. * from two places in distNodeReassemble() which takes the node DB lock.
  704. *
  705. * AVAILABILITY
  706. * This routine is distributed as a component of the unbundled distributed
  707. * message queues option, VxFusion.
  708. *
  709. * RETURNS: N/A
  710. * NOMANUAL
  711. */
  713. LOCAL void distNodeSetState
  714.     (
  715.     DIST_NODE_DB_NODE * pNode,     /* node */
  716.     int                 newState   /* new state */
  717.     )
  718.     {
  719.     int             curState = pNode->nodeState;
  720.     DIST_NODE_ID    nodeId   = pNode->nodeId;
  721.     int             i;
  723. #ifdef DIST_DIAGNOSTIC
  724.     if (curState != newState)
  725.         distLog ("distNodeSetState: state of node 0x%lx shifts from %s to %sn",
  726.                  nodeId,
  727.                  distNodeStateToName (curState),
  728.                  distNodeStateToName (newState));
  729. #endif
  731.     if ((! DIST_NODE_STATE_IS_ALIVE (curState)) &&
  732.          DIST_NODE_STATE_IS_ALIVE (newState))
  733.         {
  734.         distNodeNumNodesAlive++;
  735.         }
  736.         
  737.     if (DIST_NODE_STATE_IS_ALIVE (curState) &&
  738.         (! DIST_NODE_STATE_IS_ALIVE (newState)))
  739.         {
  740.         distNodeNumNodesAlive--;
  741.         }
  743.     if (curState < DIST_NODE_STATE_OPERATIONAL &&
  744.             newState == DIST_NODE_STATE_OPERATIONAL)
  745.         {
  747.         distNodeNumNodesOperational++;
  749.         for (i = 0; i < DIST_NODE_MAX_HOOKS; i++)
  750.             {
  751.             if (distNodeOperationalHook[i] != NULL)
  752.                 (* (distNodeOperationalHook[i])) (nodeId);
  753.             }
  755.         pNode->nodeState = newState;
  756.         return;
  757.         }
  759.     if (curState == DIST_NODE_STATE_OPERATIONAL &&
  760.         newState < DIST_NODE_STATE_OPERATIONAL)
  761.         {
  762.         distNodeNumNodesOperational--;
  764.         if (newState == DIST_NODE_STATE_CRASHED)
  765.             {
  766.             for (i = 0; i < DIST_NODE_MAX_HOOKS; i++)
  767.                 {
  768.                 if (distNodeCrashedHook[i] != NULL)
  769.                     (* (distNodeCrashedHook[i])) (nodeId);
  770.                 }
  771.             }
  773.         pNode->nodeState = newState;
  774.         return;
  775.         }
  777.     pNode->nodeState = newState;
  778.     }
  780. /***************************************************************************
  781. *
  782. * distNodeOperational - sets a node to be alive (VxFusion option)
  783. *
  784. * This routine changes the state of the node with the id <distNodeId>
  785. * to OPERATIONAL.
  786. *
  787. * This routine takes <distNodeDbLock>.
  788. *
  789. * AVAILABILITY
  790. * This routine is distributed as a component of the unbundled distributed
  791. * message queues option, VxFusion.
  792. *
  793. * RETURNS: Pointer to node, or NULL.
  794. * NOMANUAL
  795. */
  797. DIST_NODE_DB_NODE * distNodeOperational
  798.     (
  799.     DIST_NODE_ID   distNodeId  /* node ID */
  800.     )
  801.     {
  802.     DIST_NODE_DB_NODE * pNodeFound;
  803.     int                 i;
  805.     DIST_NODE_DB_LOCK;
  806.      
  807.     if ((pNodeFound = distNodeFindById (distNodeId)) != NULL)
  808.         {
  810.         /*
  811.          * This node is already in the node database. If state is already
  812.          * OPERATIONAL, do nothing - we even have not recognized the absence.
  813.          */
  815.         if (! DIST_NODE_IS_ALIVE (pNodeFound))
  816.             distNodeNumNodesAlive++;
  818.         if (pNodeFound->nodeState < DIST_NODE_STATE_OPERATIONAL)
  819.             {
  821. #ifdef DIST_DIAGNOSTIC
  822.         distLog ("distNodeOperational: Set node 0x%lx to state OPERATIONALn",
  823.                 distNodeId);
  824. #endif
  825.             distNodeNumNodesOperational++;
  827.             for (i = 0; i < DIST_NODE_MAX_HOOKS; i++)
  828.                 {
  829.                 if (distNodeOperationalHook[i] != NULL)
  830.                     (* (distNodeOperationalHook[i])) (distNodeId);
  831.                 }
  832.             }
  834.         pNodeFound->nodeState = DIST_NODE_STATE_OPERATIONAL;
  835.         }
  836.     
  837.     DIST_NODE_DB_UNLOCK;
  838.     
  839.     return (pNodeFound);
  840.     }
  842. /***************************************************************************
  843. *
  844. * distNodeCrashed - sets a node to be dead (VxFusion option)
  845. *
  846. * This routine changes the state of the node with the id <distNodeId>
  847. * to CRASHED.
  848. *
  849. * This routine takes <distNodeDbLock>.
  850. *
  851. * AVAILABILITY
  852. * This routine is distributed as a component of the unbundled distributed
  853. * message queues option, VxFusion.
  854. *
  855. * RETURNS: Pointer to node, or NULL.
  856. * NOMANUAL
  857. */
  859. DIST_NODE_DB_NODE * distNodeCrashed
  860.     (
  861.     DIST_NODE_ID    distNodeId  /* node ID */
  862.     )
  863.     {
  864.     DIST_NODE_DB_NODE * pNodeFound;
  865.     int                 i;
  867.     /* you cannot kill the broadcast node */
  868.     
  869.     if (distNodeId == DIST_IF_BROADCAST_ADDR)
  870.         return (NULL);
  871.     
  872.     DIST_NODE_DB_LOCK;
  874.     /* find the node */
  876.     if ((pNodeFound = distNodeFindById (distNodeId)) != NULL)
  877.         {
  879.         if (DIST_NODE_IS_ALIVE (pNodeFound))
  880.             distNodeNumNodesAlive--;
  882.         if (pNodeFound->nodeState >= DIST_NODE_STATE_OPERATIONAL)
  883.             {
  885. #ifdef DIST_DIAGNOSTIC
  886.         distLog ("distNodeCrashed: Set node 0x%lx to state CRASHEDn",
  887.                 distNodeId);
  888. #endif
  889.             distNodeNumNodesOperational--;
  891.             for (i = 0; i < DIST_NODE_MAX_HOOKS; i++)
  892.                 {
  893.                 if (distNodeCrashedHook[i] != NULL)
  894.                     (* (distNodeCrashedHook[i])) (distNodeId);
  895.                 }
  896.             }
  898.         pNodeFound->nodeState = DIST_NODE_STATE_CRASHED;
  899.         }
  900.     
  901.     DIST_NODE_DB_UNLOCK;
  902.     
  903.     return (pNodeFound);
  904.     }
  906. /***************************************************************************
  907. *
  908. * distNodeLocalGetId - returns the local node id (VxFusion option)
  909. *
  910. * This routine returns the id of the local node.
  911. *
  912. * AVAILABILITY
  913. * This routine is distributed as a component of the unbundled distributed
  914. * message queues option, VxFusion.
  915. *
  916. * RETURNS: Local node ID.
  917. * NOMANUAL
  918. */
  920. DIST_NODE_ID distNodeLocalGetId (void)
  921.     {
  922.     return (distNodeLocalId);
  923.     }
  925. /***************************************************************************
  926. *
  927. * distNodeLocalSetId - sets the local node id (VxFusion option)
  928. *
  929. * This routine sets the id of the local node.
  930. *
  931. * AVAILABILITY
  932. * This routine is distributed as a component of the unbundled distributed
  933. * message queues option, VxFusion.
  934. *
  935. * RETURNS: N/A
  936. * NOMANUAL
  937. */
  939. void distNodeLocalSetId
  940.     (
  941.     DIST_NODE_ID    myNodeId    /* node ID */
  942.     )
  943.     {
  944.     distNodeLocalId = myNodeId;
  945.     }
  947. /***************************************************************************
  948. *
  949. * distNodeLocalSetState - sets the state of the local node (VxFusion option)
  950. *
  951. * This routine sets the state of the local node.
  952. *
  953. * AVAILABILITY
  954. * This routine is distributed as a component of the unbundled distributed
  955. * message queues option, VxFusion.
  956. *
  957. * RETURNS: N/A
  958. * NOMANUAL
  959. */
  961. void distNodeLocalSetState
  962.     (
  963.     int    state    /* new state for local node */
  964.     )
  965.     {
  966.     distNodeLocalState = state;
  967.     }
  969. /***************************************************************************
  970. *
  971. * distNodeLocalGetState - get state of the local node (VxFusion option)
  972. *
  973. * This routine returns the state of the local node.
  974. *
  975. * AVAILABILITY
  976. * This routine is distributed as a component of the unbundled distributed
  977. * message queues option, VxFusion.
  978. *
  979. * RETURNS: Node state.
  980. * NOMANUAL
  981. */
  983. int distNodeLocalGetState (void)
  984.     {
  985.     return (distNodeLocalState);
  986.     }
  988. /***************************************************************************
  989. *
  990. * distNodeGetNumNodes - returns the number of nodes in a specified state (VxFusion option)
  991. *
  992. * This routine returns the number of nodes currently in a specified state.
  993. *
  994. * AVAILABILITY
  995. * This routine is distributed as a component of the unbundled distributed
  996. * message queues option, VxFusion.
  997. *
  998. * RETURNS: Count of nodes, or -1 if <typeSet> is invalid.
  999. * NOMANUAL
  1000. */
  1002. int distNodeGetNumNodes
  1003.     (
  1004.     int    typeSet   /* state to look for */
  1005.     )
  1006.     {
  1007.     switch (typeSet)
  1008.         {
  1009.         case DIST_NODE_NUM_NODES_ALL:
  1010.             return (distNodeNumNodesAll);
  1012.         case DIST_NODE_NUM_NODES_ALIVE:
  1013.             return (distNodeNumNodesAlive);
  1015.         case DIST_NODE_NUM_NODES_OPERATIONAL:
  1016.             return (distNodeNumNodesOperational);
  1018.         default:
  1019.             return (-1);
  1020.         }
  1021.     }
  1023. /***************************************************************************
  1024. *
  1025. * distNodeGetGodfatherId - get the id of our godfather (VxFusion option)
  1026. *
  1027. * This routine returns the id of this node's godfather.
  1028. * If no godfather is available, distNodeGetGodfatherId() returns ERROR,
  1029. * else OK.
  1030. *
  1031. * AVAILABILITY
  1032. * This routine is distributed as a component of the unbundled distributed
  1033. * message queues option, VxFusion.
  1034. *
  1035. * RETURNS: OK, if godfather exists.
  1036. * NOMANUAL
  1037. */
  1039. STATUS distNodeGetGodfatherId
  1040.     (
  1041.     DIST_NODE_ID * nodeId  /* where to return godfather ID */
  1042.     )
  1043.     {
  1044.     if ((*nodeId = distNodeGodfatherId) == DIST_IF_BROADCAST_ADDR)
  1045.         return (ERROR);
  1047.     return (OK);
  1048.     }
  1050. #ifdef UNDEFINED
  1052. /* This routine does not seem to be used */
  1054. /***************************************************************************
  1055. *
  1056. * distNodeTouch - touch a node (increment message counter) (VxFusion option)
  1057. *
  1058. * This routine touches a node.
  1059. *
  1060. * NOTE: Must take <distNodeDbLock> before.
  1061. *
  1062. * AVAILABILITY
  1063. * This routine is distributed as a component of the unbundled distributed
  1064. * message queues option, VxFusion.
  1065. *
  1066. * RETURNS: OK, unless <distNodeId> is invalid.
  1067. * NOMANUAL
  1068. */
  1070. STATUS distNodeTouch
  1071.     (
  1072.     DIST_NODE_ID    distNodeId   /* ID of node to touch */
  1073.     )
  1074.     {
  1075.     DIST_NODE_DB_NODE * pDistNodeFound;
  1077.     if ((pDistNodeFound = distNodeFindById (distNodeId)) == NULL)
  1078.         {
  1079.         /*  Node does not exist. Should never happen. */
  1081.         distStat.nodeDBNoMatch++;
  1082.         return (ERROR);
  1083.         }
  1085.     DIST_NODE_IN_PKT_INC(pDistNodeFound);
  1087.     return (OK);
  1088.     }
  1089. #endif
  1091. /***************************************************************************
  1092. *
  1093. * distNodeEach - call a routine for each node in the database (VxFusion option)
  1094. *
  1095. * This routine calls a user-supplied routine once for each node.
  1096. * The user-supplied routine should return TRUE if distNodeEach() is to
  1097. * continue calling it with the remaining nodes, or FALSE if it is done and
  1098. * distNodeEach() can exit.
  1099. *
  1100. * distNodeEach() returns NULL if traversed whole database, or pointer to
  1101. * node entry that distNodeEach() ended with.
  1102. *
  1103. * NOTE: Takes <distNodeDbLock>.
  1104. *
  1105. * AVAILABILITY
  1106. * This routine is distributed as a component of the unbundled distributed
  1107. * message queues option, VxFusion.
  1108. *
  1109. * RETURNS: Pointer to last node visited.
  1110. * NOMANUAL
  1111. */
  1113. DIST_NODE_DB_NODE * distNodeEach
  1114.     (
  1115.     FUNCPTR    routine,      /* function to call at each node */
  1116.     int        routineArg    /* argument to function */
  1117.     )
  1118.     {
  1119.     DIST_NODE_DB_NODE * lastNode;
  1121.     distNodeDbLock ();
  1122.     lastNode = (DIST_NODE_DB_NODE *) hashTblEach (distNodeDbId,
  1123.                                                   routine,
  1124.                                                   routineArg);
  1126.     distNodeDbUnlock ();
  1128.     return (lastNode);
  1129.     }
  1131. /***************************************************************************
  1132. *
  1133. * distNodeCleanup - Flush out all communication queues (VxFusion option)
  1134. *
  1135. * This routine flushes all communication queues of <pNode>.  It is only
  1136. * called from two places in distNodeReassemble() which takes the node
  1137. * DB lock first.
  1138. *
  1139. * AVAILABILITY
  1140. * This routine is distributed as a component of the unbundled distributed
  1141. * message queues option, VxFusion.
  1142. *
  1143. * RETURNS: N/A
  1144. * NOMANUAL
  1145. */
  1147. LOCAL void distNodeCleanup
  1148.     (
  1149.     DIST_NODE_DB_NODE * pNode    /* node to clean-up */
  1150.     )
  1151.     {
  1152.     DIST_TBUF_HDR *  pPkt;
  1153.     DIST_NODE_COMM * pCommUnicast, *pCommBroadcast;
  1155.     pCommUnicast = &pNode->nodeUnicast;
  1156.     pCommBroadcast = &pNode->nodeBroadcast;
  1158.     /*
  1159.      * Clean up communication structures of bootstrapping
  1160.      * node.
  1161.      *
  1162.      * Delete all unicast packets waiting for reassemly.
  1163.      */
  1165.     while (pCommUnicast->pCommQReass)
  1166.         {
  1167.         DIST_TBUF_HDR_DEQUEUE (pCommUnicast->pCommQReass, pPkt);
  1168.         DIST_TBUF_FREEM (pPkt);
  1169.         }
  1171.     /*
  1172.      * Dequeue all unicast packets waiting for a slide in
  1173.      * the window. Wakeup sender.
  1174.      */
  1176.     while (pCommUnicast->pCommQWinOut)
  1177.         {
  1178.         DIST_TBUF_HDR_DEQUEUE (pCommUnicast->pCommQWinOut, pPkt);
  1179.         pPkt->pTBufHdrTm->tmStatus = DIST_TM_STATUS_UNREACH;
  1180.         semGive (&pPkt->pTBufHdrTm->tmWait4);
  1181.         /* distNodePktSend() frees packet */
  1182.         }
  1184.     /*
  1185.      * Dequeue all unicast packets waiting for an ACK.
  1186.      * Wakeup sender.
  1187.      */
  1189.     while (pCommUnicast->pCommQAck)
  1190.         {
  1191.         DIST_TBUF_HDR_DEQUEUE (pCommUnicast->pCommQAck, pPkt);
  1192.         pPkt->pTBufHdrTm->tmStatus = DIST_TM_STATUS_UNREACH;
  1193.         semGive (&pPkt->pTBufHdrTm->tmWait4);
  1194.         /* distNodePktSend() frees packet */
  1195.         }
  1196.     
  1197.     bzero ((char *) pCommUnicast->pCommCompleted,
  1198.             DIST_NODE_BFLD_SIZEOF (DIST_IF_RNG_BUF_SZ));
  1200.     pCommUnicast->commPktNextExpect = 0;
  1201.     pCommUnicast->commAckNextExpect = 0;
  1202.     pCommUnicast->commPktNextSend = 0;
  1203.     pCommUnicast->commAckDelayed = FALSE;
  1205.     /* Delete all broadcast packets waiting for reassembly. */
  1207.     while (pCommBroadcast->pCommQReass)
  1208.         {
  1209.         DIST_TBUF_HDR_DEQUEUE (pCommBroadcast->pCommQReass, pPkt);
  1210.         DIST_TBUF_FREEM (pPkt);
  1211.         }
  1212.     
  1213.     bzero ((char *) pCommBroadcast->pCommCompleted,
  1214.             DIST_NODE_BFLD_SIZEOF (DIST_IF_RNG_BUF_SZ));
  1216.     }
  1218. /***************************************************************************
  1219. *
  1220. * distNodeGetReassembled - get a reassembled packet (VxFusion option)
  1221. *
  1222. * This routine gets a reassembled data packet.  It takes the node DB lock.
  1223. *
  1224. * AVAILABILITY
  1225. * This routine is distributed as a component of the unbundled distributed
  1226. * message queues option, VxFusion.
  1227. *
  1228. * RETURNS: Ptr to tbuf header.
  1229. * NOMANUAL
  1230. */
  1232. DIST_TBUF_HDR * distNodeGetReassembled
  1233.     (
  1234.     DIST_NODE_COMM * pComm  /* ptr to comm node */
  1235.     )
  1236.     {
  1237.     DIST_TBUF_HDR *  pHead;         /* head of tbufs */
  1238.     short            nextExpected;  /* next expected seq number */
  1239.     DIST_TBUF_HDR *  pNext;         /* next tbuf */
  1241.     distNodeDbLock ();    
  1243.     pHead = pComm->pCommQNextDeliver;
  1244.     nextExpected = pComm->commPktNextExpect;
  1246.     if (pHead && pHead->tBufHdrId == nextExpected &&
  1247.             DIST_NODE_BTST (pComm->pCommCompleted, nextExpected))
  1248.         {
  1250.         /* Remove the packet from the reassembly queue. */
  1252.         DIST_TBUF_HDR_UNLINK (pComm->pCommQReass, pHead);
  1254.         if (nextExpected == (DIST_IF_RNG_BUF_SZ - 1))
  1255.             pNext = pComm->pCommQReass;
  1256.         else
  1257.             pNext = pHead->pTBufHdrNext;
  1259.         if (pNext && winAdd (nextExpected, 1) != pNext->tBufHdrId)
  1260.             {
  1261.             /* if a complete packet is missing, we fall in here */
  1263.             pComm->pCommQNextDeliver = NULL;
  1264.             }
  1265.         else
  1266.             {
  1267.             /* next packet in ring is the one, we expect next */
  1268.             
  1269.             pComm->pCommQNextDeliver = pNext;
  1270.             }
  1272.         distNodeDbUnlock ();            /* UNLOCK */
  1274. #ifdef DIST_NODE_REPORT
  1275.         /* this is a hack */
  1277.         printf ("distNodeGetReassembled: return %p (type %d/%d)n", pHead,
  1278.                 *((char *) (DIST_TBUF_GET_NEXT (pHead)->pTBufData)),
  1279.                 *((char *) (DIST_TBUF_GET_NEXT (pHead)->pTBufData) + 1));
  1280. #endif
  1281.         return (pHead);
  1282.         }
  1284.     distNodeDbUnlock ();                /* UNLOCK */
  1286. #ifdef DIST_NODE_REPORT
  1287.     printf ("distNodeGetReassembled: no packet availablen");
  1288. #endif
  1289.     return (NULL);
  1290.     }
  1292. /***************************************************************************
  1293. *
  1294. * distNodeReassemble - reassemble a packet (VxFusion option)
  1295. *
  1296. * This routine takes a telegram and tries to reassemble it with other
  1297. * fragments already stored in the reassembly list.
  1298. * distNodeReassemble() updates the communication windows managed
  1299. * within the database.
  1300. *
  1301. * If <nodeIdSrc> is unknown by now, distNodeReassemble() creates a
  1302. * new node in the node database. Creating a node is only allowed when
  1303. * receiving a XACK telegram in bootstrap mode or a BOOTSTRAP telegram.
  1304. *
  1305. * If a packet has been reassembled successfully, distNodeReassemble()
  1306. * returns a pointer to the communication structure. Else a NULL pointer
  1307. * is returned.
  1308. *
  1309. * NOTE: Takes <distNodeDbLock>.
  1310. *
  1311. * AVAILABILITY
  1312. * This routine is distributed as a component of the unbundled distributed
  1313. * message queues option, VxFusion.
  1314. *
  1315. * RETURNS: Ptr to comm node.
  1316. * NOMANUAL
  1317. */
  1319. DIST_NODE_COMM * distNodeReassemble
  1320.     (
  1321.     DIST_NODE_ID nodeIdSrc,    /* source node */
  1322.     int          prio,         /* priority */
  1323.     DIST_TBUF *  pTBufNew      /* ptr to tbuf */
  1324.     )
  1325.     {
  1326.     DIST_NODE_DB_NODE *   pNode;
  1327.     DIST_NODE_DB_NODE *   pNodeSrc;
  1328.     DIST_NODE_COMM *      pCommIn;
  1329.     DIST_NODE_COMM *      pCommOut;
  1330.     DIST_TBUF_HDR *       pReassHdr;
  1331.     DIST_TBUF_HDR *       pReassHdrPrev;
  1332.     DIST_TBUF_HDR *       pTBufAck;
  1333.     DIST_TBUF_HDR *       pTBufAckNext;
  1334.     DIST_TBUF_HDR *       pTBufHdrNew;
  1335.     short                 winLo, winHi, pktNext;
  1336.     short                 id;
  1337.     short                 ack;
  1338.     short                 seq;
  1339.     short                 nBytes;
  1340.     short                 type;
  1341.     int                   hasMf;
  1343.     /* Get information from telegram */
  1344.     
  1345.     id = pTBufNew->tBufId;
  1346.     ack = pTBufNew->tBufAck;
  1347.     seq = pTBufNew->tBufSeq;
  1348.     type = pTBufNew->tBufType;
  1349.     nBytes = pTBufNew->tBufNBytes;
  1350.     hasMf = DIST_TBUF_HAS_MF (pTBufNew);
  1352.     distNodeDbLock ();
  1354.     /* Find node in hash table. */
  1355.     
  1356.     pNodeSrc = distNodeFindById (nodeIdSrc);
  1357.     
  1358.     if (pNodeSrc == NULL)
  1359.         {
  1360.         if (nodeIdSrc == distNodeLocalId)
  1361.             distPanic ("distNodePktReass: someone has same node id we haven");
  1363.         /*
  1364.          * Node is unknown by now. Create new node, if this is a
  1365.          * XACK in bootstrapping mode or this is a BOOTSTRAP telegram.
  1366.          * Else ignore telegram.
  1367.          */
  1369.         if (distNodeLocalState == DIST_NODE_STATE_BOOT
  1370.                 && type == DIST_TBUF_TTYPE_BACK
  1371.                 && nBytes == sizeof (DIST_XACK))
  1372.             {
  1373.             DIST_XACK    *pXAck = (DIST_XACK *) pTBufNew->pTBufData;
  1374.             int            state;
  1376.             state = ntohs (pXAck->xAckPktNodeState);
  1378.             if ((pNodeSrc = distNodeCreate (nodeIdSrc, state)) == NULL)
  1379.                 {
  1380.                 /*
  1381.                  * We had problems creating the new node.
  1382.                  * This is a fatal error, since we will miss that one.
  1383.                  */
  1385.                 distStat.nodeDBFatal++;
  1386.                 DIST_TBUF_FREE (pTBufNew);
  1387.                 distPanic ("distNodePktReass: error accepting node 0x%lxn",
  1388.                         nodeIdSrc);
  1389.                 }
  1390.             }
  1391.         else if (type == DIST_TBUF_TTYPE_BOOTSTRAP)
  1392.             {
  1393.             int state = DIST_NODE_STATE_BOOT;
  1395. #ifdef DIST_NODE_REPORT
  1396.             printf ("distNodeReassemble: got bootstrap pkt from node 0x%lxn",
  1397.                     nodeIdSrc);
  1398. #endif
  1399.             if ((pNodeSrc = distNodeCreate (nodeIdSrc, state)) == NULL)
  1400.                 {
  1401.                 /*
  1402.                  * We had problems creating the new node.
  1403.                  * This is a fatal error, since we will miss that one.
  1404.                  */
  1406.                 distStat.nodeDBFatal++;
  1407.                 DIST_TBUF_FREE (pTBufNew);
  1408.                 distPanic ("distNodePktReass: error accepting node 0x%lxn",
  1409.                         nodeIdSrc);
  1410.                 }
  1412.             /*
  1413.              * If this is a BOOTSTRAP telegram, we have to clean up
  1414.              * the communication structures.
  1415.              */
  1417.             distNodeCleanup (pNodeSrc);
  1418.             pNodeSrc->nodeBroadcast.commPktNextExpect = (INT16)winAdd (id, 1);
  1419.             }
  1420.         else
  1421.             {
  1422.             /*
  1423.              * Packets from unknown nodes are ignored, unless they
  1424.              * are of type XACK or BOOTSTRAP.
  1425.              */
  1427. #ifdef DIST_DIAGNOSTIC
  1428.             distLog ("distNodePktReass: packet from unknown node 0x%lx 
  1429.                      --ignoredn", nodeIdSrc);
  1430. #endif
  1431.             DIST_TBUF_FREE (pTBufNew);
  1432.             distNodeDbUnlock();
  1433.             return (NULL);
  1434.             }
  1436. #ifdef DIST_DIAGNOSTIC
  1437.         distLog ("distNodePktReass: new node 0x%lx created (state %s)n",
  1438.                  nodeIdSrc,
  1439.                  distNodeStateToName (distNodeGetState (pNodeSrc)));
  1440. #endif
  1441.         }
  1443.     if (DIST_TBUF_IS_BROADCAST (pTBufNew))
  1444.         {
  1445.         if ((pNode = distNodeFindById (DIST_IF_BROADCAST_ADDR)) == NULL)
  1446.             {
  1447.             distStat.nodeDBNoMatch++;
  1448.             DIST_TBUF_FREE (pTBufNew);
  1449.             distNodeDbUnlock();
  1450.             return (NULL);
  1451.             }
  1452.         pCommIn = &pNodeSrc->nodeBroadcast;
  1453.         pCommOut = &pNode->nodeBroadcast;
  1454.         }
  1455.     else
  1456.         pCommIn = pCommOut = &pNodeSrc->nodeUnicast;
  1458. #ifdef DIST_NODE_REPORT
  1459.     {
  1460.     char * typeName;
  1461.     switch (type)
  1462.         {
  1463.         case DIST_TBUF_TTYPE_DTA:
  1464.             typeName = "DATA";
  1465.             break;
  1466.         case DIST_TBUF_TTYPE_ACK:
  1467.             typeName = "ACK";
  1468.             break;
  1469.         case DIST_TBUF_TTYPE_BDTA:
  1470.             typeName = "B-DATA";
  1471.             break;
  1472.         case DIST_TBUF_TTYPE_BACK:
  1473.             typeName = "ACK-B";
  1474.             break;
  1475.         case DIST_TBUF_TTYPE_BOOTSTRAP:
  1476.             typeName = "BOOTSTRAP";
  1477.             break;
  1478.         case DIST_TBUF_TTYPE_NACK:
  1479.             typeName = "NACK";
  1480.             break;
  1481.         default:
  1482.             typeName = "unknown";
  1483.         }
  1484.     printf ("dist..Reass:0x%lx/%s: ", nodeIdSrc, typeName);
  1485.     printf ("id %d, ack %d, seq %d, type %d, len %d, hasMf %dn",
  1486.             id, ack, seq, pTBufNew->tBufType, nBytes, hasMf);
  1487.     }
  1488. #endif
  1490.     if (type == DIST_TBUF_TTYPE_BOOTSTRAP)
  1491.         {
  1492.         int    bootTp;
  1494.         if (distNodeGetState (pNodeSrc) != DIST_NODE_STATE_BOOT)
  1495.             {
  1496.             /*
  1497.              * If we receive a BOOTSTARP telegram from a node, that is
  1498.              * not in booting mode, it must be rebooting. Cleanup
  1499.              * the structure.
  1500.              */
  1502.             distNodeCleanup (pNodeSrc);
  1503.             }
  1505.         pNodeSrc->nodeBroadcast.commPktNextExpect = (INT16) winAdd (id, 1);
  1507.         bootTp = ((DIST_PKT_BOOT *) pTBufNew->pTBufData)->pktBootType;
  1508.         switch (bootTp)
  1509.             {
  1510.             case DIST_BOOTING_REQ:
  1511.                 /* Send XACK. */
  1512.                 
  1513.                 distNodeSetState (pNodeSrc, DIST_NODE_STATE_BOOT);
  1514.                 distNodeSendAck (pNodeSrc, 1, DIST_NODE_ACK_EXTENDED);
  1515.                 break;
  1517. #ifdef DIST_DIAGNOSTIC
  1518.             default:
  1519.                 distLog ("dist..Reass: unknown type of BOOTSTRAP telegramn");
  1520. #endif
  1521.             }
  1523.         distNodeDbUnlock ();
  1524.         DIST_TBUF_FREE (pTBufNew);
  1525.         return (NULL);
  1526.         }
  1528.     /* We will fall in here, when we receive a XACK. */
  1529.     
  1530.     if (type == DIST_TBUF_TTYPE_BACK && nBytes == sizeof (DIST_XACK))
  1531.         {
  1532.         DIST_XACK    *pXAck = (DIST_XACK *) pTBufNew->pTBufData;
  1534.         distNodeSetState (pNodeSrc, ntohs (pXAck->xAckPktNodeState));
  1536.         /* In bootstrapping mode. */
  1538.         if (distNodeLocalState == DIST_NODE_STATE_BOOT)
  1539.             {
  1540.             short        pktNextSend;
  1542. #ifdef DIST_NODE_REPORT
  1543.             printf ("dist..Reass: XACK in bootstrap moden");
  1544. #endif
  1545.             /*
  1546.              * The first node that answers to our BOOTSTRAP packet,
  1547.              * will be our godfather.
  1548.              */
  1550.             if (distNodeGodfatherId == DIST_IF_BROADCAST_ADDR)
  1551.                 distNodeGodfatherId = nodeIdSrc;
  1553.             pktNextSend = ntohs (pXAck->xAckPktNextSend);
  1555.             pCommIn->commPktNextExpect = pktNextSend;
  1556.             }
  1558.         distNodeDbUnlock ();
  1559.         DIST_TBUF_FREE (pTBufNew);
  1560.         return (NULL);
  1561.         }
  1563. #ifdef DIST_NODE_REPORT
  1564.     if (! DIST_NODE_IS_ALIVE (pNodeSrc))
  1565.         {
  1566.         printf ("dist..Reass: got a telegram from crashed noden");
  1567.         }
  1568. #endif
  1570.     /*
  1571.      * Process ACK field. An ACK commits all packets with
  1572.      * an id lower or equal to the value of the ACK.
  1573.      *
  1574.      * Traverse the list of non-acknowledged packets and
  1575.      * decrease the ACK counters of all acknowledged packets
  1576.      * in the list. If no ACK is missing for a certain packet,
  1577.      * dequeue it and wakeup the sender.
  1578.      */
  1580.     pktNext = pCommOut->commPktNextSend;
  1581.     if (winWithin (pCommOut->commAckNextExpect, ack, pktNext))
  1582.         {
  1583.         short    ackLo = pCommOut->commAckNextExpect;
  1584.         short    ackHi = (INT16) winAdd (ack, 1);
  1586.         pTBufAck = pCommOut->pCommQAck;
  1587.         pCommIn->commAckLastRecvd = ack;
  1589. #ifdef DIST_NODE_REPORT
  1590.         printf ("dist..Reass: ACK %d, AckQ %p:n", ack, pTBufAck);
  1591. #endif
  1592.         while (pTBufAck)
  1593.             {
  1594.             pTBufAckNext = DIST_TBUF_HDR_GET_NEXT (pTBufAck);
  1596.             if (winWithin (ackLo, pTBufAck->tBufHdrId, ackHi))
  1597.                 {
  1598. #ifdef DIST_NODE_REPORT
  1599.                 printf ("dist..Reass: within window, wait for %d more ACKsn",
  1600.                         pTBufAck->pTBufHdrTm->tmAckCnt - 1);
  1601. #endif
  1602.                 if (--pTBufAck->pTBufHdrTm->tmAckCnt == 0)
  1603.                     {
  1604.                     pCommOut->commAckNextExpect =
  1605.                         (INT16) winAdd (pCommOut->commAckNextExpect, 1);
  1607.                     DIST_TBUF_HDR_UNLINK (pCommOut->pCommQAck, pTBufAck);
  1608.                     pTBufAck->pTBufHdrTm->tmStatus = DIST_TM_STATUS_OK;
  1610. #ifdef DIST_NODE_REPORT
  1611.                     printf ("dist..Reass: %d ACKedn", pTBufAck->tBufHdrId);
  1612. #endif
  1613.                     semGive (&pTBufAck->pTBufHdrTm->tmWait4);
  1614.                     
  1615.                     /*
  1616.                      * distNodePktSend()--which should awake now--frees
  1617.                      * the packet.
  1618.                      */
  1619.                     }
  1620.                 }
  1622.             pTBufAck = pTBufAckNext;
  1623.             }
  1624.         }
  1626.     /*
  1627.      * If this is a NACK telegram, resend the single fragment, the
  1628.      * NACK asks us for. <id> holds the id of the packet, <seq> the
  1629.      * sequence number of the fragment within the packet.
  1630.      * The NACK telegram can be destroyed at this point, since
  1631.      * it contains no data.
  1632.      */
  1634.     if (type == DIST_TBUF_TTYPE_NACK)
  1635.         {
  1636.         DIST_TBUF_FREE (pTBufNew);
  1638. #ifdef DIST_NODE_REPORT
  1639.         printf ("dist..Reass: got NACK for (id %d, seq %d) from node 0x%lxn",
  1640.                 id, seq, nodeIdSrc);
  1641. #endif
  1642.         pTBufAck = pCommOut->pCommQAck;
  1643.         while (pTBufAck)
  1644.             {
  1645.             if (pTBufAck->tBufHdrId == id)
  1646.                 {
  1647.                 DIST_TBUF    *pFrag;
  1649.                 for (pFrag = DIST_TBUF_GET_NEXT (pTBufAck);
  1650.                         pFrag && pFrag->tBufSeq != seq;
  1651.                         pFrag = DIST_TBUF_GET_NEXT (pFrag));
  1653.                 pFrag->tBufAck =
  1654.                       (UINT16)winSub (pCommIn->commPktNextExpect, 1);
  1656. #ifdef DIST_NODE_REPORT
  1657.                 if (DIST_IF_SEND (nodeIdSrc, pFrag, 0) == ERROR)
  1658.                     printf ("dist..Reass: error sending NACK to node 0x%lxn",
  1659.                             nodeIdSrc);
  1660. #else
  1661.                 /* If this fails, there is no need to mess around. */
  1663.                 DIST_IF_SEND (nodeIdSrc, pFrag, 0);
  1664. #endif
  1665.                 distStat.nodeFragResend++;
  1667.                 distNodeDbUnlock ();
  1668.                 return (NULL);
  1669.                 }
  1670.             
  1671.             pTBufAck = DIST_TBUF_HDR_GET_NEXT (pTBufAck);
  1672.             }
  1674.         distNodeDbUnlock ();
  1675.         return (NULL);
  1676.         }
  1678.     /* If this is not a DTA (data) telegram, destroy it. */
  1679.     
  1680.     if (type != DIST_TBUF_TTYPE_DTA && type != DIST_TBUF_TTYPE_BDTA)
  1681.         {
  1682.         distNodeDbUnlock ();
  1683.         DIST_TBUF_FREE (pTBufNew);
  1684.         return (NULL);
  1685.         }
  1687.     /*
  1688.      * Go on processing this telegram?
  1689.      * 1) The packet id must be within the window.
  1690.      *    The window starts with the id of the packet
  1691.      *    expected next and has a size of DIST_NODE_WIN_SZ.
  1692.      * 2) The packet should not be reassembled already.
  1693.      */
  1695.     winLo = pCommIn->commPktNextExpect;
  1696.     winHi = (INT16) winAdd (winLo, DIST_NODE_WIN_SZ);
  1697.     if (! winWithin (winLo, id, winHi) ||
  1698.             DIST_NODE_BTST (pCommIn->pCommCompleted, id))
  1699.         {
  1700.         /* Not within the window. Destroy the telegram. */
  1702.         distNodeDbUnlock ();
  1704. #ifdef DIST_NODE_REPORT
  1705.         if (DIST_NODE_BTST (pCommIn->pCommCompleted, id))
  1706.             printf ("distNodePktReass: packet already reassembledn");
  1707.         else
  1708.             printf ("distNodePktReass: not within window (%d..%d)n",
  1709.                     winLo, winSub (winHi, 1));
  1710. #endif
  1711.         DIST_TBUF_FREE (pTBufNew);
  1712.         return (NULL);
  1713.         }
  1715.     /* Find queue for packet in reassembly list. */
  1716.      
  1717.     pReassHdr = pCommIn->pCommQReass;
  1718.     pReassHdrPrev = NULL;
  1719.     while (pReassHdr)
  1720.         {
  1721.         if (pReassHdr->tBufHdrId > id)
  1722.             break;    /* insert a new packet */
  1724.         if (pReassHdr->tBufHdrId == id)
  1725.             {
  1726.             /* We have found the packet, the new telegram belongs to. */
  1728.             DIST_TBUF    *pTBuf, *pTBufPrev;
  1729.             DIST_TBUF    *pTBufLast = DIST_TBUF_GET_LAST (pReassHdr);
  1730.             short        seqLast = pTBufLast->tBufSeq;
  1732. #ifdef DIST_NODE_REPORT
  1733.             printf ("dist..Reass: packet found, inserting telegramn");
  1734. #endif
  1735.             if (seqLast < seq)
  1736.                 {
  1737.                 /*
  1738.                  * This telegram has the highest sequence number ever
  1739.                  * received. Put it to the tail of the TBuf list.
  1740.                  */
  1742.                 DIST_TBUF_ENQUEUE (pReassHdr, pTBufNew);
  1743.                 pReassHdr->tBufHdrNLeaks += seq - seqLast - 1;
  1745. #ifdef UNDEFINED  /* unused, but be careful before excising for good */
  1746.                 pTBufLast = pTBufNew;
  1747. #endif
  1748.                 if (! DIST_TBUF_IS_BROADCAST (pTBufNew)
  1749.                     && (seq - seqLast) > 1)
  1750.                     {
  1751.                     INT16 i;
  1753.                     /*
  1754.                      * unlock database--not strictly necessary, but
  1755.                      * we are consuming time, holding the lock...
  1756.                      */
  1758.                     for (i = seqLast + 1; i < seq; i++)
  1759.                         distNodeSendNegAck (pNodeSrc, id, i);
  1760.                     }
  1761.                 }
  1762.             else
  1763.                 {
  1764.                 /*
  1765.                  * This telegram must be inserted somewhere in the TBuf list.
  1766.                  */
  1768.                 pTBufPrev = (DIST_TBUF *) pReassHdr;
  1769.                 pTBuf = DIST_TBUF_GET_NEXT (pReassHdr);
  1770.                 while (pTBuf != NULL)
  1771.                     {
  1772.                     if (pTBuf->tBufSeq == seq)
  1773.                         {
  1774.                         /* We have already received this one. */
  1775.                         
  1776.                         distNodeDbUnlock ();
  1777.                         DIST_TBUF_FREE (pTBufNew);
  1778.                         return (NULL);
  1779.                         }
  1780.     
  1781.                     if (pTBuf->tBufSeq > seq)
  1782.                         {
  1783.                         DIST_TBUF_INSERT_AFTER (pReassHdr,
  1784.                                                 pTBufPrev,
  1785.                                                 pTBufNew);
  1786.                         pReassHdr->tBufHdrNLeaks--;
  1787.                         break;
  1788.                         }
  1790.                     pTBufPrev = pTBuf;
  1791.                     pTBuf = DIST_TBUF_GET_NEXT (pTBuf);
  1792.                     }
  1793.                 }
  1795.             /* The telegram is in the reassembly list now. */
  1797.             pReassHdr->tBufHdrOverall += nBytes;
  1799.             /*
  1800.              * Test if the packet is fully reassembled (there
  1801.              * are no leaks within the packet and the last
  1802.              * fragment has the 'more fragments' bit cleared).
  1803.              */
  1805.             pTBufLast = DIST_TBUF_GET_LAST (pReassHdr);
  1806.             if (!pReassHdr->tBufHdrNLeaks && !(DIST_TBUF_HAS_MF (pTBufLast)))
  1807.                 {
  1808.                 /*
  1809.                  * We have received a complete packet and
  1810.                  * do not expect to receive more fragments.
  1811.                  */
  1813. #ifdef DIST_NODE_REPORT
  1814.                 {
  1815.                 DIST_TBUF    *pFirst = DIST_TBUF_GET_NEXT (pReassHdr);
  1816.                 
  1817.                 /* this is a hack */
  1818.                 
  1819.                 printf ("dist..Reass: packet %d complete (type %d/%d)n", id,
  1820.                         *((char *) (pFirst->pTBufData)),
  1821.                         *((char *) (pFirst->pTBufData) + 1));
  1822.                 }
  1823. #endif
  1824.                 DIST_NODE_BSET (pCommIn->pCommCompleted, id);
  1826.                 /*
  1827.                  * Test if the reassembled packet is the one
  1828.                  * we expect next.
  1829.                  */
  1831.                 if (pReassHdr->tBufHdrId == pCommIn->commPktNextExpect)
  1832.                     {
  1833.                     /*
  1834.                      * This is the packet, we expect next.
  1835.                      * Return communication pointer to caller,
  1836.                      * so that it knows about the job.
  1837.                      */
  1839.                     distNodeDbUnlock ();
  1840.                     return (pCommIn);
  1841.                     }
  1842.                 }
  1843.             distNodeDbUnlock ();
  1844.             return (NULL);
  1845.             }
  1847.         pReassHdrPrev = pReassHdr;
  1848.         pReassHdr = DIST_TBUF_HDR_GET_NEXT (pReassHdr);
  1849.         }
  1850.     
  1851.     /*
  1852.      * This is the first fragment of a new packet. There are either
  1853.      * more fragments, or this packet is not the one, we are waiting
  1854.      * for.
  1855.      */
  1856.      
  1857. #ifdef DIST_NODE_REPORT
  1858.     printf ("dist..Reass: first fragment of new packetn");
  1859. #endif
  1861.     if ((pTBufHdrNew = DIST_TBUF_HDR_ALLOC ()) == NULL)
  1862.         {
  1863.         distStat.netInDiscarded++; /* nodeInDiscarded counts pkts discarded */
  1864.         distNodeDbUnlock ();
  1865.         DIST_TBUF_FREE (pTBufNew);
  1866.         return (NULL);
  1867.         }
  1869.     pTBufHdrNew->tBufHdrSrc = nodeIdSrc;
  1870.     pTBufHdrNew->tBufHdrId = id;
  1871.     pTBufHdrNew->tBufHdrPrio = prio;
  1872.     pTBufHdrNew->tBufHdrOverall = nBytes;
  1873.     pTBufHdrNew->tBufHdrNLeaks = seq;
  1874.     pTBufHdrNew->tBufHdrTimo = WAIT_FOREVER;    /* reassembly timeout */
  1876.     DIST_TBUF_SET_NEXT (pTBufHdrNew, pTBufNew);
  1877.     DIST_TBUF_SET_LAST (pTBufHdrNew, pTBufNew);
  1879.     DIST_TBUF_HDR_INSERT_AFTER (pCommIn->pCommQReass, pReassHdrPrev,
  1880.         pTBufHdrNew);
  1882.     if (pCommIn->commPktNextExpect == id)
  1883.         /* this is the packet we want to deliver next */
  1884.         pCommIn->pCommQNextDeliver = pTBufHdrNew;
  1886.     if (seq == 0 && ! hasMf)
  1887.         {
  1888.         /* We have received a complete packet in a single telegram. */
  1890. #ifdef DIST_NODE_REPORT
  1891.         /* this is a hack */
  1892.         
  1893.         printf ("dist..Reass: packet %d complete (type %d/%d)n", id,
  1894.                 *((char *) (pTBufNew->pTBufData)),
  1895.                 *((char *) (pTBufNew->pTBufData) + 1));
  1896. #endif
  1897.         DIST_NODE_BSET (pCommIn->pCommCompleted, id);
  1898.         if (id == pCommIn->commPktNextExpect)
  1899.             {
  1900.             /*
  1901.              * This is the packet, we expect next.
  1902.              * Return communication pointer to caller,
  1903.              * so that it knows about the job.
  1904.              */
  1906.             distNodeDbUnlock ();
  1907.             return (pCommIn);
  1908.             }
  1909.         }
  1911.     distNodeDbUnlock ();
  1913.     /* Check if we have to send negative acknowledges (NACKs). */
  1914.     
  1915.     if (! DIST_TBUF_IS_BROADCAST (pTBufNew))
  1916.         {
  1917.         DIST_TBUF_HDR    *pPrev;
  1918.         DIST_TBUF        *pLastOfPrev;
  1920.         /*
  1921.          * Check if we have received the last fragment of the previous
  1922.          * packet. We cannot find out how many fragments are missing at
  1923.          * the end of the packet. So just send a NACK for the successor
  1924.          * of the last telegram received.
  1925.          */
  1927.         if ((pPrev = DIST_TBUF_HDR_GET_PREV (pTBufHdrNew)) != NULL &&
  1928.                 (pLastOfPrev = DIST_TBUF_GET_LAST (pTBufHdrNew)) != NULL &&
  1929.                 DIST_TBUF_HAS_MF (pLastOfPrev))
  1930.             {
  1931.             distNodeSendNegAck (pNodeSrc, pPrev->tBufHdrId,
  1932.                     pLastOfPrev->tBufSeq + 1);
  1933.             }
  1935.         /*
  1936.          * Check if some telegrams from the beginning of the new packet
  1937.          * are missing. Send a NACK for each of them.
  1938.          */
  1940.         if (seq > 0)
  1941.             {
  1942.             INT16 i;
  1943.             for (i = 0; i < seq; i++)
  1944.                 distNodeSendNegAck (pNodeSrc, id, i);
  1945.             }
  1947.         }
  1949.     return (NULL);
  1950.     }
  1952. /***************************************************************************
  1953. *
  1954. * distNodePktSend - send a packet already stored in TBufs (VxFusion option)
  1955. *
  1956. * This routine waits for an open send window and transmits the packet,
  1957. * by sending the data of each single TBuf.
  1958. * Afterwards it blocks for an ACK from the remote node.
  1959. *
  1960. * NOTE: Takes <distNodeDbLock>.
  1961. *
  1962. * AVAILABILITY
  1963. * This routine is distributed as a component of the unbundled distributed
  1964. * message queues option, VxFusion.
  1965. *
  1966. * RETURNS: OK, if successful.
  1967. * NOMANUAL
  1968. */
  1970. STATUS distNodePktSend
  1971.     (
  1972.     DIST_TBUF_HDR * pTBufHdr    /* message to send */
  1973.     )
  1974.     {
  1975.     DIST_NODE_DB_NODE *  pDistNodeFound;
  1976.     DIST_TRANSMISSION    distTm;
  1977.     DIST_NODE_COMM *     pComm;
  1978.     DIST_NODE_ID         nodeIdDest = pTBufHdr->tBufHdrDest;
  1979.     DIST_TBUF *          pTBuf;
  1980.     short                winLo, winHi;
  1981.     short                pktId;
  1982.     int                  priority;
  1983.     int                  wait4NumAcks;
  1985.     distNodeDbLock ();
  1987.     distStat.nodeOutReceived++;
  1989.     if ((pDistNodeFound = distNodeFindById (nodeIdDest)) == NULL)
  1990.         {
  1991. #ifdef DIST_DIAGNOSTIC
  1992.         distLog ("distNodePktSend: unknown node 0x%lxn", nodeIdDest);
  1993. #endif
  1994.         distStat.nodeDBNoMatch++;
  1995.         distNodeDbUnlock ();
  1996.         DIST_TBUF_FREEM (pTBufHdr);
  1997.         return (ERROR);
  1998.         }
  2000.     if (! DIST_NODE_IS_ALIVE (pDistNodeFound))
  2001.         {
  2002. #ifdef DIST_DIAGNOSTIC
  2003.         distLog ("distNodePktSend: destination not aliven");
  2004. #endif
  2005.         distStat.nodeNotAlive++;
  2006.         distNodeDbUnlock ();
  2007.         DIST_TBUF_FREEM (pTBufHdr);
  2008.         return (ERROR);
  2009.         }
  2011. #ifdef DIST_DIAGNOSTIC
  2012.     if (DIST_TBUF_GET_NEXT (pTBufHdr) == NULL)
  2013.         {
  2014.         distLog ("distNodePktSend: TBuf chain contains no datan");
  2015.         }
  2016. #endif
  2018.     if (nodeIdDest == DIST_IF_BROADCAST_ADDR)
  2019.         {
  2020.         pComm = &pDistNodeFound->nodeBroadcast;
  2021.         wait4NumAcks = distNodeNumNodesAlive - 1;
  2022.         }
  2023.     else
  2024.         {
  2025.         pComm = &pDistNodeFound->nodeUnicast;
  2026.         wait4NumAcks = 1;
  2027.         }
  2029.     semBInit (&distTm.tmWait4, SEM_Q_FIFO, SEM_EMPTY);
  2030.     distTm.tmStatus = DIST_TM_STATUS_OK;
  2031.     distTm.tmRetmTimo = distNodeRetryTimeout;
  2032.     distTm.tmNumTm = 1;
  2033.     distTm.tmAckCnt = wait4NumAcks;
  2035.     pTBufHdr->pTBufHdrTm = &distTm;
  2037.     /*
  2038.      * Each node has two queues for outgoing packets.
  2039.      * One is the queue for packets that should been
  2040.      * send, but the "window is closed" (full).
  2041.      * The other is the queue for packets send, and
  2042.      * waiting for an ACK.
  2043.      * In both of the two queues, packets get older.
  2044.      */
  2046.     pTBufHdr->tBufHdrId = pktId = pComm->commPktNextSend;
  2047.     winLo = pComm->commAckNextExpect;
  2048.     winHi = (INT16) winAdd (pComm->commAckNextExpect, DIST_NODE_WIN_SZ);
  2050.     /*
  2051.      * Can we send the packet directly or do we have to wait
  2052.      * until the "window opens".
  2053.      */
  2055.     if (! winWithin (winLo, pktId, winHi))
  2056.         {
  2057.         /* Put packet to the WaitToSend queue. */
  2058.         
  2059.         DIST_TBUF_HDR_ENQUEUE (pComm->pCommQWinOut, pTBufHdr);
  2061.         distNodeDbUnlock ();    /* unlock before sleep */
  2063.         /* Wait for a place within the window. */
  2064.         
  2065.         semTake (&distTm.tmWait4, WAIT_FOREVER);
  2067.         if (distTm.tmStatus != DIST_TM_STATUS_OK)
  2068.             {
  2069.             distNodeDbUnlock ();       /* this is not needed here! */
  2070.             DIST_TBUF_FREEM (pTBufHdr);
  2071.             return (ERROR);            /* e.g. timeout, node removed */
  2072.             }
  2074.         distNodeDbLock ();
  2075.         
  2076.         /* Remove it from the WaitToSend queue. */
  2077.         
  2078.         DIST_TBUF_HDR_UNLINK (pComm->pCommQWinOut, pTBufHdr);
  2079.         }
  2081.     pComm->commPktNextSend = (INT16) winAdd (pComm->commPktNextSend, 1);
  2083.     /* Put packet to the WaitForAck queue and send it. */
  2084.     
  2085.     DIST_TBUF_HDR_ENQUEUE (pComm->pCommQAck, pTBufHdr);
  2086.     
  2087.     pTBuf = (DIST_TBUF *) pTBufHdr;
  2088.     priority = pTBufHdr->tBufHdrPrio;
  2089.     while ((pTBuf = DIST_TBUF_GET_NEXT (pTBuf)) != NULL)
  2090.         {
  2091.         pTBuf->tBufId = pktId;
  2092.         pTBuf->tBufAck = (UINT16) winSub (pComm->commPktNextExpect, 1);
  2094. #ifdef DIST_NODE_REPORT
  2095.         printf ("distNodePktSend:%p: id %d, ack %d, seq %d, len %d, type %dn",
  2096.                 pTBuf, pTBuf->tBufId, pTBuf->tBufAck, pTBuf->tBufSeq,
  2097.                 pTBuf->tBufNBytes, pTBuf->tBufType);
  2098. #endif
  2099.         if (DIST_IF_SEND (nodeIdDest, pTBuf, priority) == ERROR)
  2100.             {
  2101.             distNodeDbUnlock ();
  2102.             DIST_TBUF_HDR_UNLINK (pComm->pCommQAck, pTBufHdr);
  2103.             DIST_TBUF_FREEM (pTBufHdr);
  2104.             return (ERROR);
  2105.             }
  2107.         pComm->commAckDelayed = FALSE;
  2108.         }
  2109.     
  2110.     distNodeDbUnlock ();    /* unlock before sleep */
  2112.     /* Wait for ACK. */
  2113.     
  2114. #ifdef DIST_NODE_REPORT
  2115.     distLog ("distNodePktSend: sleep while waiting for ACKn");
  2116. #endif
  2117.     semTake (&distTm.tmWait4, WAIT_FOREVER);
  2118. #ifdef DIST_NODE_REPORT
  2119.     distLog ("distNodePktSend: awacked: status of tm is %dn",
  2120.              distTm.tmStatus);
  2121. #endif
  2123.     /*
  2124.      * distNodePktReass() or distNodeTimer() has already
  2125.      * dequeued the packet.
  2126.      */
  2128.     DIST_TBUF_FREEM (pTBufHdr);
  2130.     if (distTm.tmStatus != DIST_TM_STATUS_OK)
  2131.         return (ERROR);                    /* e.g. timeout */
  2133.     return (OK);
  2134.     }
  2136. /***************************************************************************
  2137. *
  2138. * distNodePktResend - resend a packet (VxFusion option)
  2139. *
  2140. * Resend a packet already transmitted with distNodePktSend ().
  2141. *
  2142. * Currently called one place in distNodeDbCommTimer() which
  2143. * has <distNodeDbLock> taken before it is called.
  2144. *
  2145. * AVAILABILITY
  2146. * This routine is distributed as a component of the unbundled distributed
  2147. * message queues option, VxFusion.
  2148. *
  2149. * RETURNS: OK, if successful.
  2150. * NOMANUAL
  2151. */
  2153. LOCAL STATUS distNodePktResend
  2154.     (
  2155.     DIST_NODE_COMM * pComm,          /* communication channel */
  2156.     DIST_TBUF_HDR *  pTBufHdr        /* packet to retransmit */
  2157.     )
  2158.     {
  2159.     DIST_TBUF *     pTBuf;
  2160.     DIST_NODE_ID    nodeIdDest = pTBufHdr->tBufHdrDest;
  2161.     int             priority = pTBufHdr->tBufHdrPrio;
  2163.     distStat.nodePktResend++;
  2165.     pTBuf = (DIST_TBUF *) pTBufHdr;
  2166.     while ((pTBuf = DIST_TBUF_GET_NEXT (pTBuf)) != NULL)
  2167.         {
  2168.         pTBuf->tBufAck = (UINT16) winSub (pComm->commPktNextExpect, 1);
  2169. #ifdef DIST_NODE_REPORT
  2170.         printf ("distNodePktResend: resend with ACK for %d (%d)n",
  2171.                 pTBuf->tBufAck, pComm->commPktNextExpect);
  2172. #endif
  2174.         if (DIST_IF_SEND (nodeIdDest, pTBuf, priority) == ERROR)
  2175.             {
  2176.             DIST_TBUF_HDR_UNLINK (pComm->pCommQAck, pTBufHdr);
  2177.             DIST_TBUF_FREEM (pTBufHdr);
  2178.             return (ERROR);
  2179.             }
  2180.         }
  2182.     return (OK);
  2183.     }
  2184.     
  2185. /***************************************************************************
  2186. *
  2187. * distNodePktAck - ask for an ACK for a received and consumed packet (VxFusion option)
  2188. *
  2189. * This routine acknowledeges a packet. If DIST_NODE_ACK_IMMEDIATELY is
  2190. * not set in <options>, an ACK telegram is not transmitted immediately
  2191. * but a flag is set. If data is sent to the ACK awaiting node, within a
  2192. * certain range of time, the ACK is sent within the data telegram.
  2193. * Else an ACK telegram is transmitted. This is known as `piggy-backing'.
  2194. * Due to the design, piggy-backing only works for unicasts. For broadcasts,
  2195. * an ACK is always sent immediately.
  2196. *
  2197. * NOTE: Takes <distNodeDbLock>.
  2198. *
  2199. * AVAILABILITY
  2200. * This routine is distributed as a component of the unbundled distributed
  2201. * message queues option, VxFusion.
  2202. *
  2203. * RETURNS: OK, if successful.
  2204. * NOMANUAL
  2205. */
  2207. STATUS distNodePktAck
  2208.     (
  2209.     DIST_NODE_ID    nodeIdSrc,  /* source node */
  2210.     DIST_TBUF_HDR * pPktAck,    /* ack packet */
  2211.     int             options     /* options */
  2212.     )
  2213.     {
  2214.     DIST_NODE_DB_NODE * pDistNodeFound;
  2215.     DIST_NODE_COMM *    pComm;
  2216.     short               id = pPktAck->tBufHdrId;
  2217.     int                 ackBroadcast;
  2219. #ifdef DIST_DIAGNOSTIC
  2220.     if (nodeIdSrc == DIST_IF_BROADCAST_ADDR)
  2221.         distPanic ("distNodePktAck: cannot send ACK to broadcast node.n");
  2222. #endif
  2224.     distNodeDbLock ();
  2226.     /*
  2227.      * Find node, the acknowledge is destinated to.
  2228.      */
  2230.     if ((pDistNodeFound = distNodeFindById (nodeIdSrc)) == NULL)
  2231.         {
  2232. #ifdef DIST_DIAGNOSTIC
  2233.         distLog ("distNodePktAck: try to send ACK to unknown node 0x%lxn",
  2234.                  nodeIdSrc);
  2235. #endif
  2236.         distStat.nodeDBNoMatch++;
  2237.         distNodeDbUnlock ();
  2238.         return (ERROR);
  2239.         }
  2241.     /* Do we respond to a broadcast? */
  2243.     ackBroadcast = DIST_TBUF_IS_BROADCAST (DIST_TBUF_GET_NEXT (pPktAck));
  2245.     if (ackBroadcast)
  2246.         {
  2247.         pComm = &pDistNodeFound->nodeBroadcast;
  2248.         if (!distNodeSupportPBB)
  2249.             options |= DIST_NODE_ACK_IMMEDIATELY;
  2250. #ifdef DIST_NODE_REPORT
  2251.         printf ("distNodePktAck: ack broadcast %d form node 0x%lx immediatelyn",
  2252.                 id, nodeIdSrc);
  2253. #endif
  2254.         }
  2255.     else
  2256.         {
  2257.         pComm = &pDistNodeFound->nodeUnicast;
  2258.         if (!distNodeSupportPBU)
  2259.             options |= DIST_NODE_ACK_IMMEDIATELY;
  2260. #ifdef DIST_NODE_REPORT
  2261.         printf ("distNodePktAck: ack unicast %d form node 0x%lx %sn",
  2262.                 id, nodeIdSrc,
  2263.                 options
  2264.                 & DIST_NODE_ACK_IMMEDIATELY ? "immediately" : "delayed");
  2265. #endif
  2266.         }
  2267.     
  2268.     if (id != pComm->commPktNextExpect)    /* should never happen */
  2269.         distPanic ("distNodePktAck: acknowledge in wrong ordern");
  2271.     DIST_NODE_BCLR (pComm->pCommCompleted, id);
  2272.     pComm->commPktNextExpect = (INT16) winAdd (pComm->commPktNextExpect, 1);
  2274.     if (options & DIST_NODE_ACK_IMMEDIATELY)
  2275.         {
  2276.         /* Ack immediately. */
  2277.         
  2278.         distNodeSendAck (pDistNodeFound, ackBroadcast, options);
  2279.         }
  2280.     else
  2281.         {
  2282.         /*
  2283.          * Ask for a delayed ACK. If data is transfered before, the ACK
  2284.          * is piggy-backed with the data packet.
  2285.          */
  2287.         pComm->commAckDelayed = TRUE;
  2288.         }
  2290.     distNodeDbUnlock ();
  2292.     distStat.nodeAcked++;
  2293.     return (OK);
  2294.     }
  2296. /***************************************************************************
  2297. *
  2298. * distNodeSendNegAck - send a negative acknowledge (NACK) (VxFusion option)
  2299. *
  2300. * Send a negative acknowledgment, asking for a resend of fragment <seq>
  2301. * of packet <id>. A NACK (like an ACK) also holds the id of the last
  2302. * correctly received packet.
  2303. *
  2304. * NOTE: Currently, NACKs cannot be used in broadcast communication.
  2305. *
  2306. * AVAILABILITY
  2307. * This routine is distributed as a component of the unbundled distributed
  2308. * message queues option, VxFusion.
  2309. *
  2310. * RETURNS: OK, if sent.
  2311. * NOMANUAL
  2312. */
  2314. LOCAL STATUS distNodeSendNegAck
  2315.     (
  2316.     DIST_NODE_DB_NODE * pNode,   /* target node */
  2317.     short               id,      /* id */
  2318.     short               seq      /* seq number */
  2319.     )
  2320.     {
  2321.     DIST_TBUF * pTBuf;
  2322.     STATUS      status;
  2324.     if (! distNodeSupportNACK)
  2325.         return (OK);
  2327. #ifdef DIST_NODE_REPORT
  2328.     printf ("distNodeSendNegAck: NACK (id %d, seq %d) to node 0x%lxn",
  2329.             id, seq, pNode->nodeId);
  2330. #endif
  2331.     if ((pTBuf = DIST_TBUF_ALLOC ()) == NULL)
  2332.         return (ERROR);
  2334.     /* fill in telegram header */
  2335.     
  2336.     pTBuf->tBufId = id;              /* id of missing telegram */
  2337.     pTBuf->tBufSeq = seq;            /* sequence number of missing telegram */
  2338.     pTBuf->tBufAck = (UINT16)winSub (pNode->nodeUnicast.commPktNextExpect, 1);
  2339.     pTBuf->tBufNBytes = 0;
  2340.     pTBuf->tBufType = DIST_TBUF_TTYPE_NACK;
  2341.     pTBuf->tBufFlags = DIST_TBUF_FLAG_HDR;
  2343.     /* Send NACK. */
  2344.     
  2345.     if ((status = DIST_IF_SEND (pNode->nodeId, pTBuf, 0)) == OK)
  2346.         pNode->nodeUnicast.commAckDelayed = FALSE;
  2348.     DIST_TBUF_FREE (pTBuf);
  2349.     return (status);
  2350.     }
  2352. /***************************************************************************
  2353. *
  2354. * distNodeSendAck - send an ACK to the network (VxFusion option)
  2355. *
  2356. * Send an acknowledge telegram to the network. The ACK holds the
  2357. * id of the last correctly received packet.
  2358. *
  2359. * This is called from three places, all of which take <distNodeDbLock>.
  2360. *
  2361. * AVAILABILITY
  2362. * This routine is distributed as a component of the unbundled distributed
  2363. * message queues option, VxFusion.
  2364. *
  2365. * RETURNS: OK, if successful.
  2366. * NOMANUAL
  2367. */
  2369. LOCAL STATUS distNodeSendAck
  2370.     (
  2371.     DIST_NODE_DB_NODE * pNode,           /* target node */
  2372.     int                 ackBroadcast,    /* broadcast flag */
  2373.     int                 options          /* options */
  2374.     )
  2375.     {
  2376.     DIST_NODE_COMM * pComm;
  2377.     DIST_NODE_ID     nodeIdDest;
  2378.     DIST_TBUF *      pTBuf;
  2379.     STATUS           status;
  2381.     nodeIdDest = pNode->nodeId;
  2382.     pComm = (ackBroadcast ? &pNode->nodeBroadcast : &pNode->nodeUnicast);
  2384.     if ((pTBuf = DIST_TBUF_ALLOC ()) == NULL)
  2385.         return (ERROR);
  2387.     pTBuf->tBufId = 0;        /* the telegram id is ignored */
  2388.     pTBuf->tBufAck = (UINT16) winSub (pComm->commPktNextExpect, 1);
  2389.     pTBuf->tBufSeq = 0;
  2390.     pTBuf->tBufNBytes = 0;
  2391.     pTBuf->tBufFlags = DIST_TBUF_FLAG_HDR;
  2393.     if (ackBroadcast)
  2394.         {
  2395.         /* This is an ACK for a broadcast. */
  2396.         
  2397.         pTBuf->tBufType = DIST_TBUF_TTYPE_BACK;
  2398.         pTBuf->tBufFlags |= DIST_TBUF_FLAG_BROADCAST;
  2399. #ifdef DIST_NODE_REPORT
  2400.         printf ("distNodeSendAck: acknowledge broadcast %d from node 0x%lxn",
  2401.                 pTBuf->tBufAck, nodeIdDest);
  2402. #endif
  2403.         if (options & DIST_NODE_ACK_EXTENDED)
  2404.             {
  2405.             DIST_NODE_DB_NODE * pBroadcastNode;
  2406.             DIST_XACK *         pXAck = pTBuf->pTBufData;
  2407.             int                 nextSend;
  2409.             pBroadcastNode = distNodeFindById (DIST_IF_BROADCAST_ADDR);
  2410.             if (pBroadcastNode == NULL)
  2411.                 {
  2412.                 DIST_TBUF_FREE (pTBuf);
  2413.                 return (ERROR);
  2414.                 }
  2416.             nextSend = htons (pBroadcastNode->nodeBroadcast.commPktNextSend);
  2417.             pXAck->xAckPktNextSend = (UINT16) nextSend;
  2418.             pXAck->xAckPktNodeState = htons ((uint16_t) distNodeLocalState);
  2419.             pTBuf->tBufNBytes += sizeof (*pXAck);
  2420.             }
  2421.         }
  2422.     else
  2423.         {
  2424.         /* This is an ACK for a unicast. */
  2425.         
  2426.         pTBuf->tBufType = DIST_TBUF_TTYPE_ACK;
  2427. #ifdef DIST_NODE_REPORT
  2428.         printf ("distNodeSendAck: acknowledge unicast %d from node 0x%lxn",
  2429.                 pTBuf->tBufAck, nodeIdDest);
  2430. #endif
  2431.         if (options & DIST_NODE_ACK_EXTENDED)
  2432.             {
  2433.             DIST_XACK * pXAck = pTBuf->pTBufData;
  2435.             pXAck->xAckPktNextSend =
  2436.                          htons (pNode->nodeUnicast.commPktNextSend);
  2437.             pXAck->xAckPktNodeState = htons ((uint16_t) distNodeLocalState);
  2438.             pTBuf->tBufNBytes += sizeof (*pXAck);
  2439.             }
  2440.         }
  2442.     /* Send ACK. */
  2443.     
  2444.     if ((status = DIST_IF_SEND (nodeIdDest, pTBuf, 0)) == OK)
  2445.         pComm->commAckDelayed = FALSE;
  2447.     DIST_TBUF_FREE (pTBuf);
  2448.     return (status);
  2449.     }
  2451. /***************************************************************************
  2452. *
  2453. * distNodePktDiscard - discard a packet (VxFusion option)
  2454. *
  2455. * This routine discards a packet. No ACK is send.
  2456. *
  2457. * NOTE: Takes <distNodeDbLock>.
  2458. *
  2459. * AVAILABILITY
  2460. * This routine is distributed as a component of the unbundled distributed
  2461. * message queues option, VxFusion.
  2462. *
  2463. * RETURNS: OK, if successful.
  2464. * NOMANUAL
  2465. */
  2467. STATUS distNodePktDiscard
  2468.     (
  2469.     DIST_NODE_ID    nodeIdSrc,
  2470.     DIST_TBUF_HDR * pPktDiscard
  2471.     )
  2472.     {
  2473.     DIST_NODE_DB_NODE * pDistNodeFound;
  2474.     DIST_NODE_COMM *    pComm;
  2475.     short               id = pPktDiscard->tBufHdrId;
  2477.     distNodeDbLock ();
  2479.     if ((pDistNodeFound = distNodeFindById (nodeIdSrc)) == NULL)
  2480.         {
  2481. #ifdef DIST_DIAGNOSTIC
  2482.         distLog ("distNodePktDel: try to delete from unknown node 0x%lxn",
  2483.                  nodeIdSrc);
  2484. #endif
  2485.         distStat.nodeDBNoMatch++;
  2486.         distNodeDbUnlock ();
  2487.         return (ERROR);
  2488.         }
  2490. #ifdef DIST_NODE_REPORT
  2491.     printf ("distNodePktDiscard: discard packet %d from node 0x%lxn",
  2492.             id, nodeIdSrc);
  2493. #endif
  2495.     if (DIST_TBUF_IS_BROADCAST (DIST_TBUF_GET_NEXT (pPktDiscard)))
  2496.         pComm = &pDistNodeFound->nodeBroadcast;
  2497.     else
  2498.         pComm = &pDistNodeFound->nodeUnicast;
  2499.     
  2500.     distStat.nodeInDiscarded++;
  2502.     /* Clear id in bitfield, but do not move the window */
  2503.     
  2504.     DIST_NODE_BCLR (pComm->pCommCompleted, id);
  2506.     distNodeDbUnlock ();
  2508.     /* Free the discarded packet. */
  2509.     
  2510.     DIST_TBUF_FREEM (pPktDiscard);
  2512.     return (OK);
  2513.     }
  2515. /***************************************************************************
  2516. *
  2517. * distNodeDBTimerTask - node database manager task (VxFusion option)
  2518. *
  2519. * For now, distNodeDBTimer() is called from a task. The caller may be a
  2520. * watchdog in future.
  2521. *
  2522. * AVAILABILITY
  2523. * This routine is distributed as a component of the unbundled distributed
  2524. * message queues option, VxFusion.
  2525. *
  2526. * RETURNS: N/A
  2527. * NOMANUAL
  2528. */
  2530. LOCAL void distNodeDBTimerTask (void)
  2531.     {
  2532.     while (1)
  2533.         {
  2534.         distNodeDBTimer ();
  2535.         taskDelay (DIST_NODE_MGR_WAKEUP_TICKS);
  2536.         }
  2537.     }
  2539. /***************************************************************************
  2540. *
  2541. * distNodeDBTimer - periodic routine, to update the node database (VxFusion option)
  2542. *
  2543. * Updates each node of the database.
  2544. *
  2545. * AVAILABILITY
  2546. * This routine is distributed as a component of the unbundled distributed
  2547. * message queues option, VxFusion.
  2548. *
  2549. * RETURNS: N/A
  2550. * NOMANUAL
  2551. */
  2553. LOCAL void distNodeDBTimer (void)
  2554.     {
  2555.     DIST_NODE_DB_NODE * pNode;
  2556.     DIST_NODE_BTIMO     btimo;
  2558.     DIST_NODE_DB_LOCK;
  2559.     
  2560.     if ((pNode = distNodeFindById (DIST_IF_BROADCAST_ADDR)) != NULL)
  2561.         distNodeDBCommTimer (pNode, &btimo, TRUE);
  2563.     DIST_NODE_DB_UNLOCK;
  2565.     distNodeEach ((FUNCPTR) distNodeDBNodeTimer, (int) &btimo);
  2566.     }
  2568. /***************************************************************************
  2569. *
  2570. * distNodeDBNodeTimer - periodic routine, to update single node of node DB (VxFusion option)
  2571. *
  2572. * This routine handles unicast and broadcast communication of a specific
  2573. * node.
  2574. *
  2575. * NOTE: <distNodeDbLock> must be taken before this function is called.
  2576. * Currently, it is only invoked by distNodeEach(), which takes the lock.
  2577. *
  2578. * AVAILABILITY
  2579. * This routine is distributed as a component of the unbundled distributed
  2580. * message queues option, VxFusion.
  2581. *
  2582. * RETURNS: TRUE, if updated.
  2583. * NOMANUAL
  2584. */
  2586. LOCAL BOOL distNodeDBNodeTimer
  2587.     (
  2588.     DIST_NODE_DB_NODE * pNode,   /* node to update */
  2589.     DIST_NODE_BTIMO *   pBtimo   /* time interval */
  2590.     )
  2591.     {
  2592.     if (pNode->nodeId == DIST_IF_BROADCAST_ADDR)
  2593.         return (TRUE);
  2595.     distNodeDBCommTimer (pNode, pBtimo, FALSE);    /* unicast */
  2596.     distNodeDBCommTimer (pNode, pBtimo, TRUE);     /* broadcast */
  2597.     return (TRUE);    /* continue */
  2598.     }
  2600. /***************************************************************************
  2601. *
  2602. * distNodeDBCommTimer - periodic routine, to update a communication link (VxFusion option)
  2603. *
  2604. * NOTE: Must take <distNodeDbLock> before.  Currently called from
  2605. * distNodeDBTimer() and distNodeDBNodeTimer(), which take the lock.
  2606. *
  2607. * AVAILABILITY
  2608. * This routine is distributed as a component of the unbundled distributed
  2609. * message queues option, VxFusion.
  2610. *
  2611. * RETURNS: N/A
  2612. * NOMANUAL
  2613. */
  2615. LOCAL void distNodeDBCommTimer
  2616.     (
  2617.     DIST_NODE_DB_NODE * pNode,            /* node to update */
  2618.     DIST_NODE_BTIMO *   pBtimo,           /* time interval */
  2619.     BOOL                isBroadcastComm   /* broadcast boolean */
  2620.     )
  2621.     {
  2622.     DIST_NODE_COMM * pComm;
  2623.     DIST_TBUF_HDR *  pTBufWin;
  2624.     DIST_TBUF_HDR *  pTBufReass;
  2625.     DIST_TBUF_HDR *  pTBufAck;
  2626.     DIST_TBUF_HDR *  pTBufNext;
  2628.     /*
  2629.      * Define, what part of the node we are working on:
  2630.      * broadcast communication or unicast communication
  2631.      */
  2633.     if (isBroadcastComm)
  2634.         pComm = &pNode->nodeBroadcast;
  2635.     else
  2636.         pComm = &pNode->nodeUnicast;
  2638.     if (pNode->nodeId == DIST_IF_BROADCAST_ADDR)
  2639.         {
  2640.         pBtimo->btimoWinLo = pComm->commAckNextExpect;
  2641.         pBtimo->btimoWinHi = pComm->commPktNextSend;
  2642.         pBtimo->btimoTimedOut = FALSE;
  2643.         }
  2644.     else
  2645.         {
  2646.         if (isBroadcastComm && pBtimo->btimoTimedOut)
  2647.             {
  2648.             short    wl = pBtimo->btimoWinLo;
  2649.             short    wh = pBtimo->btimoWinHi;
  2651.             /*
  2652.              * Last ACK should fall in between this range. If not,
  2653.              * the node is declared dead.
  2654.              */
  2656.             if (! winWithin (wl, pComm->commAckLastRecvd, wh))
  2657.                 {
  2658.                 distNodeCrashed (pNode->nodeId);
  2659.                 return;
  2660.                 }
  2661.             }
  2662.         }
  2664.     /* Look for delayed ACKs. */
  2666.     if (pComm->commAckDelayed)
  2667.         {
  2668.         /* Send ACK. */
  2669.         
  2670.         distNodeSendAck (pNode, isBroadcastComm, 0);
  2671.         }
  2673.     /*
  2674.      * Look for packets that waited for first transmission,
  2675.      * but have timed out meanwhile.
  2676.      */
  2678.     pTBufNext = pComm->pCommQWinOut;
  2679.     while ((pTBufWin = pTBufNext) != NULL)
  2680.         {
  2681.         pTBufNext = DIST_TBUF_HDR_GET_NEXT (pTBufWin);
  2683.         /* Has user's timeout expired?  */
  2684.         
  2685.         if ((pTBufWin->tBufHdrTimo != WAIT_FOREVER) &&
  2686.             (pTBufWin->tBufHdrTimo -= DIST_NODE_MGR_WAKEUP_TICKS) <= 0)
  2687.             {
  2688.             /* Timed out. */
  2689.             
  2690.             pTBufWin->pTBufHdrTm->tmStatus = DIST_TM_STATUS_TIMEOUT;
  2691.             semGive (&pTBufWin->pTBufHdrTm->tmWait4);
  2692.             }
  2693.         }
  2695.     /*
  2696.      * Look for packets that were send, but not acknowledged
  2697.      * by now.
  2698.      */
  2700.     pTBufNext = pComm->pCommQAck;
  2701.     while ((pTBufAck = pTBufNext) != NULL)
  2702.         {
  2703.         pTBufNext = DIST_TBUF_HDR_GET_NEXT (pTBufAck);
  2705.         /* Has user's timeout expired? */
  2706.         
  2707.         if ((pTBufAck->tBufHdrTimo != WAIT_FOREVER) &&
  2708.             (pTBufAck->tBufHdrTimo -= DIST_NODE_MGR_WAKEUP_TICKS) <= 0)
  2709.             {
  2710.             /* Timed out. */
  2711.             
  2712.             if (distNodeLocalState == DIST_NODE_STATE_BOOT)
  2713.                 {
  2714.                 pComm->commAckNextExpect =
  2715.                     (INT16) winAdd (pComm->commAckNextExpect, 1);
  2716.                 pTBufAck->pTBufHdrTm->tmStatus = DIST_TM_STATUS_OK;
  2717.                 distNodeLocalState = DIST_NODE_STATE_NETWORK;
  2718.                 }
  2719.             else
  2720.                 pTBufAck->pTBufHdrTm->tmStatus = DIST_TM_STATUS_TIMEOUT;
  2722.             DIST_TBUF_HDR_UNLINK (pComm->pCommQAck, pTBufAck);
  2723.             semGive (&pTBufAck->pTBufHdrTm->tmWait4);
  2724.             continue;
  2725.             }
  2727.         /* Timeout for retransmisson. */
  2729.         if (pTBufAck->pTBufHdrTm->tmRetmTimo--  <= 0)
  2730.             {
  2731.             /* Timed out. Retransmit. */
  2732.     
  2733.             if (pTBufAck->pTBufHdrTm->tmNumTm >= distNodeMaxRetries + 1)
  2734.                 {
  2735.                 if (distNodeLocalState == DIST_NODE_STATE_BOOT)
  2736.                     {
  2737.                     pComm->commAckNextExpect =
  2738.                         (INT16) winAdd (pComm->commAckNextExpect, 1);
  2739.                     pTBufAck->pTBufHdrTm->tmStatus = DIST_TM_STATUS_OK;
  2740.                     distNodeLocalState = DIST_NODE_STATE_NETWORK;
  2741.                     }
  2742.                 else
  2743.                     {
  2744.                     if (pNode->nodeId == DIST_IF_BROADCAST_ADDR)
  2745.                         {
  2746.                         short    wl = pBtimo->btimoWinLo;
  2747.                         short    wh = pComm->commPktNextSend;
  2749.                         if (winWithin (wl, pTBufAck->tBufHdrId, wh))
  2750.                             pBtimo->btimoWinLo = pTBufAck->tBufHdrId;
  2752.                         pBtimo->btimoTimedOut = TRUE;
  2753.                         }
  2754.                     else
  2755.                         {
  2756.                         /*
  2757.                          * The remote note cannot be reached. Declare the
  2758.                          * node to be dead and set the transmission status
  2759.                          * to "unreachable". Wake up sender.
  2760.                          */
  2762.                         (void) distNodeCrashed (pNode->nodeId);
  2763.                         }
  2765.                     pTBufAck->pTBufHdrTm->tmStatus = DIST_TM_STATUS_UNREACH;
  2766.                     }
  2768.                 DIST_TBUF_HDR_UNLINK (pComm->pCommQAck, pTBufAck);
  2769.                 semGive (&pTBufAck->pTBufHdrTm->tmWait4);
  2770.                 }
  2771.             else
  2772.                 {
  2773.                 distNodePktResend (pComm, pTBufAck);
  2775.                 pTBufAck->pTBufHdrTm->tmNumTm++;
  2776.                 pTBufAck->pTBufHdrTm->tmRetmTimo =
  2777.                         pTBufAck->pTBufHdrTm->tmNumTm * distNodeRetryTimeout;
  2778.                 }
  2779.             }
  2780.         }
  2782.     /*
  2783.      * Look for packets that were received in fragments, but not completed
  2784.      * by now.
  2785.      */
  2787.     pTBufNext = pComm->pCommQReass;
  2788.     while ((pTBufReass = pTBufNext) != NULL)
  2789.         {
  2790.         pTBufNext = DIST_TBUF_HDR_GET_NEXT (pTBufReass);
  2792.         if ((pTBufReass->tBufHdrTimo != WAIT_FOREVER) &&
  2793.             (pTBufReass->tBufHdrTimo -= DIST_NODE_MGR_WAKEUP_TICKS) <= 0)
  2794.             {
  2795.             distNodeCrashed (pTBufReass->tBufHdrSrc);
  2796.             DIST_TBUF_HDR_UNLINK (pComm->pCommQReass, pTBufReass);
  2797.             pComm->commPktNextExpect =
  2798.                              (INT16)winAdd (pTBufReass->tBufHdrId, 1);
  2799.             DIST_TBUF_FREEM (pTBufReass);
  2800.             }
  2801.         }
  2802.     }
  2804. #ifdef DIST_DIAGNOSTIC_SHOW
  2806. /***************************************************************************
  2807. *
  2808. * distNodeShow - show routine for a node from node database (VxFusion option)
  2809. *
  2810. * This routine looks up node <distNodeId> and displays information
  2811. * from node database.
  2812. * Only intended for internal use.
  2813. *
  2814. * NOTE: Takes <distNodeDbLock>.
  2815. *
  2816. * AVAILABILITY
  2817. * This routine is distributed as a component of the unbundled distributed
  2818. * message queues option, VxFusion.
  2819. *
  2820. * RETURNS: OK, if processed.
  2821. * NOMANUAL
  2822. */
  2824. STATUS distNodeShow
  2825.     (
  2826.     DIST_NODE_ID    distNodeId    /* node ID to show */
  2827.     )
  2828.     {
  2829.     DIST_NODE_DB_NODE * pDistNodeFound;
  2831.     distNodeDbLock ();
  2833.     if ((pDistNodeFound = distNodeFindById (distNodeId)) == NULL)
  2834.             {
  2835.             distNodeDbUnlock ();
  2836.             return (ERROR);
  2837.             }
  2839.     printf ("%d: state %dn",
  2840.             (int) pDistNodeFound->nodeId,
  2841.             pDistNodeFound->nodeState);
  2843.     printf ("  unicast/in: buffer: ");
  2844.     DIST_NODE_BPRINT (pDistNodeFound->nodeUnicast.pCommCompleted,
  2845.             DIST_IF_RNG_BUF_SZ);
  2846.     printf ("n");
  2847.     printf ("  unicast/in: reassQ %p, dlvr nxt %pn",
  2848.             pDistNodeFound->nodeUnicast.pCommQReass,
  2849.             pDistNodeFound->nodeUnicast.pCommQNextDeliver);
  2850.     printf ("  unicast/in: expect nxt %d, last ack rcvd %dn",
  2851.             pDistNodeFound->nodeUnicast.commPktNextExpect,
  2852.             pDistNodeFound->nodeUnicast.commAckLastRecvd);
  2853.     printf ("  unicast/out: winOutQ %p, ackQ %p,n",
  2854.             pDistNodeFound->nodeUnicast.pCommQWinOut,
  2855.             pDistNodeFound->nodeUnicast.pCommQAck);
  2856.     printf ("  unicast/out: nxt ack xpct %d, nxt snd %d, ack dlyd %dn",
  2857.             pDistNodeFound->nodeUnicast.commAckNextExpect,
  2858.             pDistNodeFound->nodeUnicast.commPktNextSend,
  2859.             pDistNodeFound->nodeUnicast.commAckDelayed);
  2861.     printf ("  broadcast/in: buffer: ");
  2862.     DIST_NODE_BPRINT (pDistNodeFound->nodeBroadcast.pCommCompleted,
  2863.             DIST_IF_RNG_BUF_SZ);
  2864.     printf ("n");
  2865.     printf ("  broadcast/in: reassQ %p, dlvr nxt %pn",
  2866.             pDistNodeFound->nodeBroadcast.pCommQReass,
  2867.             pDistNodeFound->nodeBroadcast.pCommQNextDeliver);
  2868.     printf ("  broadcast/in: expect next %d, last ack rcvd %dn",
  2869.             pDistNodeFound->nodeBroadcast.commPktNextExpect,
  2870.             pDistNodeFound->nodeBroadcast.commAckLastRecvd);
  2871.     printf ("  broadcast/out: winOutQ %p, ackQ %p,n",
  2872.             pDistNodeFound->nodeBroadcast.pCommQWinOut,
  2873.             pDistNodeFound->nodeBroadcast.pCommQAck);
  2874.     printf ("  broadcast/out: nxt ack xpct %d, nxt snd %d, ack dlyd %dn",
  2875.             pDistNodeFound->nodeBroadcast.commAckNextExpect,
  2876.             pDistNodeFound->nodeBroadcast.commPktNextSend,
  2877.             pDistNodeFound->nodeBroadcast.commAckDelayed);
  2879.     distNodeDbUnlock ();
  2880.     return (OK);
  2881.     }
  2883. /***************************************************************************
  2884. *
  2885. * distNodeDbShow - brief overview of node database (VxFusion option)
  2886. *
  2887. * Print a summary of the database.
  2888. *
  2889. * AVAILABILITY
  2890. * This routine is distributed as a component of the unbundled distributed
  2891. * message queues option, VxFusion.
  2892. *
  2893. * RETURNS: OK.
  2894. * NOMANUAL
  2895. */
  2897. STATUS distNodeDbShow (void)
  2898.     {
  2899.     printf ("                       UNICAST------------------- ");
  2900.     printf ("BROADCAST-----------------n");
  2901.     printf ("NODE ID    STATE       PKT XPCT ACK XPCT  PKT SND ");
  2902.     printf ("PKT XPCT ACK XPCT  PKT SNDn");
  2903.     printf ("---------- ----------- -------- -------- -------- ");
  2904.     printf ("-------- -------- --------n");
  2905.     distNodeEach ((FUNCPTR) distNodeNodeShow, 0);
  2906.     return (OK);
  2907.     }
  2909. /***************************************************************************
  2910. *
  2911. * distNodeNodeShow - helper for distNodeDbShow() (VxFusion option)
  2912. *
  2913. * Print database node info.
  2914. *
  2915. * AVAILABILITY
  2916. * This routine is distributed as a component of the unbundled distributed
  2917. * message queues option, VxFusion.
  2918. *
  2919. * RETURNS: TRUE.
  2920. * NOMANUAL
  2921. */
  2923. LOCAL BOOL distNodeNodeShow
  2924.     (
  2925.     DIST_NODE_DB_NODE * pNode,    /* node to display */
  2926.     int                 dummy     /* unused argument */
  2927.     )
  2928.     {
  2929.     char * stateNm;
  2931.     printf ("0x%8lx ", pNode->nodeId);
  2932.     switch (pNode->nodeState)
  2933.         {
  2934.         case DIST_NODE_STATE_BOOT:
  2935.             stateNm = "booting";
  2936.             break;
  2937.         case DIST_NODE_STATE_NETWORK:
  2938.             stateNm = "network";
  2939.             break;
  2940.         case DIST_NODE_STATE_OPERATIONAL:
  2941.             stateNm = "operational";
  2942.             break;
  2943.         case DIST_NODE_STATE_CRASHED:
  2944.             stateNm = "crashed";
  2945.             break;
  2946.         case DIST_NODE_STATE_RESYNC:
  2947.             stateNm = "resync";
  2948.             break;
  2949.         default:
  2950.             stateNm = "unknown";
  2951.         }
  2952.     printf ("%11s ", stateNm);
  2954.     if (pNode->nodeId == DIST_IF_BROADCAST_ADDR)
  2955.         {
  2956.         printf ("      --       --       -- ");
  2957.         printf ("      -- %8d %8dn",
  2958.                 pNode->nodeBroadcast.commAckNextExpect,
  2959.                 pNode->nodeBroadcast.commPktNextSend);
  2960.         }
  2961.     else
  2962.         {
  2963.         printf ("%8d %8d %8d ",
  2964.                 pNode->nodeUnicast.commPktNextExpect,
  2965.                 pNode->nodeUnicast.commAckNextExpect,
  2966.                 pNode->nodeUnicast.commPktNextSend);
  2967.         printf ("%8d       --       --n",
  2968.                 pNode->nodeBroadcast.commPktNextExpect);
  2969.         }
  2971.     return (TRUE);
  2972.     }
  2976. /***************************************************************************
  2977. *
  2978. * distNodeStateToName - helper for distNodeDbShow() (VxFusion option)
  2979. *
  2980. * Return string describing constant.
  2981. *
  2982. * AVAILABILITY
  2983. * This routine is distributed as a component of the unbundled distributed
  2984. * message queues option, VxFusion.
  2985. *
  2986. * RETURNS: Descriptive string.
  2987. * NOMANUAL
  2988. */
  2989. LOCAL const char *distNodeStateToName
  2990.     (
  2991.     int state      /* state to decode */
  2992.     )
  2993.     {
  2994. LOCAL const char *const dead[] =
  2995.     {"CRASHED"};
  2997. LOCAL const char *const alive[] =
  2998.     {"BOOTING", "RESYNC", "NETWORK", "OPERATIONAL"};
  3000.     if (DIST_NODE_STATE_IS_ALIVE (state))
  3001.         return (alive[((state & ~DIST_NODE_STATE_ALIVE) - 1)]);
  3002.     else
  3003.         return (dead[((state & ~DIST_NODE_STATE_ALIVE) - 1)]);
  3004.     }
  3006. #endif    /* DIST_DIAGNOSTIC_SHOW */