DIST_TBUF_FREE (pTBufNew);                        return (NULL);                        }                        if (pTBuf->tBufSeq > seq)                        {                        DIST_TBUF_INSERT_AFTER (pReassHdr,                                                pTBufPrev,                                                pTBufNew);                        pReassHdr->tBufHdrNLeaks--;                        break;                        }                    pTBufPrev = pTBuf;                    pTBuf = DIST_TBUF_GET_NEXT (pTBuf);                    }                }            /* The telegram is in the reassembly list now. */            pReassHdr->tBufHdrOverall += nBytes;            /*             * Test if the packet is fully reassembled (there             * are no leaks within the packet and the last             * fragment has the 'more fragments' bit cleared).             */            pTBufLast = DIST_TBUF_GET_LAST (pReassHdr);            if (!pReassHdr->tBufHdrNLeaks && !(DIST_TBUF_HAS_MF (pTBufLast)))                {                /*                 * We have received a complete packet and                 * do not expect to receive more fragments.                 */#ifdef DIST_NODE_REPORT                {                DIST_TBUF    *pFirst = DIST_TBUF_GET_NEXT (pReassHdr);                                /* this is a hack */                                printf ("dist..Reass: packet %d complete (type %d/%d)\n", id,                        *((char *) (pFirst->pTBufData)),                        *((char *) (pFirst->pTBufData) + 1));                }#endif                DIST_NODE_BSET (pCommIn->pCommCompleted, id);                /*                 * Test if the reassembled packet is the one                 * we expect next.                 */                if (pReassHdr->tBufHdrId == pCommIn->commPktNextExpect)                    {                    /*                     * This is the packet, we expect next.                     * Return communication pointer to caller,                     * so that it knows about the job.                     */                    distNodeDbUnlock ();                    return (pCommIn);                    }                }            distNodeDbUnlock ();            return (NULL);            }        pReassHdrPrev = pReassHdr;        pReassHdr = DIST_TBUF_HDR_GET_NEXT (pReassHdr);        }        /*     * This is the first fragment of a new packet. There are either     * more fragments, or this packet is not the one, we are waiting     * for.     */     #ifdef DIST_NODE_REPORT    printf ("dist..Reass: first fragment of new packet\n");#endif    if ((pTBufHdrNew = DIST_TBUF_HDR_ALLOC ()) == NULL)        {        distStat.netInDiscarded++; /* nodeInDiscarded counts pkts discarded */        distNodeDbUnlock ();        DIST_TBUF_FREE (pTBufNew);        return (NULL);        }    pTBufHdrNew->tBufHdrSrc = nodeIdSrc;    pTBufHdrNew->tBufHdrId = id;    pTBufHdrNew->tBufHdrPrio = prio;    pTBufHdrNew->tBufHdrOverall = nBytes;    pTBufHdrNew->tBufHdrNLeaks = seq;    pTBufHdrNew->tBufHdrTimo = WAIT_FOREVER;    /* reassembly timeout */    DIST_TBUF_SET_NEXT (pTBufHdrNew, pTBufNew);    DIST_TBUF_SET_LAST (pTBufHdrNew, pTBufNew);    DIST_TBUF_HDR_INSERT_AFTER (pCommIn->pCommQReass, pReassHdrPrev,        pTBufHdrNew);    if (pCommIn->commPktNextExpect == id)        /* this is the packet we want to deliver next */        pCommIn->pCommQNextDeliver = pTBufHdrNew;    if (seq == 0 && ! hasMf)        {        /* We have received a complete packet in a single telegram. */#ifdef DIST_NODE_REPORT        /* this is a hack */                printf ("dist..Reass: packet %d complete (type %d/%d)\n", id,                *((char *) (pTBufNew->pTBufData)),                *((char *) (pTBufNew->pTBufData) + 1));#endif        DIST_NODE_BSET (pCommIn->pCommCompleted, id);        if (id == pCommIn->commPktNextExpect)            {            /*             * This is the packet, we expect next.             * Return communication pointer to caller,             * so that it knows about the job.             */            distNodeDbUnlock ();            return (pCommIn);            }        }    distNodeDbUnlock ();    /* Check if we have to send negative acknowledges (NACKs). */        if (! DIST_TBUF_IS_BROADCAST (pTBufNew))        {        DIST_TBUF_HDR    *pPrev;        DIST_TBUF        *pLastOfPrev;        /*         * Check if we have received the last fragment of the previous         * packet. We cannot find out how many fragments are missing at         * the end of the packet. So just send a NACK for the successor         * of the last telegram received.         */        if ((pPrev = DIST_TBUF_HDR_GET_PREV (pTBufHdrNew)) != NULL &&                (pLastOfPrev = DIST_TBUF_GET_LAST (pTBufHdrNew)) != NULL &&                DIST_TBUF_HAS_MF (pLastOfPrev))            {            distNodeSendNegAck (pNodeSrc, pPrev->tBufHdrId,                    pLastOfPrev->tBufSeq + 1);            }        /*         * Check if some telegrams from the beginning of the new packet         * are missing. Send a NACK for each of them.         */        if (seq > 0)            {            INT16 i;            for (i = 0; i < seq; i++)                distNodeSendNegAck (pNodeSrc, id, i);            }        }    return (NULL);    }/***************************************************************************** distNodePktSend - send a packet already stored in TBufs (VxFusion option)** This routine waits for an open send window and transmits the packet,* by sending the data of each single TBuf.* Afterwards it blocks for an ACK from the remote node.** NOTE: Takes <distNodeDbLock>.** AVAILABILITY* This routine is distributed as a component of the unbundled distributed* message queues option, VxFusion.** RETURNS: OK, if successful.* NOMANUAL*/STATUS distNodePktSend    (    DIST_TBUF_HDR * pTBufHdr    /* message to send */    )    {    DIST_NODE_DB_NODE *  pDistNodeFound;    DIST_TRANSMISSION    distTm;    DIST_NODE_COMM *     pComm;    DIST_NODE_ID         nodeIdDest = pTBufHdr->tBufHdrDest;    DIST_TBUF *          pTBuf;    short                winLo, winHi;    short                pktId;    int                  priority;    int                  wait4NumAcks;    distNodeDbLock ();    distStat.nodeOutReceived++;    if ((pDistNodeFound = distNodeFindById (nodeIdDest)) == NULL)        {#ifdef DIST_DIAGNOSTIC        distLog ("distNodePktSend: unknown node 0x%lx\n", nodeIdDest);#endif        distStat.nodeDBNoMatch++;        distNodeDbUnlock ();        DIST_TBUF_FREEM (pTBufHdr);        return (ERROR);        }    if (! DIST_NODE_IS_ALIVE (pDistNodeFound))        {#ifdef DIST_DIAGNOSTIC        distLog ("distNodePktSend: destination not alive\n");#endif        distStat.nodeNotAlive++;        distNodeDbUnlock ();        DIST_TBUF_FREEM (pTBufHdr);        return (ERROR);        }#ifdef DIST_DIAGNOSTIC    if (DIST_TBUF_GET_NEXT (pTBufHdr) == NULL)        {        distLog ("distNodePktSend: TBuf chain contains no data\n");        }#endif    if (nodeIdDest == DIST_IF_BROADCAST_ADDR)        {        pComm = &pDistNodeFound->nodeBroadcast;        wait4NumAcks = distNodeNumNodesAlive - 1;        }    else        {        pComm = &pDistNodeFound->nodeUnicast;        wait4NumAcks = 1;        }    semBInit (&distTm.tmWait4, SEM_Q_FIFO, SEM_EMPTY);    distTm.tmStatus = DIST_TM_STATUS_OK;    distTm.tmRetmTimo = distNodeRetryTimeout;    distTm.tmNumTm = 1;    distTm.tmAckCnt = wait4NumAcks;    pTBufHdr->pTBufHdrTm = &distTm;    /*     * Each node has two queues for outgoing packets.     * One is the queue for packets that should been     * send, but the "window is closed" (full).     * The other is the queue for packets send, and     * waiting for an ACK.     * In both of the two queues, packets get older.     */    pTBufHdr->tBufHdrId = pktId = pComm->commPktNextSend;    winLo = pComm->commAckNextExpect;    winHi = (INT16) winAdd (pComm->commAckNextExpect, DIST_NODE_WIN_SZ);    /*     * Can we send the packet directly or do we have to wait     * until the "window opens".     */    if (! winWithin (winLo, pktId, winHi))        {        /* Put packet to the WaitToSend queue. */                DIST_TBUF_HDR_ENQUEUE (pComm->pCommQWinOut, pTBufHdr);        distNodeDbUnlock ();    /* unlock before sleep */        /* Wait for a place within the window. */                semTake (&distTm.tmWait4, WAIT_FOREVER);        if (distTm.tmStatus != DIST_TM_STATUS_OK)            {            distNodeDbUnlock ();       /* this is not needed here! */            DIST_TBUF_FREEM (pTBufHdr);            return (ERROR);            /* e.g. timeout, node removed */            }        distNodeDbLock ();                /* Remove it from the WaitToSend queue. */                DIST_TBUF_HDR_UNLINK (pComm->pCommQWinOut, pTBufHdr);        }    pComm->commPktNextSend = (INT16) winAdd (pComm->commPktNextSend, 1);    /* Put packet to the WaitForAck queue and send it. */        DIST_TBUF_HDR_ENQUEUE (pComm->pCommQAck, pTBufHdr);        pTBuf = (DIST_TBUF *) pTBufHdr;    priority = pTBufHdr->tBufHdrPrio;    while ((pTBuf = DIST_TBUF_GET_NEXT (pTBuf)) != NULL)        {        pTBuf->tBufId = pktId;        pTBuf->tBufAck = (UINT16) winSub (pComm->commPktNextExpect, 1);#ifdef DIST_NODE_REPORT        printf ("distNodePktSend:%p: id %d, ack %d, seq %d, len %d, type %d\n",                pTBuf, pTBuf->tBufId, pTBuf->tBufAck, pTBuf->tBufSeq,                pTBuf->tBufNBytes, pTBuf->tBufType);#endif        if (DIST_IF_SEND (nodeIdDest, pTBuf, priority) == ERROR)            {            distNodeDbUnlock ();            DIST_TBUF_HDR_UNLINK (pComm->pCommQAck, pTBufHdr);            DIST_TBUF_FREEM (pTBufHdr);            return (ERROR);            }        pComm->commAckDelayed = FALSE;        }        distNodeDbUnlock ();    /* unlock before sleep */    /* Wait for ACK. */    #ifdef DIST_NODE_REPORT    distLog ("distNodePktSend: sleep while waiting for ACK\n");#endif    semTake (&distTm.tmWait4, WAIT_FOREVER);#ifdef DIST_NODE_REPORT    distLog ("distNodePktSend: awacked: status of tm is %d\n",             distTm.tmStatus);#endif    /*     * distNodePktReass() or distNodeTimer() has already     * dequeued the packet.     */    DIST_TBUF_FREEM (pTBufHdr);    if (distTm.tmStatus != DIST_TM_STATUS_OK)        return (ERROR);                    /* e.g. timeout */    return (OK);    }/***************************************************************************** distNodePktResend - resend a packet (VxFusion option)** Resend a packet already transmitted with distNodePktSend ().** Currently called one place in distNodeDbCommTimer() which* has <distNodeDbLock> taken before it is called.** AVAILABILITY* This routine is distributed as a component of the unbundled distributed* message queues option, VxFusion.** RETURNS: OK, if successful.* NOMANUAL*/LOCAL STATUS distNodePktResend    (    DIST_NODE_COMM * pComm,          /* communication channel */    DIST_TBUF_HDR *  pTBufHdr        /* packet to retransmit */    )    {    DIST_TBUF *     pTBuf;    DIST_NODE_ID    nodeIdDest = pTBufHdr->tBufHdrDest;    int             priority = pTBufHdr->tBufHdrPrio;    distStat.nodePktResend++;    pTBuf = (DIST_TBUF *) pTBufHdr;    while ((pTBuf = DIST_TBUF_GET_NEXT (pTBuf)) != NULL)        {        pTBuf->tBufAck = (UINT16) winSub (pComm->commPktNextExpect, 1);#ifdef DIST_NODE_REPORT        printf ("distNodePktResend: resend with ACK for %d (%d)\n",                pTBuf->tBufAck, pComm->commPktNextExpect);#endif        if (DIST_IF_SEND (nodeIdDest, pTBuf, priority) == ERROR)            {            DIST_TBUF_HDR_UNLINK (pComm->pCommQAck, pTBufHdr);            DIST_TBUF_FREEM (pTBufHdr);            return (ERROR);            }        }    return (OK);    }    /***************************************************************************** distNodePktAck - ask for an ACK for a received and consumed packet (VxFusion option)** This routine acknowledeges a packet. If DIST_NODE_ACK_IMMEDIATELY is* not set in <options>, an ACK telegram is not transmitted immediately* but a flag is set. If data is sent to the ACK awaiting node, within a* certain range of time, the ACK is sent within the data telegram.* Else an ACK telegram is transmitted. This is known as `piggy-backing'.* Due to the design, piggy-backing only works for unicasts. For broadcasts,* an ACK is always sent immediately.** NOTE: Takes <distNodeDbLock>.** AVAILABILITY* This routine is distributed as a component of the unbundled distributed* message queues option, VxFusion.** RETURNS: OK, if successful.* NOMANUAL*/STATUS distNodePktAck    (    DIST_NODE_ID    nodeIdSrc,  /* source node */    DIST_TBUF_HDR * pPktAck,    /* ack packet */    int             options     /* options */    )    {    DIST_NODE_DB_NODE * pDistNodeFound;    DIST_NODE_COMM *    pComm;    short               id = pPktAck->tBufHdrId;    int                 ackBroadca