distnodelib.c

VXWORKS源代码

/* distNodeLib - node library (VxFusion option) */

/* Copyright 1999 - 2002 Wind River Systems, Inc. */

/*
modification history
--------------------
01q,15apr02,jws  fix SPR74878 (not locking node database properly)
01p,23oct01,jws  fix compiler warnings (SPR 71117)
01o,04oct01,jws  final fixes for SPR 34770
01n,27sep01,p_r  Fixes for SPR#34770
01m,11jun99,drm  Changing unicast piggy-backing default to FALSE.
01l,24may99,drm  added vxfusion prefix to VxFusion related includes
01k,11sep98,drm  added #include to pick up distPanic()
01j,12aug98,drm  added #include stmt for distObjTypeP.h
01i,08aug98,drm  added code to set broadcast flag when sending broadcast msgs
01h,08apr98,ur   made `piggy backing' switchable with distNodeCtl()
01g,13mar98,ur   added support for multiple crashed and operational hooks
01f,29jan98,ur   added support for sending negative acknowledgments
01e,23jan98,ur   bug fixed in the delivery of reassembled packets
01d,20jan98,ur   splited distNodeInit() in two parts
01c,19jan98,ur   bug fixed in distNodePktAck()
01b,16jan98,ur   XACK contains state of sending node
01a,10jun97,ur   written.
*/

/*
DESCRIPTION
This library contains the node database, and routines to handle it.
For every node in the system, one entry should be placed in the database.
The database knows about the state of the node (as it looks from the local
node) and about communication between the local node and the remote one.

INTERNAL
The semaphore <distNodeDbSem> is used to lock the node database.  The
database is a hash table of nodes in the system.  Packets chained for
input/output are part of each node's entry.  The table must be locked
when entering an entry.  It is also locked when finding and entry, although
it is not clear that this is necessary because entries are never removed.

The DIST_NODE_COMM structures are the packet queues.  The database must
be locked when dealing with these structures.  It is believed that this
is now true.  Because of the way the communications routines are called,
<distNodeDbSem> is now a mutex, and is sometimes called recursively.

AVAILABILITY
This module is distributed as a component of the unbundled distributed
message queues option, VxFusion.
*/

#include "vxWorks.h"

#undef  DIST_DIAGNOSTIC   /* defining this seems to break VxFusion! */
#undef  DIST_DIAGNOSTIC_SHOW

#if defined (DIST_NODE_REPORT) \
 || defined (DIST_DIAGNOSTIC) \
 || defined (DIST_DIAGNOSTIC_SHOW)

#include "stdio.h"

#endif

#include "stdlib.h"
#include "hashLib.h"
#include "semLib.h"
#include "taskLib.h"
#include "string.h"
#include "errnoLib.h"
#include "netinet/in.h"
#include "private/distObjTypeP.h"
#include "vxfusion/distLib.h"
#include "vxfusion/distIfLib.h"
#include "vxfusion/distNodeLib.h"
#include "vxfusion/distStatLib.h"
#include "vxfusion/private/distLibP.h"
#include "vxfusion/private/distNodeLibP.h"
#include "vxfusion/private/distTBufLibP.h"
#include "vxfusion/private/distPktLibP.h"

/* defines */

/* make two macros look like macros */

#define DIST_NODE_DB_LOCK    distNodeDbLock()
#define DIST_NODE_DB_UNLOCK  distNodeDbUnlock()


#define UNUSED_ARG(x)  if(sizeof(x)) {} /* to suppress compiler warnings */

/* test if b is within [a,c) */
#define winWithin(a, b, c) \
    (((a <= b) && (b < c)) || ((c < a) && (a <= b)) || ((b < c) && (c < a)))

/* add a to b and fold it back */
#define winAdd(a, b) \
    (((a) + (b)) & (DIST_IF_RNG_BUF_SZ - 1))

/* sub a by b and fold it back */
#define winSub(a, b) \
    (((a) - (b)) & (DIST_IF_RNG_BUF_SZ - 1))

#define KEY_ARG              65537        /* used by hash function */
#define KEY_CMP_ARG          0            /* unused */

#define DIST_NODE_MAX_HOOKS  8

/* global variables */

DIST_NODE_ID distNodeLocalId;   /* windSh needs this global */

/*
 * The semaphore used to be binary, but is now initialized as mutex.
 * It cannot be changed back, because it is sometimes taken recursively,
 * and will block if not mutex.
 */
SEM_ID       distNodeDbSem;     /* Should be taken before DIST_TBUF_FREEM */

/* local variables */

LOCAL HASH_ID     distNodeDbId;
LOCAL BOOL        distNodeLibInstalled = FALSE;

LOCAL int        distNodeNumNodesAll = 0;
LOCAL int        distNodeNumNodesAlive = 0;
LOCAL int        distNodeNumNodesOperational = 0;

LOCAL int        distNodeLocalState;

LOCAL DIST_NODE_ID    distNodeGodfatherId;

LOCAL FUNCPTR        distNodeOperationalHook[DIST_NODE_MAX_HOOKS];
LOCAL FUNCPTR        distNodeCrashedHook[DIST_NODE_MAX_HOOKS];

LOCAL int        distNodeMaxRetries = DIST_NODE_MAX_RETRIES;
LOCAL int        distNodeRetryTimeout = DIST_NODE_RETRY_TIMEOUT;

LOCAL BOOL        distNodeSupportNACK = TRUE;   /* negative acknowlege */
LOCAL BOOL        distNodeSupportPBB = FALSE;   /* piggy backing broadcast */
LOCAL BOOL        distNodeSupportPBU = FALSE;   /* piggy backing unicast */

/* local prototypes */

LOCAL DIST_NODE_DB_NODE * distNodeFindById (DIST_NODE_ID nodeId);

LOCAL BOOL   distNodeHCmp (DIST_NODE_DB_NODE *pMatchNode,
                           DIST_NODE_DB_NODE *pHNode,
                           int keyArg);
LOCAL BOOL   distNodeHFunc (int elements, DIST_NODE_DB_NODE *pHNode,
                            int keyArg);
LOCAL STATUS distNodePktResend (DIST_NODE_COMM *pComm,
                                DIST_TBUF_HDR *pTBufHdr);
LOCAL void   distNodeDBTimerTask (void);
LOCAL void   distNodeDBTimer (void);
LOCAL BOOL   distNodeDBNodeTimer (DIST_NODE_DB_NODE *pNode,
                                  DIST_NODE_BTIMO *pBtimo);
LOCAL void   distNodeDBCommTimer (DIST_NODE_DB_NODE *pNode,
                                  DIST_NODE_BTIMO *pBtimo,
                                  BOOL isBroadcastComm);
LOCAL STATUS distNodeSendNegAck (DIST_NODE_DB_NODE *pNode,
                                 short id,
                                 short seq);
LOCAL STATUS distNodeSendAck (DIST_NODE_DB_NODE *pNode,
                              int ackBroadcast,
                              int options);
LOCAL void   distNodeCleanup (DIST_NODE_DB_NODE *pNode);
LOCAL void   distNodeSetState (DIST_NODE_DB_NODE *pNode,
                               int state);
LOCAL STATUS distNodeSendBootstrap (DIST_NODE_ID dest,
                                    int type,
                                    int timeout);

#if defined(DIST_DIAGNOSTIC) || defined(DIST_DIAGNOSTIC_SHOW)
LOCAL BOOL         distNodeNodeShow (DIST_NODE_DB_NODE *pNode,
                                     int dummy);
LOCAL const char * distNodeStateToName (int state);
#endif

/***************************************************************************
*
* distNodeLibInit - initialize this module (VxFusion option)
*
* This routine currently does nothing.
*
* AVAILABILITY
* This routine is distributed as a component of the unbundled distributed
* message queues option, VxFusion.
*
* RETURNS: N/A
* NOMANUAL
*/

void distNodeLibInit (void)
    {
    }

/***************************************************************************
*
* distNodeInit - initializes the node library (VxFusion option)
*
* This routine initializes the node database. The database can handle
* up to 2^<sizeLog2> entries.
*
* AVAILABILITY
* This routine is distributed as a component of the unbundled distributed
* message queues option, VxFusion.
*
* RETURNS: OK, if successful.
* NOMANUAL
*/

STATUS distNodeInit
    (
    int sizeLog2   /* init database with 2^^sizeLog2 entries */
    )
    {
    int    hashTblSizeLog2;
    int    i, tid;

    if (sizeLog2 < 1)
        return (ERROR);

    if (distNodeLibInstalled == TRUE)
        return (OK);

    for (i = 0; i < DIST_NODE_MAX_HOOKS; i++)
        {
        distNodeOperationalHook[i] = NULL;
        distNodeCrashedHook[i] = NULL;
        }

    distNodeLocalState = DIST_NODE_STATE_BOOT;

    /* Initialize the node database. */

    if (hashLibInit () == ERROR)
        return (ERROR);    /* hashLibInit() failed */

    hashTblSizeLog2 = sizeLog2 - 1;
    distNodeDbId = hashTblCreate (hashTblSizeLog2, distNodeHCmp,
            distNodeHFunc, KEY_ARG);
    if (distNodeDbId == NULL)
        return (ERROR);    /* hashTblCreate() failed */

#ifdef UNDEFINED
    distNodeDbSem = semBCreate (SEM_Q_PRIORITY, SEM_FULL) ; 
#else
    distNodeDbSem = semMCreate (SEM_Q_PRIORITY + SEM_INVERSION_SAFE) ; 
#endif

    /* Get the node database manager running. */

    tid = taskSpawn ("tDiNodeMgr",
                     DIST_NODE_MGR_PRIO,
                     VX_SUPERVISOR_MODE | VX_UNBREAKABLE,
                     DIST_NODE_MGR_STACK_SZ,
                     (FUNCPTR) distNodeDBTimerTask,
                     0, 0, 0, 0, 0, 0, 0, 0, 0, 0);
    if (tid == ERROR)
        return (ERROR);    /* taskSpawn() failed */

    distNodeGodfatherId = DIST_IF_BROADCAST_ADDR;

    /*
     * Add a virtual broadcast node.
     * State information (sliding windows, etc.) for outgoing broadcasts
     * must be handled at a centralized place. This is the virtual
     * broadcast node.
     * State information for incoming broadcasts is managed within
     * the broadcast communication substructure for the corresponding
     * node. When a note enters the incorporation phase, all other
     * nodes in the system get aware of it and update their node
     * databases.
     */

    if (distNodeCreate (DIST_IF_BROADCAST_ADDR, DIST_NODE_STATE_OPERATIONAL)
         ==
        NULL)
        {
        return (ERROR);
        }
        
    distNodeLibInstalled = TRUE;

    return (OK);
    }

/***************************************************************************
*
* distNodeHCmp - compare function for hashing in node DB (VxFusion option)
*
* This routine is the hash compare function for node IDs.
*
* AVAILABILITY
* This routine is distributed as a component of the unbundled distributed
* message queues option, VxFusion.
*
* RETURNS: True, if node IDs match.
* NOMANUAL
*/

LOCAL BOOL distNodeHCmp
    (
    DIST_NODE_DB_NODE * pMatchNode, /* first node */
    DIST_NODE_DB_NODE * pHNode,     /* second node */
    int                 keyArg      /* unused arg */
    )
    {
    DIST_NODE_ID  distNodeId1 = pMatchNode->nodeId;
    DIST_NODE_ID  distNodeId2 = pHNode->nodeId;
    
    UNUSED_ARG(keyArg);

    return (distNodeId1 == distNodeId2);
    }

/***************************************************************************
*
* distNodeHFunc - hash function for node DB (VxFusion option)
*
* This is the hash function for node IDs.
*
* AVAILABILITY
* This routine is distributed as a component of the unbundled distributed
* message queues option, VxFusion.
*
* RETURNS: A hash index.
* NOMANUAL
*/

LOCAL BOOL distNodeHFunc
    (
    int                 elements,  /* size of hash table */
    DIST_NODE_DB_NODE * pHNode,    /* node whose ID to hash */
    int                 divisor    /* used by hash computation */
    )
    {
    return ((pHNode->nodeId % divisor) & (elements - 1));
    }

/***************************************************************************
*
* distNodeCtl - control function for node database (VxFusion option)
*
* This routine performs a control function on the node database.
* The following functions are accepted:
* \is
* \i DIST_CTL_RETRY_TIMEOUT
* Set send-timeout in ticks to start with. When no ACK is received
* within the timeout, the packet is resent. The timeout for the
* <n>th sending is: <n> * DIST_CTL_RETRY_TIMEOUT .
* \i DIST_CTL_MAX_RETRIES
* Set a limit for number of retries, when sending fails.
* \i DIST_CTL_GET_LOCAL_ID
* Get local node id.
* \i DIST_CTL_GET_LOCAL_STATE
* Get state of local node.
* \i DIST_CTL_NACK_SUPPORT
* Negative acknowledges (NACKs) are used for requesting a resend of a single
* missing fragment from a packet. NACKs are sent immediately after a
* fragment is found to be missing.
* If <arg> is FALSE (0), the sending of negative acknowledges is disabled.
* If <arg> is TRUE (1), sending of NACKs is enabled. This is the default.
* \i DIST_CTL_PGGYBAK_UNICST_SUPPORT
* If this is enabled, the system waits a version dependent time until it
* sends an acknowledge for a previously received packet.
* If a data packet is sent to
* the acknowledge awaiting host in meantime, the acknowlege is delivered
* in that packet. This switch turns on/off piggy backing for unicast
* communication only.
* If <arg> is FALSE (0), piggy backing is disabled.
* If <arg> is TRUE (1), piggy backing is enabled.
* Piggy backing is enabled for unicast communication by default.
* \i DIST_CTL_PGGYBAK_BRDCST_SUPPORT
* If this is enabled, the system waits
* a version dependent time until it sends an
* acknowledge for a previously received packet. If a data packet is sent to
* the acknowledge awaiting host in meantime, the acknowlege is delivered
* in that packet. This switch turns on/off piggy backing for broadcast
* communication only.
* If <arg> is FALSE (0), piggy backing is disabled.
* If <arg> is TRUE (1), piggy backing is enabled.
* Piggy backing is disabled for broadcast communication by default.
* \i DIST_CTL_OPERATIONAL_HOOK
* Set a function to be called, each time a node shifts to operational state.
* \i DIST_CTL_CRASHED_HOOK
* Set a function to be called, each time a node shifts to crashed state.
* \ie
*
* The prototype of the function hooked to the database, should look
* like this:
* \cs
*    void fnc (DIST_NODE_ID nodeStateChanged);
* \ce
* 
* RETURNS: OK, the value requested, or ERROR if <function> is unknown.
*
* ERRNO:
* S_distLib_UNKNOWN_REQUEST
*
* AVAILABILITY
* This routine is distributed as a component of the unbundled distributed
* message queues option, VxFusion.
* NOMANUAL
*/

int distNodeCtl
    (
    int function,        /* function code      */
    int arg              /* arbitrary argument */
    )
    {
    int i;

    switch (function)
        {
        case DIST_CTL_MAX_RETRIES:
            if (arg < 0)
                return (ERROR);
            distNodeMaxRetries = arg;
            return (OK);

        case DIST_CTL_RETRY_TIMEOUT:
            if (arg < 0)
                return (ERROR);
            /*
             * <arg> is in ticks;
             * <distNodeRetryTimeout> is in manager wakeups;
             */

            distNodeRetryTimeout = (1000 * arg / sysClkRateGet() +
                    DIST_NODE_MGR_WAKEUP_MSEC - 1) / DIST_NODE_MGR_WAKEUP_MSEC;

            return (OK);

        case DIST_CTL_OPERATIONAL_HOOK:
            {
            for (i = 0; i < DIST_NODE_MAX_HOOKS; i++)
                {
                if (distNodeOperationalHook[i] == NULL)
                    {
                    distNodeOperationalHook[i] = (FUNCPTR) arg;
                    return (OK);
                    }