libckpt.c

linux集群服务器软件代码包

/* $Id: libckpt.c,v 1.18 2004/12/06 21:15:45 msoffen Exp $ */
/* 
 * ckptlib.c: data checkpoint API library
 *
 * Copyright (C) 2003 Jerry Yu <[e-mail]>
 * 
 * This library is free software; you can redistribute it and/or
 * modify it under the terms of the GNU Lesser 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 library 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
 * Lesser General Public License for more details.
 * 
 * You should have received a copy of the GNU Lesser 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
 */
 
/* 
 * This library is an implementation of the Application Interface 
 * Specification on Service Availability Forum. Refer to: www.saforum.org/
 * specification
 */
#include <stdio.h>
#include <stdlib.h>
#include <unistd.h>
#include <sys/types.h>
#include <sys/stat.h>
#include <sys/socket.h>
#include <sys/time.h>
#include <sys/un.h>
#include <errno.h>
#include <string.h>
#include <time.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 <saf/ais.h>
#include <checkpointd/clientrequest.h>
#include <clplumbing/cl_malloc.h>

#ifndef AF_LOCAL
#	define AF_LOCAL	AF_UNIX
#endif

/* 
 * TODO list: 
 * 1. make all APIs thread safe
 */

/* 
 * the request timeout value in seconds 
 */

#define	LIB_REQUEST_TIMEOUT	10

/* 
 * the client structure  
 */
typedef struct _SaCkptLibClientT{
	char		hostName[SA_MAX_NAME_LENGTH];
	pid_t		pid;
	int		threadID;

	/* The handle returned by the checkpoint service daemon */
	SaCkptHandleT	clientHandle;

	/* 
	 * The client channel 
	 * channel[0] is for the sync calls
	 * channel[1] is for the async calls
	 */
	IPC_Channel*	channel[2];
	
	/* 
	 * the opened checkpoints
	 */
	GList*		checkpointList;

	SaCkptCallbacksT callbacks;
} SaCkptLibClientT; 

/* 
 * the request structure  
 */
typedef struct _SaCkptLibRequestT{
	SaCkptLibClientT*	client;

	SaCkptClientRequestT*	clientRequest;

	/* request timeout handler tag */
	guint			timeoutTag;

} SaCkptLibRequestT;

typedef struct _SaCkptLibCheckpointT{
	SaCkptLibClientT*	client;

	/* 
	 * opened checkpoint handle. 
	 * Returned by checkpoint service daemon
	 */
	SaCkptCheckpointHandleT checkpointHandle; 
	
	SaNameT ckptName; 

	/*
	 * checkpoint attributes
	 */
	SaCkptCheckpointOpenFlagsT		openFlag;
	SaCkptCheckpointCreationAttributesT	createAttributes;

	GList* sectionList;
} SaCkptLibCheckpointT;

GList*	libClientList = NULL;
GList*	libCheckpointList = NULL;

GList* 	libResponseList = NULL;

GList*	libAsyncRequestList = NULL;
GList*	libAsyncResponseList = NULL;

GHashTable*	libIteratorHash = NULL;

static IPC_Channel*
SaCkptClientChannelInit(char* pathname)
{
	IPC_Channel *clientChannel = NULL;
	mode_t mask;
	char path[] = IPC_PATH_ATTR;
	char domainsocket[] = IPC_DOMAIN_SOCKET;

	GHashTable *attrs = g_hash_table_new(g_str_hash,g_str_equal);

	g_hash_table_insert(attrs, path, pathname);

	mask = umask(0);
	clientChannel = ipc_channel_constructor(domainsocket, attrs);
	if (clientChannel == NULL){
		cl_log(LOG_ERR, 
			"Checkpoint library Can't create client channel");
		return NULL;
	}
	mask = umask(mask);

	g_hash_table_destroy(attrs);

	return clientChannel;
}

static SaUint32T 
SaCkptLibGetReqNO(void)
{
	static SaUint32T ckptLibRequestNO = 1;
	return ckptLibRequestNO++;
}

/*	FIXME it should not be a global static variable */
static SaCkptSectionIteratorT 
SaCkptLibGetIterator(void)
{
	static SaCkptSectionIteratorT ckptLibSecIterator = 1;
	
	SaCkptSectionIteratorT iterator = 0;
	GList* sectionList = NULL;

	do {
		iterator = ckptLibSecIterator++;
		sectionList = g_hash_table_lookup(
			libIteratorHash, &iterator);
	} while (sectionList != NULL);

	return iterator;
}

static SaCkptLibRequestT*
SaCkptGetLibRequestByReqno(SaUint32T reqno)
{
	SaCkptLibRequestT* libRequest = NULL;
	GList* list = NULL;
	
	list = libAsyncRequestList;
	while( list != NULL ) {
		libRequest= (SaCkptLibRequestT*)(list->data);
		if(libRequest->clientRequest->requestNO == reqno) {
			return libRequest;
		}
		list = g_list_next(list);
	}
	return NULL;
}

static SaCkptClientResponseT*
SaCkptGetLibResponseByReqno(SaUint32T reqno)
{
	SaCkptClientResponseT* libResponse = NULL;
	GList* list = NULL;
	
	list = libResponseList;
	while( list != NULL ) {
		libResponse= (SaCkptClientResponseT*)(list->data);
		if(libResponse->requestNO == reqno) {
			libResponseList = g_list_remove(
				libResponseList,
				libResponse);
			return libResponse;
		}
		list = g_list_next(list);
	}
	return NULL;

}

static SaCkptLibClientT*
SaCkptGetLibClientByHandle(SaCkptHandleT clientHandle)
{
	SaCkptLibClientT* libClient = NULL;
	GList* list = NULL;
	
	list = libClientList;
	while( list != NULL ) {
		libClient = (SaCkptLibClientT*)(list->data);
		if(libClient->clientHandle == clientHandle) {
			return libClient;
		}
		list = g_list_next(list);
	}
	return NULL;
}

static SaCkptLibCheckpointT*
SaCkptGetLibCheckpointByHandle(
	SaCkptCheckpointHandleT checkpointHandle)
{
	GList* list = NULL;
	SaCkptLibCheckpointT* libCheckpoint = NULL;

	list = libCheckpointList;
	while(list != NULL ) {
		libCheckpoint = (SaCkptLibCheckpointT*)(list->data);
		if(libCheckpoint->checkpointHandle == checkpointHandle) {
			return libCheckpoint;
		}
		list = g_list_next(list);
	}
	return NULL;
}

static SaErrorT
SaCkptLibRequestSend(IPC_Channel* ch, 
	SaCkptClientRequestT* ckptReq) 
{
	IPC_Message* ipcMsg = NULL;
	
	int rc = IPC_OK;
	char* p = NULL;

	if(ch->ch_status != IPC_CONNECT) {
		cl_log(LOG_WARNING, 
			"IPC is in state %d before send message",
			ch->ch_status);
		return SA_ERR_LIBRARY;
	}
	
	/* build response message */
	ipcMsg = (IPC_Message*)ha_malloc(sizeof(IPC_Message));
	if (ipcMsg == NULL) {
		cl_log(LOG_ERR, "No memory in checkpoint library");
		return SA_ERR_NO_MEMORY;
	}
	
	memset(ipcMsg, 0, sizeof(IPC_Message));
	ipcMsg->msg_private = NULL;
	ipcMsg->msg_done = NULL;
	ipcMsg->msg_ch = ch;
	ipcMsg->msg_len = sizeof(SaCkptClientRequestT) - 
		2* sizeof(void*) + 
		ckptReq->dataLength +
		ckptReq->reqParamLength ;
	ipcMsg->msg_body = ha_malloc(ipcMsg->msg_len);
	if (ipcMsg->msg_body == NULL) {
		cl_log(LOG_ERR, "No memory in checkpoint library");
		ha_free(ipcMsg);
		return SA_ERR_NO_MEMORY;
	}

	p = ipcMsg->msg_body;
	
	memcpy(p, ckptReq, 
		sizeof(SaCkptClientRequestT) - 2*sizeof(void*));
	p += sizeof(SaCkptClientRequestT) - 2*sizeof(void*);

	if (ckptReq->reqParamLength > 0) {
		memcpy(p, ckptReq->reqParam,
			ckptReq->reqParamLength);
		p += ckptReq->reqParamLength;
	}
	
	if (ckptReq->dataLength> 0) {
		memcpy(p, ckptReq->data, 
			ckptReq->dataLength);
		p += ckptReq->dataLength;
	}
		
	/* send request message */
	while (ch->ops->send(ch, ipcMsg) == IPC_FAIL) {
		cl_log(LOG_ERR, "Checkpoint library send request failed");
		cl_log(LOG_ERR, "Sleep for a while and try again");
		cl_shortsleep();
	}
	if(ch->ch_status != IPC_CONNECT) {
		cl_log(LOG_WARNING, 
			"IPC is in state %d after send message",
			ch->ch_status);
	}

	ch->ops->waitout(ch);
	
	/* free ipc message */
	if (ipcMsg != NULL) {
		if (ipcMsg->msg_body != NULL) {
			ha_free(ipcMsg->msg_body);
		}
		ha_free(ipcMsg);
	}

	if (rc == IPC_OK) {
		return SA_OK;
	} else {
		return SA_ERR_LIBRARY;
	}
	
}

static SaErrorT 
SaCkptLibResponseReceive(IPC_Channel* ch, 
	SaUint32T requestNO,
	SaCkptClientResponseT** pCkptResp) 
{

	SaCkptClientResponseT* ckptResp = NULL;

	IPC_Message	*ipcMsg = NULL;
	int		rc = IPC_OK;
	SaErrorT	retval = SA_OK;
	char		*p = NULL;

	ckptResp = SaCkptGetLibResponseByReqno(requestNO);
	if (ckptResp != NULL) {
		*pCkptResp = ckptResp;
		return SA_OK;
	}
	
	if(ch->ch_status != IPC_CONNECT) {
		cl_log(LOG_WARNING, 
			"IPC is in state %d before receive message",
			ch->ch_status);
		return SA_ERR_LIBRARY;
	}
	
	while (ch->ops->is_message_pending(ch) != TRUE) {
		cl_shortsleep();
	}

	while (ch->ops->is_message_pending(ch) == TRUE) {
		/* receive ipc message */
		rc = ch->ops->recv(ch, &ipcMsg);
		if (rc != IPC_OK) {
			cl_log(LOG_ERR, "Receive response failed");
			if (ipcMsg->msg_body != NULL) {
				free(ipcMsg->msg_body);
			}
			free(ipcMsg);
			retval =  SA_ERR_LIBRARY;
			break;
		}

		if (ipcMsg->msg_len <
			sizeof(SaCkptClientResponseT) - sizeof(void*)) {
			cl_log(LOG_ERR, "Received error response");
			if (ipcMsg->msg_body != NULL) {
				free(ipcMsg->msg_body);
			}
			free(ipcMsg);
			retval = SA_ERR_LIBRARY;
			break;
		}
		p = ipcMsg->msg_body;

		ckptResp = ha_malloc(sizeof(SaCkptClientResponseT));
		if (ckptResp == NULL) {
			cl_log(LOG_ERR, 
				"No memory in checkpoint library");
			if (ipcMsg != NULL) {
				if (ipcMsg->msg_body != NULL) {
					free(ipcMsg->msg_body);
				}
				free(ipcMsg);
			}
			retval = SA_ERR_NO_MEMORY;
			break;
		}

		memset(ckptResp, 0, sizeof(SaCkptClientResponseT));
		memcpy(ckptResp, p, 
			sizeof(SaCkptClientResponseT) - sizeof(void*));
		p += (sizeof(SaCkptClientResponseT) - sizeof(void*));

		if (ckptResp->dataLength > 0) {
			ckptResp->data = ha_malloc(ckptResp->dataLength);
			if (ckptResp->data == NULL) {
				cl_log(LOG_ERR, 
					"No memory in checkpoint library");
				if (ipcMsg != NULL) {
					if (ipcMsg->msg_body != NULL) {
						free(ipcMsg->msg_body);
					}
					free(ipcMsg);
				}
				ha_free(ckptResp);
				retval = SA_ERR_NO_MEMORY;
				break;
			} else {
				memcpy(ckptResp->data, p, 
					ckptResp->dataLength);
				p += ckptResp->dataLength;
			}
		} else {
			ckptResp->data = NULL;
		}

		/* free ipc message */
		if (ipcMsg->msg_body != NULL) {
			free(ipcMsg->msg_body);
		}
		free(ipcMsg);
		
		libResponseList = g_list_append(libResponseList,
			ckptResp);
	}
	
	ckptResp = SaCkptGetLibResponseByReqno(requestNO);
	if (ckptResp != NULL) {
		*pCkptResp = ckptResp;
		return SA_OK;
	}

	return retval;
}

static SaErrorT 
SaCkptLibResponseReceiveAsync(IPC_Channel* ch) 
{
	SaCkptClientResponseT* ckptResp = NULL;

	IPC_Message	*ipcMsg = NULL;
	int		rc = IPC_OK;
	SaErrorT	retval = SA_OK;
	char		*p = NULL;

	if(ch->ch_status != IPC_CONNECT) {
		cl_log(LOG_WARNING, 
			"IPC is in state %d before receive message",
			ch->ch_status);
		return SA_ERR_LIBRARY;
	}
	
	/* receive ipc message */
	rc = ch->ops->recv(ch, &ipcMsg);
	if (rc != IPC_OK) {
		cl_log(LOG_ERR, "Receive response failed");
		if (ipcMsg->msg_body != NULL) {
			free(ipcMsg->msg_body);
		}
		free(ipcMsg);
		return  SA_ERR_LIBRARY;
	}

	if (ipcMsg->msg_len <
		sizeof(SaCkptClientResponseT) - sizeof(void*)) {
		cl_log(LOG_ERR, "Received error response");
		if (ipcMsg->msg_body != NULL) {
			free(ipcMsg->msg_body);
		}
		free(ipcMsg);
		return SA_ERR_LIBRARY;
	}
	p = ipcMsg->msg_body;

	ckptResp = ha_malloc(sizeof(SaCkptClientResponseT));
	if (ckptResp == NULL) {
		cl_log(LOG_ERR, 
			"No memory in checkpoint library");
		if (ipcMsg != NULL) {
			if (ipcMsg->msg_body != NULL) {
				free(ipcMsg->msg_body);
			}
			free(ipcMsg);
		}
		return SA_ERR_NO_MEMORY;
	}

	memset(ckptResp, 0, sizeof(SaCkptClientResponseT));
	memcpy(ckptResp, p, 
		sizeof(SaCkptClientResponseT) - sizeof(void*));
	p += (sizeof(SaCkptClientResponseT) - sizeof(void*));

	if (ckptResp->dataLength > 0) {
		ckptResp->data = ha_malloc(ckptResp->dataLength);
		if (ckptResp->data == NULL) {
			cl_log(LOG_ERR, 
				"No memory in checkpoint library");
			if (ipcMsg != NULL) {
				if (ipcMsg->msg_body != NULL) {
					free(ipcMsg->msg_body);
				}
				free(ipcMsg);