request.c

linux集群服务器软件代码包

/* $Id: request.c,v 1.12 2004/12/06 21:19:20 msoffen Exp $ */
/* 
 * request.c: 
 *
 * Copyright (C) 2003 Deng Pan <[e-mail]>
 * 
 * This program is free software; you can redistribute it and/or
 * modify it under the terms of the GNU General Public
 * License as published by the Free Software Foundation; either
 * version 2.1 of the License, or (at your option) any later version.
 * 
 * This software is distributed in the hope that it will be useful,
 * but WITHOUT ANY WARRANTY; without even the implied warranty of
 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
 * General Public License for more details.
 * 
 * You should have received a copy of the GNU General Public
 * License along with this library; if not, write to the Free Software
 * Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA  02111-1307  USA
 */

#ifdef HAVE_CONFIG_H
#include "config.h"
#endif
#include <stdio.h>
#include <stdlib.h>
#include <unistd.h>
#include <errno.h>
#include <string.h>
#include <sys/types.h>
#include <sys/stat.h>

#include <glib.h>

#include <clplumbing/cl_log.h>
#include <clplumbing/cl_signal.h>
#include <clplumbing/ipc.h>
#include <clplumbing/Gmain_timeout.h>
#include <clplumbing/realtime.h>
#include <hb_api_core.h>
#include <hb_api.h>
#include <ha_msg.h>
#include <heartbeat.h>

#include <saf/ais.h>
#include <checkpointd/clientrequest.h>
#include "checkpointd.h"
#include "client.h"
#include "replica.h"
#include "message.h"
#include "request.h"
#include "response.h"
#include "operation.h"
#include "utils.h"

#ifdef USE_DMALLOC
#include <dmalloc.h>
#endif

extern SaCkptServiceT* saCkptService;

/* Process client request */
gboolean
SaCkptRequestProcess(IPC_Channel* clientChannel)
{
	SaCkptRequestT 		*ckptReq = NULL;

	while (clientChannel->ops->is_message_pending(clientChannel) 
		== TRUE) {
		ckptReq = SaCkptRequestReceive(clientChannel);
		if (ckptReq != NULL) {
			SaCkptRequestStart(ckptReq);

			return TRUE;
		} else {
			return FALSE;
		}
	}

	return TRUE;

}

int 
SaCkptRequestStart(SaCkptRequestT* ckptReq) 
{
	SaCkptMessageT		*ckptMsg = NULL;
	SaCkptResponseT 	*ckptResp = NULL;
	
	SaCkptClientT		*client = NULL;
	SaCkptReplicaT		*replica = NULL;
	SaCkptOpenCheckpointT	*openCkpt = NULL;

	SaCkptReqInitParamT	*initParam = NULL;
	SaCkptReqOpenParamT	*openParam = NULL;
	SaCkptReqCloseParamT	*closeParam = NULL;
	SaCkptReqSecExpSetParamT	*secExpSetParam = NULL;
	SaCkptReqRtnParamT	*rtnParam = NULL;
/*	SaCkptReqUlnkParamT	*unlinkParam = NULL; */
	SaCkptReqSecQueryParamT	*secQueryParam = NULL;
	
	SaCkptSectionT		*section = NULL;
	SaCkptCheckpointStatusT *checkpointStatus = NULL;
	SaCkptCheckpointCreationAttributesT	*attr = NULL;
	const char *sectionName 	= NULL;
	char* strReq = NULL;
	SaCkptHandleT		clientHandle = 0;
	SaCkptCheckpointHandleT	checkpointHandle = 0;
	void			*reqParam = NULL;
	SaNameT			*unlinkName = NULL;
	

	SaTimeT			timeout = 0;
	
	SaCkptSectionDescriptorT	*sectionDescriptor = NULL;
	int 				secListPass = 0;
	unsigned int 				secListTotalSize = 0;
	int			sectNumber = 0;
	int			descNumber = 0;
	
	char			*p = NULL;
	int			sectSelected = 0;
	
	GList			*list = NULL;

	timeout = REQUEST_TIMEOUT * 1000LL * 1000LL * 1000LL;

	client = ckptReq->client;
	clientHandle = ckptReq->clientRequest->clientHandle;
	reqParam = ckptReq->clientRequest->reqParam;

	ckptResp = SaCkptResponseCreate(ckptReq);

	switch (ckptReq->clientRequest->req) {
	case REQ_SERVICE_INIT:
		initParam = reqParam;
	
		/* FIXME: different version should work together */
		if (SaCkptVersionCompare(initParam->ver,
			saCkptService->version) != 0) {
			ckptResp->resp->retVal = SA_ERR_VERSION;
			SaCkptResponseSend(&ckptResp);
			break;
		}

		if (clientHandle <= 0) {
			/* create new client */
			client = SaCkptClientCreate(initParam);
			client->channel[0] = ckptReq->clientChannel;

			ckptReq->client = client;
			ckptReq->clientRequest->clientHandle = 
				client->clientHandle;

			/* 
			 * initialize is a special operation
			 * the handle will be returned in the response header
			 *
			 * should be sync with the client library
			 */
			ckptResp->resp->clientHandle = client->clientHandle;
		} else {
			/* add channel for async operation */
			client = g_hash_table_lookup(
				saCkptService->clientHash,
				(gpointer)&clientHandle);
			if (client == NULL) {
				ckptResp->resp->retVal = 
					SA_ERR_FAILED_OPERATION;
			} else {
				client->channel[1] = 
					ckptReq->clientChannel;
			}
		}
		SaCkptResponseSend(&ckptResp);
		
		break;

	case REQ_SERVICE_FINL:
		if ((g_list_length(client->pendingRequestList) != 0) ||
			(g_hash_table_size(client->requestHash) != 0)){
			ckptResp->resp->retVal = SA_ERR_BUSY;
			SaCkptResponseSend(&ckptResp);
			break;
		}

		/* send response first */
		SaCkptResponseSend(&ckptResp);
		
		SaCkptClientDelete(&client);

		break;
		
	case REQ_CKPT_OPEN:
	case REQ_CKPT_OPEN_ASYNC:
		ckptReq->operation = OP_CKPT_OPEN;

		openParam = reqParam;

		unlinkName = g_hash_table_lookup(
			saCkptService->unlinkedCheckpointHash,
			(gconstpointer)(openParam->ckptName.value));
		if (unlinkName != NULL) {
			cl_log(LOG_ERR, 
				"checkpoint has already been unlinked");
			/* FIXME  No return value defined on SPEC A*/
			ckptResp->resp->retVal = 
				SA_ERR_INVALID_PARAM;
			SaCkptResponseSend(&ckptResp);
			break;
		}

		replica = g_hash_table_lookup(saCkptService->replicaHash, 
			(gconstpointer)(openParam->ckptName.value));
		
		/* if local replica exist */
		if (replica != NULL) { 
			/* if unlinked, return error */
			if (replica->flagUnlink == TRUE) {
				cl_log(LOG_ERR, 
				"checkpoint has already been unlinked");
				
				ckptResp->resp->retVal = 
					SA_ERR_INVALID_PARAM;
				SaCkptResponseSend(&ckptResp);
				break;
			}

			/* 
			 * if the createattribute is not null and is 
			 * different from the replica, return error
			 */
			attr = &(openParam->attr);
			if ((attr->checkpointSize != 
				replica->maxCheckpointSize) ||
			    (attr->creationFlags != 
				replica->createFlag) || 
			    (attr->maxSectionIdSize !=
				replica->maxSectionIDSize) || 
			    (attr->maxSections != 
				replica->maxSectionNumber) ||
			    (attr->maxSectionSize !=
				replica->maxSectionSize)) {
				cl_log(LOG_ERR, 
					"Attribute is different");
				
				ckptResp->resp->retVal = SA_ERR_EXIST;
				SaCkptResponseSend(&ckptResp);
				break;
			}
			
			/* if the retention timer started, stop it */
			if (replica->retentionTimeoutTag > 0) {
				SaCkptReplicaStopTimer(replica);
			}

			openCkpt = SaCkptCheckpointOpen(client, 
				replica, openParam);
			/* update the request */
			ckptReq->openCkpt = openCkpt;

			/* send response */
			ckptResp->resp->dataLength = 
				sizeof(SaCkptCheckpointHandleT);
			ckptResp->resp->data = SaCkptMalloc( 
				ckptResp->resp->dataLength);
			SACKPTASSERT(ckptResp->resp->data != NULL);
			memcpy(ckptResp->resp->data, 
				&(openCkpt->checkpointHandle), 
				ckptResp->resp->dataLength);
			
			SaCkptResponseSend(&ckptResp);
			break;

		} else { 
			/* 
			 * no local replica.
			 * add request into queue and broadcast message 
			 */
			g_hash_table_insert(client->requestHash,
				(gpointer)&(ckptReq->clientRequest->requestNO),
				(gpointer)ckptReq);
				
			initOpenReqNodeStatus(ckptReq->clientRequest);

			ckptMsg = SaCkptMessageCreateReq(ckptReq, 
				M_CKPT_OPEN_BCAST);
			strcpy(ckptMsg->checkpointName, openParam->ckptName.value);
			SaCkptMessageBroadcast(ckptMsg);
			SaCkptFree((void*)&ckptMsg);

			 if (openParam->timetout < timeout) {
			 	timeout = openParam->timetout;
			 }
			SaCkptRequestStartTimer(ckptReq, timeout);
		}

		break;

	case REQ_CKPT_CLOSE:
		closeParam = reqParam;
		
		openCkpt = g_hash_table_lookup(
			saCkptService->openCheckpointHash,
			(gpointer)&(closeParam->checkpointHandle));
		if (openCkpt == NULL) {
			ckptResp->resp->retVal = SA_ERR_BAD_HANDLE;
			SaCkptResponseSend(&ckptResp);
			break;
		}
		ckptReq->openCkpt = openCkpt;
		
		/* no local replica, send message to active replica */
		if (openCkpt->flagLocalReplica != TRUE) {
			strcpy(ckptReq->toNodeName, openCkpt->activeNodeName);
			
			/* add it to request queue first */
			g_hash_table_insert(client->requestHash,
				(gpointer)&(ckptReq->clientRequest->requestNO),
				(gpointer)ckptReq);

			ckptMsg = SaCkptMessageCreateReq(ckptReq, 
				M_CKPT_CLOSE_REMOTE);
			ckptMsg->data = SaCkptMalloc(
				sizeof(openCkpt->checkpointRemoteHandle));
			ckptMsg->dataLength = 
				sizeof(openCkpt->checkpointRemoteHandle);
			memcpy(ckptMsg->data, 
				&(openCkpt->checkpointRemoteHandle),
				sizeof(openCkpt->checkpointRemoteHandle));
			SaCkptMessageSend(ckptMsg, openCkpt->activeNodeName);
			SaCkptFree((void*)&ckptMsg);
			
			SaCkptRequestStartTimer(ckptReq, timeout);
			 
			break;
		}

		/* local replica exist, close it */
		ckptResp->resp->retVal = SaCkptCheckpointClose(&openCkpt);
		SaCkptResponseSend(&ckptResp);
		
		break;

	case REQ_CKPT_ACT_SET:
		ckptReq->operation = OP_CKPT_ACT_SET;

		/* the first field of reqParam is the handle */
		checkpointHandle= *(SaCkptCheckpointHandleT*)reqParam;

		openCkpt = g_hash_table_lookup(
			saCkptService->openCheckpointHash,
			(gpointer)&(checkpointHandle));
		if (openCkpt == NULL) {
			ckptResp->resp->retVal = SA_ERR_BAD_HANDLE;
			SaCkptResponseSend(&ckptResp);
			break;
		}
		ckptReq->openCkpt = openCkpt;

		replica = openCkpt->replica;
		if (replica == NULL) {
			/* no local replica */
			ckptResp->resp->retVal = SA_ERR_FAILED_OPERATION;
			SaCkptResponseSend(&ckptResp);
			break;
		}
		
		if (replica->flagIsActive == TRUE) {
			/* is already active replica */
			SaCkptResponseSend(&ckptResp);
			break;
		}

		strcpy(ckptReq->toNodeName, openCkpt->activeNodeName);
		if (replica->flagReplicaPending == TRUE) {
			client->pendingRequestList = g_list_append(
				client->pendingRequestList,
				ckptReq);
			break;
		}

		g_hash_table_insert(client->requestHash,
			(gpointer)&(ckptReq->clientRequest->requestNO),
			(gpointer)ckptReq);

		ckptMsg = SaCkptMessageCreateReq(ckptReq, M_CKPT_ACT_SET_BCAST);
		SaCkptMessageBroadcast(ckptMsg);
		SaCkptFree((void*)&ckptMsg);
		ckptMsg = NULL;
		
		SaCkptRequestStartTimer(ckptReq, timeout);
		
		break;	
		
	case REQ_SEC_CRT:
		sectionName = ((SaCkptReqSecCrtParamT *)(ckptReq->clientRequest->reqParam))->sectionID.id;
		goto begin;
	case REQ_SEC_DEL:
		sectionName = ((SaCkptReqSecDelParamT *)(ckptReq->clientRequest->reqParam))->sectionID.id;
		goto begin;
	case REQ_SEC_WRT:
		sectionName = ((SaCkptReqSecWrtParamT *)(ckptReq->clientRequest->reqParam))->sectionID.id;
		goto begin;
	case REQ_SEC_OWRT:
		sectionName = ((SaCkptReqSecOwrtParamT *)(ckptReq->clientRequest->reqParam))->sectionID.id;
		goto begin;
	case REQ_SEC_RD:
		sectionName = ((SaCkptReqSecReadParamT *)(ckptReq->clientRequest->reqParam))->sectionID.id;
		goto begin;
	case REQ_CKPT_SYNC:
	case REQ_CKPT_SYNC_ASYNC:
begin:		strReq = SaCkptReq2String(ckptReq->clientRequest->req);
		if ((ckptReq->clientRequest->req != REQ_CKPT_SYNC) &&
			(ckptReq->clientRequest->req != REQ_CKPT_SYNC_ASYNC)){
			if(saCkptService->flagVerbose){
				cl_log(LOG_INFO,"Request %lu(%s), section is %s\n",	\
				ckptReq->clientRequest->requestNO, strReq,sectionName );
			}
		}
		if (ckptReq->clientRequest->req == REQ_SEC_RD) {
			ckptReq->operation = OP_CKPT_READ;
		} else if ((ckptReq->clientRequest->req == REQ_CKPT_SYNC) ||
			(ckptReq->clientRequest->req == REQ_CKPT_SYNC_ASYNC)){
			ckptReq->operation = OP_CKPT_SYNC;
		} else {
			ckptReq->operation = OP_CKPT_UPD;
		}

		/* the first field of reqParam is the handle */
		checkpointHandle= *(SaCkptCheckpointHandleT*)reqParam;

		openCkpt = g_hash_table_lookup(
			saCkptService->openCheckpointHash,
			(gpointer)&(checkpointHandle));
		if (openCkpt == NULL) {
			ckptResp->resp->retVal = SA_ERR_BAD_HANDLE;
			SaCkptResponseSend(&ckptResp);
			break;
		}
		replica = openCkpt->replica;
		ckptReq->openCkpt = openCkpt;

		strcpy(ckptReq->toNodeName, openCkpt->activeNodeName);
		
		if ((replica != NULL) && 
			(replica->flagReplicaPending == TRUE)) {
			client->pendingRequestList = g_list_append(
				client->pendingRequestList,
				ckptReq);
			break;
		}
		/* FIXME */
		/* How about replica is NULL ? */

		/* add request to hash table */
		g_hash_table_insert(client->requestHash,
			&(ckptReq->clientRequest->requestNO),
			ckptReq);

		if (ckptReq->clientRequest->req == REQ_SEC_RD) {
			ckptMsg = SaCkptMessageCreateReq(ckptReq, M_CKPT_READ);
		} else if ((ckptReq->clientRequest->req == REQ_CKPT_SYNC) ||
			(ckptReq->clientRequest->req == REQ_CKPT_SYNC_ASYNC)){
			SaCkptReqSyncParamT* syncParam = NULL;

			syncParam = (SaCkptReqSyncParamT*)reqParam;
			if (syncParam->timeout < timeout) {
				timeout = syncParam->timeout;
			}
			ckptMsg = SaCkptMessageCreateReq(ckptReq, M_CKPT_SYNC);
		} else {
			ckptMsg = SaCkptMessageCreateReq(ckptReq, M_CKPT_UPD);
		}
		SaCkptMessageSend(ckptMsg, openCkpt->activeNodeName);
		SaCkptFree((void*)&ckptMsg);
		ckptMsg = NULL;
		
		SaCkptRequestStartTimer(ckptReq, timeout);
		
		break;

	case REQ_SEC_EXP_SET:
		secExpSetParam = reqParam;
		checkpointHandle= secExpSetParam->checkpointHandle;

		openCkpt = g_hash_table_lookup(
			saCkptService->openCheckpointHash,
			(gpointer)&(checkpointHandle));
		if ((openCkpt == NULL) || 
			(openCkpt->replica == NULL)) {
			ckptResp->resp->retVal = SA_ERR_LIBRARY;
			SaCkptResponseSend(&ckptResp);
			break;
		}
		ckptReq->openCkpt = openCkpt;
		strcpy(ckptReq->toNodeName, openCkpt->activeNodeName);
		
		replica = openCkpt->replica;
		section = SaCkptSectionFind(replica, 
					&secExpSetParam->sectionID);
		if (section == NULL) {
			ckptResp->resp->retVal = SA_ERR_LIBRARY;
			SaCkptResponseSend(&ckptResp);
			break;
		}

		section->expirationTime = secExpSetParam->expireTime;
		/* FIXME: start expiration timer */
		
		ckptResp->resp->retVal = SA_OK;
		SaCkptResponseSend(&ckptResp);
		
		break;
		
	case REQ_CKPT_STAT_GET:
		/* the first field of reqParam is the handle */
		checkpointHandle= *(SaCkptCheckpointHandleT*)reqParam;

		openCkpt = g_hash_table_lookup(
			saCkptService->openCheckpointHash,
			(gpointer)&(checkpointHandle));
		if ((openCkpt == NULL) || 
			(openCkpt->replica == NULL)) {
			ckptResp->resp->retVal = SA_ERR_LIBRARY;
			SaCkptResponseSend(&ckptResp);
			break;
		}
		ckptReq->openCkpt = openCkpt;
		strcpy(ckptReq->toNodeName, openCkpt->activeNodeName);
		
		replica = openCkpt->replica;

		checkpointStatus = SaCkptMalloc(
			sizeof(SaCkptCheckpointStatusT));
		if (checkpointStatus == NULL) {
			ckptResp->resp->retVal = SA_ERR_NO_MEMORY;
			SaCkptResponseSend(&ckptResp);
			break;
		}

		attr = &(checkpointStatus->checkpointCreationAttributes);
		
		attr->checkpointSize = replica->maxCheckpointSize;
		attr->creationFlags = replica->createFlag;
		attr->maxSectionIdSize = replica->maxSectionIDSize;
		attr->maxSections = replica->maxSectionNumber;
		attr->maxSectionSize = replica->maxSectionSize;
		attr->retentionDuration = replica->retentionDuration;
		checkpointStatus->numberOfSections = replica->sectionNumber;
		checkpointStatus->memoryUsed = replica->checkpointSize; 
					
		ckptResp->resp->retVal = SA_OK;
		ckptResp->resp->data = checkpointStatus;
		ckptResp->resp->dataLength = sizeof(SaCkptCheckpointStatusT);
		SaCkptResponseSend(&ckptResp);
		
		break;
				
	case REQ_CKPT_RTN_SET:
		rtnParam = reqParam;
		checkpointHandle= rtnParam->checkpointHandle;

		openCkpt = g_hash_table_lookup(
			saCkptService->openCheckpointHash,
			(gpointer)&(checkpointHandle));
		if ((openCkpt == NULL) || 
			(openCkpt->replica == NULL)) {
			ckptResp->resp->retVal = SA_ERR_LIBRARY;
			SaCkptResponseSend(&ckptResp);
			break;
		}
		ckptReq->openCkpt = openCkpt;
		strcpy(ckptReq->toNodeName, openCkpt->activeNodeName);
		
		replica = openCkpt->replica;
		replica->retentionDuration = rtnParam->retention;

		ckptResp->resp->retVal = SA_OK;
		SaCkptResponseSend(&ckptResp);