replica.c

linux集群服务器软件代码包

		sec->dataUpdateState = OP_STATE_COMMITTED;
		sec->lastUpdateTime = time_longclock();
		
		break;
		
	default:
		break;
			
	}

	if (saCkptService->flagVerbose) {
		strReq = SaCkptReq2String(req);
		strErr = SaCkptErr2String(retVal);
		cl_log(LOG_INFO, 
			"Replica %s update committed, request %s, status %s",
			replica->checkpointName, strReq, strErr);
		SaCkptFree((void*)&strReq);
		SaCkptFree((void*)&strErr);
	}
	
	return SA_OK;
}


int 
SaCkptReplicaUpdRollback(SaCkptReplicaT* replica, SaCkptReqT req, 
	int dataLength, void* data,
	int paramLength, void* param)
{
	SaCkptSectionT* sec = NULL;
	
	SaCkptReqSecCrtParamT* secCrtParam = NULL;
	SaCkptReqSecDelParamT* secDelParam = NULL;
/*	SaCkptReqSecReadParamT* secReadParam = NULL; */
	SaCkptReqSecWrtParamT* secWrtParam = NULL;
	SaCkptReqSecOwrtParamT* secOwrtParam = NULL;

	char*	strReq = NULL;
	char*	strErr = NULL;
	int	retVal = SA_OK;

	int	index = 0;
	
	switch (req) {
	case REQ_SEC_CRT:
		secCrtParam = (SaCkptReqSecCrtParamT*)param;
		
		sec = SaCkptSectionFind(replica, 
			&(secCrtParam->sectionID));
		if (sec == NULL) {
			retVal = SA_ERR_NOT_EXIST;
			break;
		}
		if (sec->sectionState != STATE_CREATE_PREPARED) {
			cl_log(LOG_ERR, 
				"Section create rollback: not prepared");
			retVal = SA_ERR_FAILED_OPERATION;
			break;
		}

		/* remove section from section list */
		replica->sectionList = g_list_remove(
			replica->sectionList,
			(gpointer)sec);
		
		if (saCkptService->flagVerbose) {
			char* strSectionID = NULL;
			strSectionID = SaCkptSectionId2String(sec->sectionID);
			cl_log(LOG_INFO, 
				"Rollback: section %s deleted from replica %s",
				sec->sectionID.id,
				replica->checkpointName);
			SaCkptFree((void*)&strSectionID);
		}
		
		/* free section */
		if (sec->data[0] != NULL) {
			SaCkptFree((void**)&(sec->data[0]));
		}
		if (sec->data[1] != NULL) {
			SaCkptFree((void**)&(sec->data[1]));
		}
		SaCkptFree((void*)&sec);
		break;
		
	case REQ_SEC_DEL:
		secDelParam = (SaCkptReqSecDelParamT*)param;

		sec = SaCkptSectionFind(replica, 
			&(secDelParam->sectionID));
		if (sec == NULL) {
			cl_log(LOG_INFO, "Can not find section");
			retVal = SA_ERR_NOT_EXIST;
			break;
		}
		if (sec->sectionState != STATE_DELETE_PREPARED) {
			cl_log(LOG_ERR, 
				"Section delete rollback: not prepared");
			retVal = SA_ERR_FAILED_OPERATION;
			break;
		}
		sec->sectionState = STATE_CREATE_COMMITTED;

		break;
		
	case REQ_SEC_WRT:
		secWrtParam = (SaCkptReqSecWrtParamT*)param;

		sec = SaCkptSectionFind(replica, 
			&(secWrtParam->sectionID));
		if (sec == NULL) {
			cl_log(LOG_INFO, "Can not find section");
			retVal = SA_ERR_NOT_EXIST;
			break;
		}
		if (sec->dataUpdateState != OP_STATE_PREPARED) {
			cl_log(LOG_ERR, 
				"Section write rollback: not prepared");
			retVal = SA_ERR_FAILED_OPERATION;
			break;
		}

		index = (sec->dataIndex + 1) % 2;
		
		/* rollback  */
		SaCkptFree((void**)&(sec->data[index]));
		sec->dataLength[index] = 0;
		sec->dataUpdateState = OP_STATE_ROLLBACKED;

		break;
		
	case REQ_SEC_OWRT:
		secOwrtParam = (SaCkptReqSecOwrtParamT*)param;

		sec = SaCkptSectionFind(replica, 
			&(secOwrtParam->sectionID));
		if (sec == NULL) {
			cl_log(LOG_INFO, "Can not find section");
			retVal = SA_ERR_NOT_EXIST;
			break;
		}
		if (sec->dataUpdateState != OP_STATE_PREPARED) {
			cl_log(LOG_ERR, 
				"Section overwrite rollback: not prepared");
			retVal = SA_ERR_FAILED_OPERATION;
			break;
		}

		index = (sec->dataIndex + 1) % 2;

		/* rollback */
		SaCkptFree((void**)&(sec->data[index]));
		sec->dataLength[index]= 0;
		sec->dataUpdateState = OP_STATE_ROLLBACKED;

		break;
		
	default:
		break;

	}

	if (saCkptService->flagVerbose) {	
		strReq = SaCkptReq2String(req);
		strErr = SaCkptErr2String(retVal);
		cl_log(LOG_INFO, 
			"Replica %s update rollbacked, request %s, status %s",
			replica->checkpointName, strReq, strErr);
		SaCkptFree((void*)&strReq);
		SaCkptFree((void*)&strErr);
	}
	
	return SA_OK;
}


SaCkptSectionT* 
SaCkptSectionFind(SaCkptReplicaT* replica, SaCkptFixLenSectionIdT* sectionID)
{
	GList* list = replica->sectionList;
	SaCkptSectionT* sec = NULL;

	while (list != NULL) {
		sec = (SaCkptSectionT*)list->data;
		if ((sectionID->idLen == sec->sectionID.idLen) && !memcmp(sec->sectionID.id, sectionID->id,
			sectionID->idLen)) {
			return sec;
		}
		
		list = list->next;
	}

	return NULL;
}

int
SaCkptSectionRead(SaCkptReplicaT* replica,
	SaCkptSectionT* sec,
	SaSizeT offset, 
	SaSizeT* dataLength,
	void** data)
{
	if (offset > sec->dataLength[sec->dataIndex]) {
		data = NULL;
		*dataLength = 0;

		cl_log(LOG_ERR,
			"Section read failed, SA_ERR_INVALID_PARAM");

		return SA_ERR_INVALID_PARAM;
	}

	if ((offset + *dataLength) > 
		sec->dataLength[sec->dataIndex]) {
		*dataLength = 
			sec->dataLength[sec->dataIndex] - offset;
		
		cl_log(LOG_ERR,
			"Section read, read beyond the end of data");
	} 

	/* read from the end of the section */
	if (*dataLength == 0) {
		*data = NULL;
		cl_log(LOG_ERR,
			"Section read, read from the end of the section");
		
		return SA_ERR_FAILED_OPERATION;
	}

	*data = SaCkptMalloc(*dataLength);
	if (*data == NULL) {
		*dataLength = 0;
		
		cl_log(LOG_ERR,
			"Section read failed, SA_ERR_NO_MEMORY");
		
		return SA_ERR_NO_MEMORY;
	}
	
	memcpy(*data, 
		(char*)sec->data[sec->dataIndex] + offset,
		*dataLength);

	return SA_OK;
}


int 
SaCkptSectionCreate(SaCkptReplicaT* replica, 
	SaCkptReqSecCrtParamT* secCrtParam, 
	SaSizeT dataLength, void* data,
	SaCkptSectionT** pSec)
{
	SaCkptSectionT* sec = NULL;

	if (dataLength > replica->maxSectionSize) {
		cl_log(LOG_ERR,
			"Section create failed, section data too huge");
		return SA_ERR_FAILED_OPERATION;
	}

	if (replica->maxSectionNumber == replica->sectionNumber) {
		cl_log(LOG_ERR,
			"Section create failed, too many sections");
		return SA_ERR_FAILED_OPERATION;
	}

	/* if section exists, return error */
	sec = SaCkptSectionFind(replica, &(secCrtParam->sectionID));
	if (sec != NULL) {
		cl_log(LOG_ERR,
			"Section create failed, section %d already existed",
			(int)(*sec->sectionID.id));
		return SA_ERR_EXIST;
	}

	/* create section */
	sec = (SaCkptSectionT*)SaCkptMalloc(sizeof(SaCkptSectionT) + secCrtParam->sectionID.idLen);
	if (sec == NULL) {
		cl_log(LOG_ERR,
			"Section create failed, no memory");
		return SA_ERR_NO_MEMORY;
	}
	sec->replica = replica;
	
	sec->sectionID.idLen = secCrtParam->sectionID.idLen;
	memcpy(sec->sectionID.id,secCrtParam->sectionID.id,secCrtParam->sectionID.idLen);
	
	sec->expirationTime = secCrtParam->expireTime;
	sec->lastUpdateTime = time_longclock();
	sec->dataIndex = 0;
	sec->dataUpdateState = OP_STATE_COMMITTED;
	sec->sectionState = STATE_CREATE_PREPARED;
	sec->dataState = SA_CKPT_SECTION_VALID;
	sec->dataLength[0] = dataLength;
	if (dataLength > 0) {
		sec->data[0] = SaCkptMalloc(dataLength);
		if (sec->data[0] == NULL) {
			SaCkptFree((void*)&sec);
			cl_log(LOG_ERR,
				"Section create failed, no memory");
			return SA_ERR_NO_MEMORY;
		} else {
			memcpy(sec->data[0], data, dataLength);
		}
	} else {
		sec->data[0] = NULL;
	}
	sec->dataLength[1] = 0;
	sec->data[1] = NULL;

	/* add it to section list */
	replica->sectionList = g_list_append(
		replica->sectionList,
		(gpointer)sec);
	
	if (saCkptService->flagVerbose) {
		char* strSectionID = NULL;
		strSectionID = SaCkptSectionId2String(sec->sectionID);
		cl_log(LOG_INFO, 
			"Section %s created in replica %s",
			sec->sectionID.id,
			replica->checkpointName);
		SaCkptFree((void*)&strSectionID);
	}

	*pSec = sec;
	
	return SA_OK;
}


int
SaCkptSectionDelete(SaCkptReplicaT* replica, 
	SaCkptFixLenSectionIdT* sectionID)
{
	SaCkptSectionT* sec = NULL;

	sec = SaCkptSectionFind(replica, sectionID);

	if (sec == NULL) {
		cl_log(LOG_ERR,
			"Section delete failed, section does not exist");
		return SA_ERR_NOT_EXIST;
	}

	/* free section */
	if (sec->data[0] != NULL) {
		SaCkptFree((void**)&(sec->data[0]));
	}
	if (sec->data[1] != NULL) {
		SaCkptFree((void**)&(sec->data[1]));
	}
	SaCkptFree((void*)&sec);

	return HA_OK;
}

int
SaCkptSectionWrite(SaCkptReplicaT* replica,
	SaCkptSectionT* sec,
	SaSizeT offset, SaSizeT dataLength, void* data)
{
	int	index = 0;
	
	index = (sec->dataIndex + 1) % 2;
	
	if ((offset + dataLength) < sec->dataLength[sec->dataIndex]){
		sec->dataLength[index] = sec->dataLength[sec->dataIndex];
	} else {
		sec->dataLength[index] = offset + dataLength;
	}

	if (sec->dataLength[index] > replica->maxSectionSize) {
		cl_log(LOG_ERR,
			"Section write failed, section date too huge");
		return SA_ERR_FAILED_OPERATION;
	}

	if (sec->data[index] != NULL) {
		SaCkptFree((void**)&(sec->data[index]));
	}
	sec->data[index] = SaCkptMalloc(sec->dataLength[index]);
	if (sec->data[index] == NULL) {
		cl_log(LOG_ERR,
			"Section write failed, no memory");
		return SA_ERR_NO_MEMORY;
	}
	memcpy(sec->data[index], 
		sec->data[sec->dataIndex],
		sec->dataLength[sec->dataIndex]);
	memcpy((char*)(sec->data[index]) + offset,
		data, dataLength);

	sec->dataUpdateState = OP_STATE_PREPARED;

	return SA_OK;

}


int
SaCkptSectionOverwrite(SaCkptReplicaT* replica, 
	SaCkptSectionT* sec,
	SaSizeT dataLength, void* data)
{
	int	index = 0;

	if (dataLength > replica->maxSectionSize) {
		cl_log(LOG_ERR,
			"Section overwrite failed, section date too huge");
		return SA_ERR_FAILED_OPERATION;
	}

	index = (sec->dataIndex + 1) % 2;

	if (sec->data[index] != NULL) {
		SaCkptFree((void**)&(sec->data[index]));
	}
	sec->data[index] = SaCkptMalloc(dataLength);
	if (sec->data[index] == NULL) {
		cl_log(LOG_ERR,
			"Section overwrite failed, no memory");
		return SA_ERR_NO_MEMORY;
	}
	memcpy(sec->data[index], data,	dataLength);

	sec->dataUpdateState = OP_STATE_PREPARED;

	return SA_OK;
}

/* start replica retention timer */
void 
SaCkptReplicaStartTimer(SaCkptReplicaT* replica)
{
	replica->retentionTimeoutTag = 
		Gmain_timeout_add(
			replica->retentionTimeoutTag/1000000, 
			SaCkptRetentionTimeout, 
			(gpointer)replica);

	if (saCkptService->flagVerbose) {
		cl_log(LOG_INFO, 
			"Start retention timer %u for checkpoint %s",
			replica->retentionTimeoutTag,
			replica->checkpointName);
	}
	
	return;
}

/* stop replica retention timer */
void 
SaCkptReplicaStopTimer(SaCkptReplicaT* replica)
{
	if (saCkptService->flagVerbose) {
		cl_log(LOG_INFO, 
			"Delete retention timer %d for checkpoint %s",
			replica->retentionTimeoutTag,
			replica->checkpointName);
	}
	
	g_source_remove(
		replica->retentionTimeoutTag);
	replica->retentionTimeoutTag = 0;
	
	return;
}

/* after a node failure, remove it from node list */
void 
SaCkptReplicaNodeFailure(gpointer key, 
	gpointer value, 
	gpointer userdata)
{
	SaCkptReplicaT* replica;
	SaCkptOpenCheckpointT* openCkpt = NULL;
	SaCkptStateT* state;
	char* strNodeName = NULL;
	GList* list = NULL;

	replica = (SaCkptReplicaT*)value;
	strNodeName = (char*)userdata;

	list = replica->nodeList;
	while (list != NULL) {
		state = (SaCkptStateT*)list->data;
		if (!strcmp(state->nodeName, strNodeName)) {
			cl_log(LOG_INFO,
				"Replica %s, remove node %s from node list",
				replica->checkpointName,
				strNodeName);
			
			replica->nodeList = g_list_remove(
				replica->nodeList,
				(gpointer)state);
			break;
		}

		list = list->next;
	}

	if (replica->flagIsActive) {
		g_hash_table_foreach(replica->operationHash,
			SaCkptOperationNodeFailure,
			strNodeName);
	} else {
		/* 
		 * the other replica has dead, set the local
		 * replica as the active replica
		 *
		 * FIXME: how to choose active replica?
		 */

		cl_log(LOG_INFO, 
			"Replica %s, set myself as active",
			replica->checkpointName);
		
		replica->flagIsActive = TRUE;
		strcpy(replica->activeNodeName, 
			saCkptService->nodeName);

		/* update opened checkpoints */
		list = replica->openCheckpointList;
		while (list != NULL) {
			openCkpt = list->data;
			strcpy(openCkpt->activeNodeName,
				replica->activeNodeName);

			list = list->next;
		}

		replica->flagReplicaLock = FALSE;
		if (saCkptService->flagVerbose) {
			cl_log(LOG_INFO,
				"Replica %s unlocked",
				replica->checkpointName);
		}
	}

	return;
}


char* 
SaCkptSectionId2String(SaCkptFixLenSectionIdT sectionId)
{
	char* strSectionId = NULL, *pSec = NULL;
	int i = 0;
	strSectionId = (char*)SaCkptMalloc(sectionId.idLen * 4 + strlen("\0"));
	pSec = strSectionId;
	SACKPTASSERT(strSectionId != NULL);

	for(i =0; i< sectionId.idLen; i++)
	{
		sprintf(pSec, "%d",sectionId.id[i]);
		pSec ++;
	}
	strncat(pSec, "\0", strlen("\0"));

	return strSectionId;
}

void 
SaCkptDumpReplica(SaCkptReplicaT* replica){
	GList * list = NULL;
	SaCkptSectionT * section = NULL;
	int i = 0;
	cl_log(LOG_INFO, "\treplica info");
	cl_log(LOG_INFO, "\tname is %s \n",replica->checkpointName);
	cl_log(LOG_INFO, "\tsection number is %d \n",(int)replica->sectionNumber);
	list = replica->sectionList;
	
	while(list != NULL){
		section = (SaCkptSectionT *)list->data;
		cl_log(LOG_INFO, "\t %d section length is %d, name is %s \n",i++,section->sectionID.idLen,section->sectionID.id);
		list=list->next;
		
	}
}