distgaplib.c

VXWORKS源代码

/* distGapLib.c - distributed group agreement protocol library */

/* Copyright 1999 Wind River Systems, Inc. */
#include "copyright_wrs.h"

/*
modification history
--------------------
01i,24may99,drm  added vxfusion prefix to VxFusion related includes
01h,11feb99,drm  fixed endian bugs in GAP protocol - added htons() / ntohs()
01g,11sep98,drm  changed #include "distStatLib.h" to "private/distLibP.h"
01f,20may98,drm  removed local variable distNodeLclState which was no longer
                 being used.
01e,13may98,ur   review of some comments
01d,15apr98,ur   replaced misleading comments
01c,13mar98,ur   fixed crash detection
01b,07jan98,ur   fixed "group id already in use"
01a,15jul97,ur   written
*/

/*
DESCRIPTION
This library contains the "group agreement protocol" (GAP).
*/

#include "vxWorks.h"
#if defined (DIST_GAP_REPORT) || defined (DIST_DIAGNOSTIC)
#include "stdio.h"
#endif
#include "stdlib.h"
#include "string.h"
#include "sllLib.h"
#include "dllLib.h"
#include "msgQLib.h"
#include "taskLib.h"
#include "netinet/in.h"
#include "private/semLibP.h"
#include "vxfusion/msgQDistGrpLib.h"
#include "vxfusion/distIfLib.h"
#include "vxfusion/private/distLibP.h"
#include "vxfusion/private/distNameLibP.h"
#include "vxfusion/private/msgQDistGrpLibP.h"
#include "vxfusion/private/distNetLibP.h"
#include "vxfusion/private/distNodeLibP.h"

LOCAL DL_LIST			gapInProgress;
LOCAL SEMAPHORE			distGapSemaphore;

/* local prototypes */

LOCAL void			distGapCatchCrashed (DIST_NODE_ID nodeId);
LOCAL void			distGapPh1Done (DIST_GRP_DB_NODE *pDistGrp);
LOCAL STATUS		distGapTry (DIST_NODE_ID distNodeId,
						DIST_GRP_DB_NODE *pDbNode);
LOCAL BOOL			distGapOutstandConc (DIST_NODE_DB_NODE *pDistNodeDbNode,
						SL_LIST *pOutstand);
LOCAL BOOL			distGapOutstandOk (DIST_GAP_NODE *pDistGapNode,
						DIST_NODE_ID distNodeIdOk);
LOCAL void			distGapOutstandDel (DIST_GAP_NODE *pDistGapNode,
						DIST_GAP_RESPONSE *pDistGapResponse,
						DIST_GAP_RESPONSE *pDistGapResponsePrev);
LOCAL DIST_STATUS	distGapInput (DIST_NODE_ID nodeIdSrc,
						DIST_TBUF_HDR *pTBufHdr);
#ifdef DIST_GAP_TASK
LOCAL void			distGapTimerTask (void);
LOCAL void			distGapTimer (void);
#endif

/*******************************************************************************
*
* distGapLibInit - 
*
* NOMANUAL
*/

STATUS distGapLibInit()
	{
#ifdef DIST_GAP_TASK
	int tid;
#endif

	dllInit (&gapInProgress);

#ifdef DIST_GAP_TASK
	tid = taskSpawn ("tDistGAP",
			DIST_GAP_MGR_PRIO,
			VX_SUPERVISOR_MODE | VX_UNBREAKABLE,
			DIST_GAP_MGR_STACK_SZ,
			(FUNCPTR) distGapTimerTask,
			0, 0, 0, 0, 0, 0, 0, 0, 0, 0);
			
	if (tid == ERROR)
		return (ERROR);
#endif

	distCtl (DIST_CTL_CRASHED_HOOK, (int) distGapCatchCrashed);

	/*
	 * Add GAP input function to table of services.
	 */
	return (distNetServAdd (DIST_PKT_TYPE_GAP, distGapInput,
			DIST_GAP_SERV_NAME, DIST_GAP_SERV_NET_PRIO,
			DIST_GAP_SERV_TASK_PRIO, DIST_GAP_SERV_TASK_STACK_SZ));
	}

/*******************************************************************************
*
* distGapCatchCrashed - get notified by a node crash
*
* NOMANUAL
*/

LOCAL void distGapCatchCrashed
	(
	DIST_NODE_ID crashedId
	)
	{
	DIST_GRP_DB_NODE	*pGrpDone;
	DIST_GAP_NODE		*pGapNode;
	BOOL				more;

#ifdef DIST_GAP_REPORT
	printf ("distGapCatchCrashed: node %ld crashed\n", crashedId);
#endif
	pGapNode = (DIST_GAP_NODE *) DLL_FIRST (&gapInProgress);

	while (pGapNode != NULL)
		{
		msgQDistGrpDbLock();

		pGrpDone = pGapNode->pGapGrp;

		distGapLock();
		more = distGapOutstandOk (pGapNode, crashedId);
		distGapUnlock();

		pGapNode = (DIST_GAP_NODE *) DLL_NEXT (pGapNode);

		if (more == FALSE)	/* this was the last node we waited for */
			{
			msgQDistGrpLclSetState (pGrpDone, DIST_GRP_STATE_GLOBAL);
			msgQDistGrpLclSetCreator (pGrpDone, distNodeLocalGetId());

			msgQDistGrpDbUnlock();

			distGapPh1Done (pGrpDone);
			}
		else
			msgQDistGrpDbUnlock();
		}
	}

/*******************************************************************************
*
* distGapOutstandLink - link all operational nodes to GAP structure
*
* NOMANUAL
*/

void distGapOutstandLink
	(
	DIST_GAP_NODE	*distGapNode
	)
	{
	sllInit (&distGapNode->gapOutstand);
	distNodeEach (distGapOutstandConc, (int) &distGapNode->gapOutstand);
	}

/*******************************************************************************
*
* distGapOutstandConc - called by distGapOutstandLink()
*
* NOMANUAL
*/

LOCAL BOOL distGapOutstandConc
	(
	DIST_NODE_DB_NODE	*pDistNodeDbNode,
	SL_LIST				*pOutstand
	)
	{
	DIST_GAP_RESPONSE	*pDistGapResp;

	if (pDistNodeDbNode->nodeState == DIST_NODE_STATE_OPERATIONAL &&
		pDistNodeDbNode->nodeId != DIST_IF_BROADCAST_ADDR)
		{
		pDistGapResp =
				(DIST_GAP_RESPONSE *) malloc (sizeof (DIST_GAP_RESPONSE));
		if (pDistGapResp == NULL)
			return (ERROR);

		pDistGapResp->pGapResponseNode = pDistNodeDbNode;

		sllPutAtHead (pOutstand, (SL_NODE *) pDistGapResp);
		}

	return (TRUE);	/* continue */
	}

/*******************************************************************************
*
* distGapOutstandUnlink - Free all nodes linked to the GAP structure
*
* NOMANUAL
*/

void distGapOutstandUnlink
	(
	DIST_GAP_NODE	*distGapNode
	)
	{
	DIST_GAP_RESPONSE	*pDistGapResponse;
	DIST_GAP_RESPONSE	*pDistGapResponsePrev = NULL;

	if (distGapNode == NULL)
		return;

	pDistGapResponse = (DIST_GAP_RESPONSE *)
			SLL_FIRST (&distGapNode->gapOutstand);

	while (pDistGapResponse != NULL)
		{
		sllRemove (&distGapNode->gapOutstand, (SL_NODE *) pDistGapResponse,
				(SL_NODE *) pDistGapResponsePrev);
		pDistGapResponsePrev = pDistGapResponse;
		pDistGapResponse = (DIST_GAP_RESPONSE *) SLL_NEXT (pDistGapResponse);
		free (pDistGapResponsePrev);
		}

	}

/*******************************************************************************
*
* distGapOutstandOk - Delete a single node from the GAP structure's node list
*
* RETURNS: TRUE if there are more outstanding responses, or FALSE if the
* job is done.
*
* NOMANUAL
*/

LOCAL BOOL distGapOutstandOk
	(
	DIST_GAP_NODE	*pDistGapNode,
	DIST_NODE_ID	distNodeIdOk
	)
	{
	DIST_GAP_RESPONSE	*pDistGapResp;
	DIST_GAP_RESPONSE	*pDistGapRespPrev;

	pDistGapResp = (DIST_GAP_RESPONSE *)
			SLL_FIRST (&pDistGapNode->gapOutstand);
	pDistGapRespPrev = NULL;

	while (pDistGapResp != NULL &&
		   pDistGapResp->pGapResponseNode->nodeId != distNodeIdOk)
		{
		pDistGapRespPrev = pDistGapResp;
		pDistGapResp = (DIST_GAP_RESPONSE *) SLL_NEXT (pDistGapResp);
		}

	if (pDistGapResp != NULL)
		{
		distGapOutstandDel (pDistGapNode, pDistGapResp, pDistGapRespPrev);
		if (SLL_FIRST (&pDistGapNode->gapOutstand) == NULL)
			return (FALSE);
		}

	return (TRUE);
	}

/*******************************************************************************
*
* distGapOutstandDel - Delete a single node 
*
* NOMANUAL
*/

LOCAL void distGapOutstandDel
	(
	DIST_GAP_NODE		*pDistGapNode,
	DIST_GAP_RESPONSE	*pDistGapResponse,
	DIST_GAP_RESPONSE	*pDistGapResponsePrev
	)
	{
	sllRemove ((SL_LIST *) &pDistGapNode->gapOutstand,
			(SL_NODE *) pDistGapResponse, (SL_NODE *) pDistGapResponsePrev);

	free (pDistGapResponse);

#ifdef DIST_GAP_REPORT
	{
	SL_NODE	*pNode = SLL_FIRST (&pDistGapNode->gapOutstand);
	int		i;

	for (i=0; pNode; i++)
		pNode = SLL_NEXT (pNode);
	printf ("distGapOutstandDel: %d nodes outstanding\n", i);
	}
#endif
	}

/*******************************************************************************
*
* distGapPh1Done - phase 1 is done
*
* NOMANUAL
*/

LOCAL void distGapPh1Done
	(
	DIST_GRP_DB_NODE	*pDistGrp
	)
	{
	/*
	 * List of outstanding responses is empty now. We are done.
	 * Send SET telegram and wakeup sleeping task.
	 */

	DIST_PKT_GAP_SET	distPktGapSet;

#ifdef DIST_GAP_REPORT
	printf ("distGapInput/OK: GAP phase 1 done.\n");
#endif
	/*
	 * Build SET telegram and broadcast it.
	 */
	distPktGapSet.setHdr.pktType = DIST_PKT_TYPE_GAP;
	distPktGapSet.setHdr.pktSubType = DIST_PKT_TYPE_GAP_SET;
	distPktGapSet.setId = htons (pDistGrp->grpDbId);
	strcpy ((char *) &distPktGapSet.setName,
			(char *) &pDistGrp->grpDbName);

#ifdef DIST_GAP_REPORT
	printf ("distGapInput/OK: GAP phase 2 -- broadcast SET `%s'/%d.\n",
			distPktGapSet.setName, distPktGapSet.setId);
#endif
	distNetSend (DIST_IF_BROADCAST_ADDR, (DIST_PKT *) &distPktGapSet,
			sizeof (distPktGapSet), WAIT_FOREVER, DIST_GAP_PRIO);

	/*
	 * Wakeup initiator.
	 */
	semGive (&pDistGrp->pGrpDbGapNode->gapWaitFor);
	}

/*******************************************************************************
*
* distGapNodeInit - Initialize a GAP node
*
* NOMANUAL
*/

void distGapNodeInit
	(
	DIST_GAP_NODE		*pDistGapNode,
	DIST_GRP_DB_NODE	*pDistGrpDbNode,
	BOOL				link
	)
	{
	if (link)
		distGapOutstandLink(pDistGapNode);

	semBInit (&(pDistGapNode->gapWaitFor), SEM_Q_FIFO, SEM_EMPTY);
	pDistGapNode->pGapGrp		= pDistGrpDbNode;
	pDistGapNode->gapTimeout	= DIST_GAP_TRY_TIMO;
	pDistGapNode->gapRetries	= 1;

	distGapLock();
	dllInsert (&gapInProgress, NULL, &(pDistGapNode->gapLink));
	distGapUnlock();
	}

/*******************************************************************************
*
* distGapNodeDelete - Delete a GAP node
*
* NOMANUAL
*/

void distGapNodeDelete
	(
	DIST_GAP_NODE	*pDistGapNode
	)
	{
	distGapLock();
	dllRemove (&gapInProgress, (DL_NODE *) &(pDistGapNode->gapLink));
	distGapUnlock();
	}

/*******************************************************************************
*
* distGapStart - Start GAP (group agreement protocol)
*
* Do a TRY in order to create a new group.
*
* NOMANUAL
*/

DIST_MSG_Q_GRP_ID distGapStart
	(
	DIST_GRP_DB_NODE	*pDistGrpDbNode
	)
	{
	DIST_GAP_NODE		distGapNode;
	DIST_MSG_Q_GRP_ID	uniqGrpId;

	if (distNodeGetNumNodes(DIST_NODE_NUM_NODES_ALIVE) == 1)
		{
		/* we are alone */
		msgQDistGrpLclSetState (pDistGrpDbNode, DIST_GRP_STATE_GLOBAL);
		msgQDistGrpLclSetCreator (pDistGrpDbNode, distNodeLocalGetId());
		return (pDistGrpDbNode->grpDbId);
		}

	/*
	 * Init GAP Node and link it to the group database node and
	 * the list of active GAPs.
	 * GAP initialization includes the linking of all nodes,
	 * currently known in the system. When a response comes in,
	 * the linked list is updated (the node is removed from the
	 * list).
	 */

	distGapNodeInit(&distGapNode, pDistGrpDbNode, TRUE);
	pDistGrpDbNode->pGrpDbGapNode = &distGapNode;

	/*
	 * Send TRY.
	 */

	distGapTry (DIST_IF_BROADCAST_ADDR, pDistGrpDbNode);

	/*
	 * Wait on GAP Node.
	 */

	semTake (&distGapNode.gapWaitFor, WAIT_FOREVER);

	/*
	 * Cleanup
	 */

	if (pDistGrpDbNode->pGrpDbGapNode)
		{
		DIST_GAP_NODE	*pGapNode = pDistGrpDbNode->pGrpDbGapNode;

		pDistGrpDbNode->pGrpDbGapNode = NULL;
		distGapNodeDelete (pGapNode);
		}

#ifdef DIST_DIAGNOSTIC
	if (pDistGrpDbNode->grpDbState != DIST_GRP_STATE_GLOBAL)
		distLog ("distGapStart: group has no global state\n");
#endif

	/*
	 * Ask for group id.
	 */

	msgQDistGrpDbLock();
	uniqGrpId = msgQDistGrpLclGetId (pDistGrpDbNode);
	msgQDistGrpDbUnlock();

	return (uniqGrpId);
	}


/*******************************************************************************
*
* distGapTry - Send a TRY, for a specified group
*
* NOMANUAL
*/

LOCAL STATUS distGapTry
	(
	DIST_NODE_ID		distNodeId,		/* node address */
	DIST_GRP_DB_NODE	*pDistGrpDbNode	/* group to try */
	)
	{
	DIST_PKT_GAP_TRY	distPktGapTry;
	STATUS				status;

	/*
	 * Build TRY telegram and send it.
	 */