ccmlib_clm.c

linux集群服务器软件代码包

{
	uint i;
	int n;
	char *p;
	const oc_ev_membership_t *oc = __ccm_data;

	for (i = 0, n = 0; i < oc->m_n_in; i++, n++) {
		MEMCHANGE(n) = SA_CLM_NODE_JOINED;
		MEMNODE(n).nodeId = oc->m_array[oc->m_in_idx+i].node_id;
		MEMNODE(n).member = 1;
		p = oc->m_array[oc->m_in_idx+i].node_uname;
		if (p) {
			strncpy(MEMNODE(n).nodeName.value, p,
					SA_MAX_NAME_LENGTH - 1);
			MEMNODE(n).nodeName.value[SA_MAX_NAME_LENGTH-1] = '\0';
		} else {
			MEMNODE(n).nodeName.value[0] = '\0';
		}
		set_misc_node_info(&MEMNODE(n));
	}
	for (i = 0; i < oc->m_n_out; i++, n++) {
		MEMCHANGE(n) = SA_CLM_NODE_LEFT;
		MEMNODE(n).nodeId = oc->m_array[oc->m_out_idx+i].node_id;
		MEMNODE(n).member = 0;
		p = oc->m_array[oc->m_out_idx+i].node_uname;
		if (p) {
			strncpy(MEMNODE(n).nodeName.value, p,
					SA_MAX_NAME_LENGTH - 1);
			MEMNODE(n).nodeName.value[SA_MAX_NAME_LENGTH-1] = '\0';
		} else {
			MEMNODE(n).nodeName.value[0] = '\0';
		}
		set_misc_node_info(&MEMNODE(n));
	}
}

SaErrorT
saClmDispatch(const SaClmHandleT *clmHandle, 
              SaDispatchFlagsT dispatchFlags)
{
	int ret;
	const oc_ev_membership_t *oc;
	uint itemnum;
	__clm_handle_t *hd = GET_CLM_HANDLE(clmHandle);

	if (!hd){
		return SA_ERR_BAD_HANDLE;
	}

	if ((ret = oc_ev_handle_event(hd->ev_token)) != 0) {
		if (ret == EINVAL){
			return SA_ERR_BAD_HANDLE;
		}

		/* else we must be evicted */
	}

	/* We did not lock for read here because other writers will set it
	 * with the same value (if there really exist some). Otherwise we
	 * need to lock here.
	 */
	if (__ccm_event == OC_EV_MS_EVICTED) {
		cl_log(LOG_WARNING
		,	"This node is evicted from the current partition!");
		return SA_ERR_LIBRARY;
	}
	if (__ccm_event == OC_EV_MS_NOT_PRIMARY
	||	__ccm_event == OC_EV_MS_PRIMARY_RESTORED) {
		cl_log(LOG_DEBUG, "Received not interested event [%d]"
		,	__ccm_event);
		return SA_OK;
	}
	if (!__ccm_data){
		return SA_ERR_INIT;
	}

	oc = __ccm_data;

	if(CLM_TRACK_STOP == hd->trackflags){
		return SA_OK;
	}

	/* SA_TRACK_CURRENT is cleared in saClmClusterTrackStart, hence we 
	 * needn't to deal with it now*/
	if (hd->trackflags & SA_TRACK_CHANGES) {
		itemnum = oc->m_n_member + oc->m_n_out;
		if (itemnum > hd->itemnum) {
			hd->callbacks.saClmClusterTrackCallback(hd->nbuf
			,	hd->itemnum, oc->m_n_member, oc->m_instance
			,	SA_ERR_NO_SPACE);
			return SA_OK;
		}
		pthread_lock();
		retrieve_changes_buffer(hd);
		pthread_unlock();
		hd->callbacks.saClmClusterTrackCallback(hd->nbuf, itemnum
		,	oc->m_n_member, oc->m_instance, SA_OK);
	} else if (hd->trackflags & SA_TRACK_CHANGES_ONLY) {
		itemnum = oc->m_n_in + oc->m_n_out;
		if (itemnum > hd->itemnum) {
			hd->callbacks.saClmClusterTrackCallback(hd->nbuf
			,	hd->itemnum, oc->m_n_member, oc->m_instance
			,	SA_ERR_NO_SPACE);
			return SA_OK;
		}
		pthread_lock();
		retrieve_changes_only_buffer(hd);
		pthread_unlock();
		hd->callbacks.saClmClusterTrackCallback(hd->nbuf, itemnum
		,	oc->m_n_member, oc->m_instance, SA_OK);

	} else {
		assert(0);
	}
	/* unlock */

	return SA_OK;
}

SaErrorT 
saClmFinalize(SaClmHandleT *clmHandle)
{
	gpointer hd, oldkey;

	if (g_hash_table_lookup_extended(__handle_hash, clmHandle
	,	&oldkey, &hd) == FALSE) {
		return SA_ERR_BAD_HANDLE;
	}
 
	oc_ev_unregister(((__clm_handle_t *)hd)->ev_token);
	/* TODO: unregister saClmClusterNodeGetCall here */

	g_free(hd);
	g_free(oldkey);

	return SA_OK;
}

SaErrorT 
saClmClusterTrackStart(const SaClmHandleT *clmHandle,
                       SaUint8T trackFlags,
                       SaClmClusterNotificationT *notificationBuffer,
                       SaUint32T numberOfItems)
{
	__clm_handle_t *hd = GET_CLM_HANDLE(clmHandle);

	if (!hd){
		return SA_ERR_BAD_HANDLE;
	}

	hd->trackflags = trackFlags;
	hd->itemnum = numberOfItems;
	hd->nbuf = notificationBuffer;

	if (trackFlags & SA_TRACK_CURRENT) {
		const oc_ev_membership_t *oc;
		SaUint32T itemnum;
		
		/* Clear SA_TRACK_CURRENT, it's no use since now. */
		hd->trackflags &= ~SA_TRACK_CURRENT;

		if (__ccm_data == NULL) {
			return SA_ERR_LIBRARY;
		}
		
		oc = __ccm_data;
		itemnum = oc->m_n_member;
		if (itemnum > numberOfItems) {
			hd->callbacks.saClmClusterTrackCallback(hd->nbuf
			,	hd->itemnum, oc->m_n_member, oc->m_instance
			,	SA_ERR_NO_SPACE);
			return SA_OK;
		}
		pthread_lock();
		retrieve_current_buffer(hd);
		pthread_unlock();
		hd->callbacks.saClmClusterTrackCallback(hd->nbuf, itemnum
		,	oc->m_n_member, oc->m_instance, SA_OK);
		return SA_OK;
	}

	return SA_OK;
}

SaErrorT 
saClmClusterTrackStop(const SaClmHandleT *clmHandle)
{
	__clm_handle_t *hd = GET_CLM_HANDLE(clmHandle);

	if (!hd){
		return SA_ERR_BAD_HANDLE;
	}

	/* This is ugly. But we currently depends on OCF interface, we have
	 * no choice. This should be fixed in the next version after we remove
	 * the dependency with OCF.
	 */
	hd->trackflags = CLM_TRACK_STOP;

	return SA_OK;
}

static SaErrorT
retrieve_node_buffer(SaClmNodeIdT nodeId, SaClmClusterNodeT *clusterNode)
{
	const oc_ev_membership_t *oc;
	uint i;
	char *p;

	oc = (const oc_ev_membership_t *)__ccm_data;

	for (i = 0; i < oc->m_n_member; i++) {
		if (oc->m_array[oc->m_memb_idx+i].node_id == nodeId) {
			clusterNode->nodeId = nodeId;
			clusterNode->member = 1;
			p = oc->m_array[oc->m_memb_idx+i].node_uname;
			if (p) {
				strncpy(clusterNode->nodeName.value, p,
						SA_MAX_NAME_LENGTH - 1);
				clusterNode->nodeName.value \
					[SA_MAX_NAME_LENGTH-1] = '\0';
			} else {
				clusterNode->nodeName.value[0] = '\0';
			}
			goto found;
		}
	}
	for (i = 0; i < oc->m_n_out; i++) {
		if (oc->m_array[oc->m_out_idx+i].node_id == nodeId) {
			clusterNode->nodeId = nodeId;
			clusterNode->member = 0;
			p = oc->m_array[oc->m_out_idx+i].node_uname;
			if (p) {
				strncpy(clusterNode->nodeName.value, p,
						SA_MAX_NAME_LENGTH - 1);
				clusterNode->nodeName.value \
					[SA_MAX_NAME_LENGTH-1] = '\0';
			} else {
				clusterNode->nodeName.value[0] = '\0';
			}
			goto found;
		}
	}
	cl_log(LOG_WARNING, "%s: no record for nodeId [%lu]"
	,	__FUNCTION__, nodeId);
	return SA_ERR_INVALID_PARAM;

found:
	set_misc_node_info(clusterNode);

	return SA_OK;
}

SaErrorT 
saClmClusterNodeGet(SaClmNodeIdT nodeId, SaTimeT timeout,
                    SaClmClusterNodeT *clusterNode)
{
	int i;
	SaErrorT ret;

	if (!clusterNode) {
		cl_log(LOG_ERR, "Invalid parameter clusterNode <%p>"
		,	clusterNode);
		return SA_ERR_INVALID_PARAM;
	}
	for (i = 0; i < timeout; i++) {
		if (__ccm_data){
			break;
		}
		sleep(1);
	}
	if (i == timeout){
		return SA_ERR_TIMEOUT;
	}

	pthread_lock();
	ret = retrieve_node_buffer(nodeId, clusterNode);
	pthread_unlock();
	return ret;
}

/*
 * This API is highly deprecated in version 1 implementation base on OCF.
 * It is actually _not_ an asynchronous call. TODO fix in version 2.
 */
SaErrorT
saClmClusterNodeGetAsync(const SaClmHandleT *clmHandle,
                         SaInvocationT invocation,
                         SaClmNodeIdT nodeId,
                         SaClmClusterNodeT *clusterNode)
{
	int ret;
	__clm_handle_t *hd = GET_CLM_HANDLE(clmHandle);

	if (!hd){
		return SA_ERR_BAD_HANDLE;
	}

	if (!clusterNode) {
		cl_log(LOG_ERR, "Invalid parameter clusterNode <%p>"
		,	clusterNode);
		return SA_ERR_INVALID_PARAM;
	}
	if (!__ccm_data) {
		cl_log(LOG_ERR, "__ccm_data is NULL");
		return SA_ERR_INIT;
	}
	pthread_lock();
	if ((ret = retrieve_node_buffer(nodeId, clusterNode)) != SA_OK) {
		cl_log(LOG_ERR, "retrieve_node_buffer error [%d]", ret);
		pthread_unlock();
		return ret;
	}
	pthread_unlock();

	hd->callbacks.saClmClusterNodeGetCallback(invocation, clusterNode
	,	SA_OK);

	return SA_OK;
}