msgqdistgrplib.c

VXWORKS源代码

/* msgQDistGrpLib.c - distributed message queue group library (VxFusion option) */

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

/*
modification history
--------------------
01q,23oct01,jws  fix compiler warnings (SPR 71117); fix man pages (SPR 71239)
01p,24jul01,r_s  changed code to be ANSI compatible so that it compiles with
                 diab. made asm macro changes for diab
01o,09jun99,drm  Allowing check for empty groups.
01n,09jun99,drm  Adding some htonl() / ntohl()
01m,09jun99,drm  Adding code to return S_msgQDistLib_OVERALL_TIMEOUT
01l,01jun99,drm  Adding check to make sure that only distributed message
                 queues (and not groups) can be added to a group.
01k,24may99,drm  added vxfusion prefix to VxFusion related includes
01j,23feb99,wlf  doc edits
01i,18feb99,wlf  doc cleanup
01h,29oct98,drm  documentation modifications
01g,09oct98,drm  fixed a bug reported by Oce
01f,13may98,ur   cleanup when msgQDistGrpInit() fails
                 locking on group send inquiries
01e,08may98,ur   removed 8 bit node id restriction
01d,08apr98,ur   optional enhancement CHECK_FOR_EMPTY_GROUP--untested
01c,20mar98,ur   set errno when database is full
01b,01jul97,ur   tested, ok.
01a,11jun97,ur   written.
*/

/*
DESCRIPTION
This library provides the grouping facility for distributed message queues.  
Single distributed message queues can join one or more groups.  A message
sent to a group is sent to all message queues that are members of that
group.  A group, however, is prohibited from sending messages.  Also, it is
an error to call msgQDistNumMsgs() with a distributed message queue group ID.

Groups are created with symbolic names and identified by a unique ID,
 MSG_Q_ID, as with normal message queues.

If the group is new to the distributed system, the group agreement 
protocol (GAP) is employed to determine a globally unique identifier.  
As part of the protocol's negotiation, all group databases throughout 
the system are updated.

The distributed message queue group library is initialized by calling
distInit().

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

INCLUDE FILES: msgQDistGrpLib.h

SEE ALSO: distLib, msgQDistGrpShow
*/

#include "vxWorks.h"
#if defined (MSG_Q_DIST_GRP_REPORT) || defined (DIST_DIAGNOSTIC)
#include "stdio.h"
#endif
#include "string.h"
#include "stdlib.h"
#include "taskLib.h"
#include "hashLib.h"
#include "sllLib.h"
#include "errnoLib.h"
#include "msgQLib.h"
#include "netinet/in.h"
#include "vxfusion/msgQDistLib.h"
#include "vxfusion/msgQDistGrpLib.h"
#include "vxfusion/distIfLib.h"
#include "vxfusion/distStatLib.h"
#include "vxfusion/private/msgQDistGrpLibP.h"
#include "vxfusion/private/distNetLibP.h"
#include "vxfusion/private/distNodeLibP.h"
#include "vxfusion/private/distObjLibP.h"
#include "vxfusion/private/distLibP.h"

/* defines */

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

#define KEY_ARG_STR        13
#define KEY_CMP_ARG_STR    0       /* not used */

#define KEY_ARG_ID        65537
#define KEY_CMP_ARG_ID    0        /* not used */

/* global data */

SEMAPHORE                    distGrpDbSemaphore;
DIST_MSG_Q_GRP_ID            distGrpIdNext = 0;

/* local data */

LOCAL HASH_ID                distGrpDbNmId = NULL;
LOCAL DIST_GRP_HASH_NODE *   distGrpDbNm = NULL;

LOCAL HASH_ID                distGrpDbIdId = NULL;
LOCAL DIST_GRP_HASH_NODE *   distGrpDbId = NULL;

LOCAL DIST_GRP_DB_NODE *     distGrpDb = NULL;
LOCAL SL_LIST                msgQDistGrpFreeList;

LOCAL BOOL                   msgQDistGrpLibInstalled = FALSE;

/* local prototypes */

LOCAL BOOL          msgQDistGrpHCmpStr (DIST_GRP_HASH_NODE *pMatchNode,
                                        DIST_GRP_HASH_NODE *pHNode,
                                        int keyArg);
LOCAL INT32         msgQDistGrpHFuncStr (int elements,
                                         DIST_GRP_HASH_NODE *pHNode,
                                         int keyArg);
LOCAL BOOL          msgQDistGrpHCmpId (DIST_GRP_HASH_NODE *pMatchNode,
                                       DIST_GRP_HASH_NODE *pHNode,
                                       int keyArg);
LOCAL INT32         msgQDistGrpHFuncId (int elements,
                                        DIST_GRP_HASH_NODE *pHNode,
                                        int keyArg);
LOCAL DIST_STATUS    msgQDistGdbInput (DIST_NODE_ID nodeIdSrc,
                                       DIST_TBUF_HDR *pTBufHdr);
LOCAL DIST_STATUS    msgQDistGrpInput (DIST_NODE_ID nodeIdSrc,
                                       DIST_TBUF_HDR *pTBufHdr);
LOCAL STATUS         msgQDistGrpLclSend (DIST_MSG_Q_GRP_SEND_INQ *pInq,
                                         DIST_MSG_Q_GRP_ID distMsgQGrpId,
                                         char *buffer,
                                         UINT nBytes, int timeout,
                                         int priority);
LOCAL void           msgQDistGrpLclSendCanWait (DIST_INQ_ID inqId,
                                                MSG_Q_ID msgQId,
                                                char *buffer, UINT nBytes,
                                                int timeout, int priority);
LOCAL STATUS         msgQDistGrpAgent (DIST_NODE_ID nodeIdSender,
                                       DIST_INQ_ID inqIdSender,
                                       DIST_MSG_Q_GRP_ID distMsgQGrpId,
                                       char *buffer,
                                       UINT nBytes, int timeout,
                                       int priority);
LOCAL STATUS         msgQDistGrpSendStatus (DIST_NODE_ID nodeIdDest,
                                            DIST_INQ_ID inqId,
                                            DIST_MSG_Q_STATUS msgQStatus);
LOCAL BOOL           msgQDistGrpBurstOne (DIST_GRP_HASH_NODE *pNode,
                                          DIST_GRP_BURST *pBurst);

/***************************************************************************
*
* msgQDistGrpLibInit - initialize the distributed message queue group package (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 msgQDistGrpLibInit (void)
    {
    }

/***************************************************************************
*
* msgQDistGrpInit - initialize the group database (VxFusion option)
*
* This routine initializes the group database.
*
* AVAILABILITY
* This routine is distributed as a component of the unbundled distributed
* message queues option, VxFusion.
*
* RETURNS: OK, if successful; ERROR, if unsucessful.
*
* NOMANUAL
*/

STATUS msgQDistGrpInit
    (
    int               sizeLog2   /* init 2^^sizeLog2 entries */
    )
    {
    DIST_GRP_DB_NODE * pDbNode;
    int                hashTblSizeLog2;
    int                grpHNBytes;
    int                grpDbNBytes;
    int                grpDbSize;
    int                ix;
    STATUS             grpDbServAddStatus;
    STATUS             msgQGrpServAddStatus;

    if (sizeLog2 < 1)
        return (ERROR);

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

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

    if (distInqInit (DIST_INQ_HASH_TBL_SZ_LOG2) == ERROR)
        return (ERROR);

    hashTblSizeLog2 = sizeLog2 - 1;

    distGrpDbNmId = hashTblCreate (hashTblSizeLog2, msgQDistGrpHCmpStr,
                                   msgQDistGrpHFuncStr, KEY_ARG_STR);
    distGrpDbIdId = hashTblCreate (hashTblSizeLog2, msgQDistGrpHCmpId,
                                   msgQDistGrpHFuncId, KEY_ARG_ID);

    if (distGrpDbNmId && distGrpDbIdId)
        {
        grpDbSize = 1 << sizeLog2;
        grpHNBytes = grpDbSize * sizeof (DIST_GRP_HASH_NODE);
        grpDbNBytes = grpDbSize * sizeof (DIST_GRP_DB_NODE);

        distGrpDbNm = (DIST_GRP_HASH_NODE *) malloc (grpHNBytes);
        distGrpDbId = (DIST_GRP_HASH_NODE *) malloc (grpHNBytes);
        distGrpDb = (DIST_GRP_DB_NODE *) malloc (grpDbNBytes);

        if (distGrpDbNm && distGrpDbId && distGrpDb)
            {
            sllInit (&msgQDistGrpFreeList);

            for (ix = 0; ix < grpDbSize; ix++)
                {
                pDbNode = &distGrpDb[ix];
                pDbNode->ixNode = ix;
                sllPutAtHead (&msgQDistGrpFreeList, (SL_NODE *) pDbNode);
                distGrpDbNm[ix].pDbNode = pDbNode;
                distGrpDbId[ix].pDbNode = pDbNode;
                }

            msgQDistGrpDbLockInit();

            /* Add distributed group database service to table of services. */

            grpDbServAddStatus = distNetServAdd (
                    DIST_PKT_TYPE_DGDB,
                    msgQDistGdbInput,
                    DIST_DGDB_SERV_NAME,
                    DIST_DGDB_SERV_NET_PRIO,
                    DIST_DGDB_SERV_TASK_PRIO,
                    DIST_DGDB_SERV_TASK_STACK_SZ);

            /* Add message group service to table of services. */
            
            msgQGrpServAddStatus =
                distNetServAdd ( DIST_PKT_TYPE_MSG_Q_GRP,
                                 msgQDistGrpInput,
                                 DIST_MSG_Q_GRP_SERV_NAME,
                                 DIST_MSG_Q_GRP_SERV_NET_PRIO,
                                 DIST_MSG_Q_GRP_SERV_TASK_PRIO,
                                 DIST_MSG_Q_GRP_SERV_TASK_STACK_SZ);

            if (grpDbServAddStatus != ERROR && msgQGrpServAddStatus != ERROR)
                {
                msgQDistGrpLibInstalled = TRUE;
                return (OK);
                }
            }
        }

        /* cleanup, when error */

        if (distGrpDbNmId)    hashTblDelete (distGrpDbNmId);
        if (distGrpDbIdId)    hashTblDelete (distGrpDbIdId);
        if (distGrpDbNm)      free (distGrpDbNm);
        if (distGrpDbId)      free (distGrpDbId);
        if (distGrpDb)        free (distGrpDb);

        return (ERROR);
    }

/***************************************************************************
*
* msgQDistGrpAdd - add a distributed message queue to a group (VxFusion option)
*
* This routine adds the queue identified by the argument <msgQId> to a group
* with the ASCII name specified by the argument <distGrpName>. 
*
* Multicasting is based on distributed message queue groups.  If the group 
* does not exist, one is created.  Any number of message queues from different 
* nodes can be bound to a single group. In addition, a message queue can
* be added into any number of groups; msgQDistGrpAdd() must be called for each
* group of which the message queue is to be a member.
*
* The <options> parameter is presently unused and must be set to 0.
*
* This routine returns a message queue ID, MSG_Q_ID, that can be used directly 
* by msgQDistSend() or by the generic msgQSend() routine.  Do not call the
* msgQReceive() or msgQNumMsgs() routines or their distributed counterparts,
* msgQDistReceive() and msgQDistNumMsgs(), with a group message queue ID.
*
* As with msgQDistCreate(), use distNameAdd() to add the group message 
* queue ID returned by this routine to the distributed name database so 
* that the ID can be used by tasks on other nodes.
*
* AVAILABILITY
* This routine is distributed as a component of the unbundled distributed
* message queues option, VxFusion.
*
* RETURNS: MSG_Q_ID, or NULL if there is an error.
*
* ERRNO:
* \is
* \i S_msgQDistGrpLib_NAME_TOO_LONG
* The name of the group is too long.
* \i S_msgQDistGrpLib_INVALID_OPTION
* The <options> parameter is invalid.
* \i S_msgQDistGrpLib_DATABASE_FULL
* The group database is full.
* \i S_distLib_OBJ_ID_ERROR
* The <msgQId> parameter is not a distributed message queue.
* \ie
*
* SEE ALSO: msgQLib, msgQDistLib, distNameLib
* 
* INTERNAL NOTE: Takes <distGrpDbSemaphore>.
*/

MSG_Q_ID msgQDistGrpAdd
    (
    char *            distGrpName,  /* new or existing group name           */
    MSG_Q_ID          msgQId,       /* message queue to add to the group    */
    DIST_GRP_OPT      options       /* group message queue options - UNUSED */
    )
    {
    DIST_GRP_DB_NODE * pDistGrpDbNode;
    DIST_MSG_Q_ID      dMsgQId;      /* distributed message queue ID */
    DIST_OBJ_NODE * pObjNode;     /* ptr to object containing real ID */ 
    DIST_GAP_NODE   dGapNode;
    DIST_GAP_NODE * pDGapNodeTemp;

    /*
     * Check the parameters:  
     * - msgQId must be a distributed message queue. It cannot be a
     *   a standard message queue or a group.
     * - the group name cannot exceed DIST_NAME_MAX_LENGTH
     * - options must be 0
     *
     * If any of the parameters are invalid, set errno and return
     * NULL to indicate failure.
     */

    if (!ID_IS_DISTRIBUTED (msgQId)) /* not a distributed message queue */
        {
        errnoSet (S_distLib_OBJ_ID_ERROR);
        return NULL;
        }
    else                             /* is distributed msgQ */
        {

        pObjNode = MSG_Q_ID_TO_DIST_OBJ_NODE (msgQId);
        if (!IS_DIST_MSG_Q_OBJ (pObjNode))
            {
            /* legal object ID, but not a message queue */

            errnoSet (S_distLib_OBJ_ID_ERROR);
            return NULL;    
            }
       
        dMsgQId = (DIST_MSG_Q_ID) pObjNode->objNodeId;
        if (IS_DIST_MSG_Q_TYPE_GRP(dMsgQId))
            {
            /* ID refers to a group, not a plain dist. message queue */

            errnoSet (S_distLib_OBJ_ID_ERROR);
            return NULL;
            }
        }

    if (strlen (distGrpName) > DIST_NAME_MAX_LENGTH)
        {
        errnoSet (S_msgQDistGrpLib_NAME_TOO_LONG);
        return (NULL);    /* name too long */
        }

    if (options != 0)
        {
        errnoSet (S_msgQDistGrpLib_INVALID_OPTION);
        return (NULL);    /* options parameter currently unused */
        }

    /* Lock the database and try to find the group. */

    msgQDistGrpDbLock();

    pDistGrpDbNode = msgQDistGrpLclFindByName (distGrpName);
    if (pDistGrpDbNode == NULL)
        {
#ifdef MSG_Q_DIST_GRP_REPORT
        DIST_MSG_Q_GRP_ID    grpId;
#endif
        /* Group is unknown by now, create it. */

#ifdef MSG_Q_DIST_GRP_REPORT
        printf ("msgQDistGrpAdd: group is unknown by now--create it\n");
#endif
        pDistGrpDbNode = msgQDistGrpLclCreate (distGrpName, distGrpIdNext++,
                                               DIST_GRP_STATE_LOCAL_TRY);

        msgQDistGrpDbUnlock();

        if (pDistGrpDbNode == NULL)
            {
#ifdef DIST_DIAGNOSTIC
            distLog ("msgQDistGrpAdd: failed to create new group\n");
#endif
            errnoSet (S_msgQDistGrpLib_DATABASE_FULL);
            return (NULL);
            }

        /*
         * Agree on a global unique identifier for the group and
         * add the fist group member.
         */

#ifdef MSG_Q_DIST_GRP_REPORT
        printf ("msgQDistGrpAdd: enter agreement phase, propose 0x%lx\n",
                (u_long) msgQDistGrpLclGetId (pDistGrpDbNode));
        grpId =
#endif
            msgQDistGrpAgree (pDistGrpDbNode);
#ifdef MSG_Q_DIST_GRP_REPORT
        printf ("msgQDistGrpAdd: agreed on 0x%lx\n", (u_long) grpId);
#endif
        msgQDistGrpLclAddMember (pDistGrpDbNode, msgQId);
        }
    else
        {
        /* Group already exists. Check the state. */

#ifdef MSG_Q_DIST_GRP_REPORT
        printf ("msgQDistGrpAdd: group already exists--add object\n");
#endif
        while (pDistGrpDbNode->grpDbState < DIST_GRP_STATE_GLOBAL)
            {
                /*
                 * Group already exists, but has a local state. This means,
                 * somebody else tries to install the group.
                 * We unlock the group database and wait for a go-ahead.