libckpt.c

linux集群服务器软件代码包

			}
			ha_free(ckptResp);
			return SA_ERR_NO_MEMORY;
		} 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);
	
	libAsyncResponseList = g_list_append(
		libAsyncResponseList,
		ckptResp);

	return retval;
	
}

/* ------------------------- exported API ------------------------------- */

/* 
 * Initialize the checkpoint service for the invoking process and register
 * the related callback functions. 
 */
SaErrorT
saCkptInitialize(SaCkptHandleT *ckptHandle/*[out]*/,
	const SaCkptCallbacksT *callbacks,
	const SaVersionT *version)
{	
	SaErrorT libError = SA_OK;
	
	SaCkptLibClientT* libClient = NULL;
	SaCkptLibRequestT* libRequest = NULL;
	SaCkptClientRequestT* clientRequest = NULL;
	SaCkptReqInitParamT* initParam = NULL;
	SaCkptClientResponseT* clientResponse = NULL;

	int i = 0;

	cl_log_set_entity("AIS");
	cl_log_enable_stderr(TRUE);
	
	if (ckptHandle == NULL) {
		cl_log(LOG_ERR, 
			"Null handle in saCkptInitialize");
		return SA_ERR_INVALID_PARAM;
	}
	
	if (version == NULL) {
		cl_log(LOG_ERR, 
			"Null version number in saCkptInitialize");
		return SA_ERR_INVALID_PARAM;
	}

	libClient = (SaCkptLibClientT*)ha_malloc(
					sizeof(SaCkptLibClientT));
  	libRequest = (SaCkptLibRequestT*)ha_malloc(
					sizeof(SaCkptLibRequestT));
	clientRequest = (SaCkptClientRequestT*)ha_malloc(
					sizeof(SaCkptClientRequestT));
	initParam = (SaCkptReqInitParamT*)ha_malloc(
					sizeof(SaCkptReqInitParamT));
	if ((libClient == NULL) || 
		(libRequest == NULL) ||
		(clientRequest == NULL) ||
		(initParam == NULL)) {
		cl_log(LOG_ERR, 
			"No memory in saCkptInitialize");
		libError = SA_ERR_NO_MEMORY;
		goto initError;
	}
	
	memset(libClient, 0, sizeof(SaCkptLibClientT));
	memset(libRequest, 0, sizeof(SaCkptLibRequestT));
	memset(clientRequest, 0, sizeof(SaCkptClientRequestT));
	memset(initParam, 0, sizeof(SaCkptReqInitParamT));
	
	libClient->hostName[0] = 0;
	libClient->pid = getpid();
	libClient->threadID = 0;
	libClient->clientHandle = 0;
	libClient->checkpointList = NULL;
	for (i=0; i<2; i++) {
		IPC_Channel* ch = NULL;
		char pathname[128];
		
		memset (pathname, 0, sizeof(pathname));
		strcpy (pathname, CKPTIPC) ;
		ch = SaCkptClientChannelInit(pathname);
		if (ch == NULL) {
			cl_log(LOG_ERR, 
			"Can not initiate connection in saCkptInitialize");
			libError = SA_ERR_LIBRARY;
			goto initError;
		}
		if (ch->ops->initiate_connection(ch)
			!= IPC_OK) {
			cl_log(LOG_ERR, 
			"Can not connect to daemon in saCkptInitialize");
			libError = SA_ERR_LIBRARY;
			goto initError;
		}

		if (i == 1)  { /* async channel */
			ch->ops->set_recv_qlen(ch, 0);
			/* ch->ops->set_send_qlen(ch, 0); */
		}

		libClient->channel[i] = ch;
	}

	libRequest->client = libClient;
	libRequest->timeoutTag = 0;
	libRequest->clientRequest = clientRequest;

	clientRequest->clientHandle = 0;
	clientRequest->requestNO = SaCkptLibGetReqNO();
	clientRequest->req = REQ_SERVICE_INIT;
	clientRequest->reqParamLength = sizeof(SaCkptReqInitParamT);
	clientRequest->dataLength = 0;
	clientRequest->reqParam = initParam;
	clientRequest->data = NULL;

	initParam->pid = libClient->pid;
	initParam->tid = libClient->threadID;
	memcpy(&(initParam->ver), version, sizeof(SaVersionT));

	libError = SaCkptLibRequestSend(libClient->channel[0], 
		libRequest->clientRequest);
	if (libError != SA_OK) {
		cl_log(LOG_ERR, 
			"Send initialize request failed");
		goto initError;
	}

	libError = SaCkptLibResponseReceive(libClient->channel[0], 
		libRequest->clientRequest->requestNO,
		&clientResponse);
	if (libError != SA_OK) {
		cl_log(LOG_ERR, 
			"Receive response failed");
		goto initError;
	}
	if (clientResponse == NULL) {
		cl_log(LOG_ERR, 
			"Received null response");
		libError = SA_ERR_LIBRARY;
		goto initError;
	}
	if (clientResponse->retVal != SA_OK) {
		cl_log(LOG_ERR, 
			"Checkpoint daemon returned error");
		libError = clientResponse->retVal;
		goto initError;
	}

	*ckptHandle = clientResponse->clientHandle;
	
	libClient->clientHandle = *ckptHandle;
	if (callbacks != NULL) {
		libClient->callbacks = *callbacks;
	}
	libClientList = g_list_append(libClientList, libClient);

	if (libIteratorHash == NULL) {
		libIteratorHash = 
			g_hash_table_new(g_int_hash, g_int_equal);
	}

	libError = SA_OK;

initError:
	if (libError != SA_OK) {
		if (libClient != NULL) {
			for (i=0; i<2; i++) {
				if (libClient->channel[i] != NULL) {
					IPC_Channel* ch = 
						libClient->channel[i];
					ch->ops->destroy(ch);
				}
			}
			ha_free(libClient);
		}
	}
	
	if (libRequest != NULL) {
		ha_free(libRequest);
	}
	
	if (clientRequest != NULL) {
		ha_free(clientRequest);
	}
	
	if (initParam != NULL) {
		ha_free(initParam);
	}

	if (clientResponse != NULL) {
		if (clientResponse->dataLength > 0) {
			ha_free(clientResponse->data);
		}
		ha_free(clientResponse);
	}

	return libError;
}

/* 
 * Returns the fd for asynchronously operation
 */
SaErrorT
saCkptSelectionObjectGet(
	const SaCkptHandleT *ckptHandle,
	SaSelectionObjectT *selectionObject/*[out]*/)
{
	SaCkptLibClientT* libClient = NULL;
	IPC_Channel* ch = NULL;

	if (ckptHandle == NULL) {
		cl_log(LOG_ERR, 
			"Null handle in saCkptSelectionObjectGet");
		return SA_ERR_INVALID_PARAM;
	}

	if (selectionObject == NULL) {
		cl_log(LOG_ERR, 
			"Null selectobject in saCkptSelectionObjectGet");
		return SA_ERR_INVALID_PARAM;
	}

	libClient = SaCkptGetLibClientByHandle(*ckptHandle);
	if (libClient == NULL) {
		cl_log(LOG_ERR, 
			"Invalid handle in saCkptSelectionObjectGet");
		return SA_ERR_INVALID_PARAM;
	}
	
	ch = libClient->channel[1];
	*selectionObject = ch->ops->get_recv_select_fd(ch);
	
	return SA_OK;
}

/* 
 * This function invokes, in the context of the calling thread,one or all 
 * of the pending callbacks for the handle ckptHandle.
 */
SaErrorT
saCkptDispatch(const SaCkptHandleT *ckptHandle,
	SaDispatchFlagsT dispatchFlags)
{
	SaCkptLibClientT* libClient = NULL;
	SaCkptLibRequestT* libRequest = NULL;
	
	SaCkptClientRequestT* clientRequest = NULL;
 	SaCkptClientResponseT* clientResponse = NULL;
	
	SaInvocationT invocation;
	SaCkptLibCheckpointT* libCheckpoint = NULL;
	SaCkptCheckpointHandleT checkpointHandle = 0;
	SaUint32T reqno;
	SaCkptReqOpenAsyncParamT* openAsyncParam = NULL;
	SaCkptReqAsyncParamT* asyncParam = NULL;
	
	SaErrorT libError = SA_OK;
	IPC_Channel* ch = NULL;
	
	if (ckptHandle == NULL) {
		cl_log(LOG_ERR, 
			"Null handle in saCkptDispatch");
		return SA_ERR_INVALID_PARAM;
	}

	if ((dispatchFlags != SA_DISPATCH_ONE) &&
		(dispatchFlags != SA_DISPATCH_ALL) && 
		(dispatchFlags != SA_DISPATCH_BLOCKING)) {
		cl_log(LOG_ERR, 
			"Invalid dispatchFlags in saCkptDispatch");
		return SA_ERR_INVALID_PARAM;
	}

	libClient = SaCkptGetLibClientByHandle(*ckptHandle);
	if (libClient == NULL) {
		cl_log(LOG_ERR, 
			"Invalid handle in saCkptDispatch");
		return SA_ERR_INVALID_PARAM;
	}
	
	ch = libClient->channel[1]; /* async channel */
	libError = SaCkptLibResponseReceiveAsync(ch);
	if (libError != SA_OK) {
		cl_log(LOG_ERR, 
			"Receive response failed");
		return SA_ERR_LIBRARY;
	}
	
	if (g_list_length(libAsyncResponseList) == 0) {
		return SA_OK;
	}

	while (libAsyncResponseList != NULL) {
		clientResponse = libAsyncResponseList->data;
		
		libError = clientResponse->retVal;
		reqno = clientResponse->requestNO;
		libRequest = SaCkptGetLibRequestByReqno(reqno);
		clientRequest = libRequest->clientRequest;
		switch (clientRequest->req) {
		case REQ_CKPT_OPEN_ASYNC:
			openAsyncParam = clientRequest->reqParam;
			invocation = openAsyncParam->invocation;
			
			if (clientResponse->data == NULL){
				libError = SA_ERR_LIBRARY;
				libClient->callbacks.saCkptCheckpointOpenCallback(
					invocation, NULL, libError);
				break;
			}
			memcpy(&checkpointHandle, clientResponse->data,
				sizeof(SaCkptCheckpointHandleT));

			/*
			 * create libCheckpoint and add it to the 
			 * opened checkpoint list
			 */
			libCheckpoint = ha_malloc(sizeof(SaCkptLibCheckpointT));
			if (libCheckpoint == NULL) {
				libError = SA_ERR_NO_MEMORY;
				libClient->callbacks.saCkptCheckpointOpenCallback(
					invocation, &checkpointHandle, libError);
				break;
			}
			libCheckpoint->client = libClient;
			libCheckpoint->checkpointHandle = checkpointHandle;
			libCheckpoint->ckptName.length = 
				openAsyncParam->ckptName.length;
			memcpy(libCheckpoint->ckptName.value, 
				openAsyncParam->ckptName.value,
				openAsyncParam->ckptName.length);
			memcpy(&(libCheckpoint->createAttributes),
				&openAsyncParam->attr,
				sizeof(SaCkptCheckpointCreationAttributesT));
			libCheckpoint->openFlag = openAsyncParam->openFlag;
				
			libClient->checkpointList = g_list_append(
				libClient->checkpointList,
				libCheckpoint);
			libCheckpointList = g_list_append(libCheckpointList,
				libCheckpoint);
			
			libClient->callbacks.saCkptCheckpointOpenCallback(
				invocation, &checkpointHandle, libError);
			break;
			
		case REQ_CKPT_SYNC_ASYNC:
			asyncParam = clientRequest->reqParam;
			invocation = asyncParam->invocation;
			libClient->callbacks.saCkptCheckpointSynchronizeCallback(
				invocation, libError);
			break;
			
		default:
			break;
		};

		libAsyncResponseList = g_list_remove(libAsyncResponseList,
			clientResponse);
		libAsyncRequestList = g_list_remove(libAsyncRequestList,
			libRequest);
		ha_free(clientResponse->data);
		ha_free(clientResponse);
		ha_free(libRequest->clientRequest->reqParam);
		ha_free(libRequest->clientRequest->data);
		ha_free(libRequest->clientRequest);
		ha_free(libRequest);
		
		if (dispatchFlags == SA_DISPATCH_ONE) {
			break;
		}
	}
	
	return SA_OK; 
}

/* 
 * closes the association, represented by ckptHandle, between the process and
 * the Checkpoint Service. It frees up resources and close all the opened 
 * checkpoints.
 */
SaErrorT
saCkptFinalize(const SaCkptHandleT *ckptHandle)
{
	SaCkptLibClientT* libClient = NULL;
	SaCkptLibRequestT* libRequest = NULL;
	SaCkptClientRequestT* clientRequest = NULL;
	SaCkptReqFinalParamT* finalParam = NULL;
 	SaCkptClientResponseT* clientResponse = NULL;
	
	SaErrorT libError = SA_OK;
	IPC_Channel* ch = NULL;
	int i = 0;
	
	GList* list = NULL;

	if (ckptHandle == NULL) {
		cl_log(LOG_ERR, 
			"Null handle in saCkptFinalize");
		return SA_ERR_INVALID_PARAM;
	}

	libClient = SaCkptGetLibClientByHandle(*ckptHandle);
	if (libClient == NULL) {
		cl_log(LOG_ERR, 
			"Invalid handle in saCkptFinalize");
		return SA_ERR_INVALID_PARAM;
	}
	
	/*
	 * close all opened checkpoints first
	 */
	list = libClient->checkpointList;
	while(list != NULL ) {
		SaCkptLibCheckpointT* libCheckpoint = NULL;
		SaCkptCheckpointHandleT* checkpointHandle = NULL;

		libCheckpoint = (SaCkptLibCheckpointT*)(list->data);
		checkpointHandle = &(libCheckpoint->checkpointHandle);
		saCkptCheckpointClose(checkpointHandle);
/*		ha_free(libCheckpoint); */

		list = libClient->checkpointList;
	}
	
  	libRequest = (SaCkptLibRequestT*)ha_malloc(
					sizeof(SaCkptLibRequestT));
	clientRequest = (SaCkptClientRequestT*)ha_malloc(
					sizeof(SaCkptClientRequestT));
	finalParam = (SaCkptReqFinalParamT*)ha_malloc(
					sizeof(SaCkptReqFinalParamT));
 	if ((libRequest == NULL) ||
		(clientRequest == NULL) ||
		(finalParam == NULL)) {
		cl_log(LOG_ERR, 
			"No memory in saCkptFinalize");
		libError = SA_ERR_NO_MEMORY;
		goto finalError;
	}
	
	memset(libRequest, 0, sizeof(SaCkptLibRequestT));
	memset(clientRequest, 0, sizeof(SaCkptClientRequestT));
	memset(finalParam, 0, sizeof(SaCkptReqFinalParamT));
	
	libRequest->client = libClient;
	libRequest->timeoutTag = 0;
	libRequest->clientRequest = clientRequest;

	clientRequest->clientHandle = libClient->clientHandle;
	clientRequest->requestNO = SaCkptLibGetReqNO();
	clientRequest->req = REQ_SERVICE_FINL;
	clientRequest->reqParamLength = sizeof(SaCkptReqFinalParamT);
	clientRequest->dataLength = 0;
	clientRequest->reqParam = finalParam;
	clientRequest->data = NULL;

	finalParam->clientHandle = *ckptHandle;

	ch = libClient->channel[0];
	libError = SaCkptLibRequestSend(ch, libRequest->clientRequest);
	if (libError != SA_OK) {
		cl_log(LOG_ERR, 
			"Send finalize request failed");
		goto finalError;
	}

	libError = SaCkptLibResponseReceive(ch, 
		libRequest->clientRequest->requestNO,
		&clientResponse);
	if (libError != SA_OK) {
		cl_log(LOG_ERR, 
			"Receive response failed");
		goto finalError;
	}
	if (clientResponse == NULL) {
		cl_log(LOG_ERR, 
			"Received null response");
		libError = SA_ERR_LIBRARY;
		goto finalError;
	}
	if (clientResponse->retVal != SA_OK) {
		cl_log(LOG_ERR, 
			"Checkpoint daemon returned error");
		libError = clientResponse->retVal;
		goto finalError;
	}

	
	/*
	 * FIXME
	 * cancel all pending callbacks related to this client
	 */
	
	/* remove client */
	libClientList = g_list_remove(libClientList, libClient);

	/* only destroy the hash table after all the clients finalized */
	if (g_list_length(libClientList) == 0) {
		g_hash_table_destroy(libIteratorHash);
		libIteratorHash = NULL;
	}
	
	libError = SA_OK;

finalError:
	if (libError == SA_OK) {
		if (libClient != NULL) {
			for (i=0; i<2; i++) {
				if (libClient->channel[i] != NULL) {
					ch = libClient->channel[i];
					ch->ops->destroy(ch);