lrmd.c

在LINUX下实现HA的源代码

	client = lookup_client(pid);
	if (NULL == client) {
		lrmd_log(LOG_ERR, -1,
			"on_connect_cbk: can not find client in client list");
		send_rc_msg(ch, HA_FAIL);
		return TRUE;
	}

	/*fill the channel of callback field*/
	client->ch_cbk = ch;
	send_rc_msg(ch, HA_OK);
	lrmd_log(LOG_INFO, -1, "on_connect_cbk: end.");
	return TRUE;
}

gboolean
on_recieve_cmd (IPC_Channel* ch, gpointer user_data)
{
	int i;
	lrmd_client_t* client = NULL;
	struct ha_msg* msg = NULL;
	const char* type = NULL;

	lrmd_log(LOG_INFO, 1, "on_recieve_cmd: start.");

	client = (lrmd_client_t*)user_data;
	if (IPC_DISCONNECT == ch->ch_status) {
		lrmd_log(LOG_INFO, -1, 
			"on_recieve_cmd: channel status is disconnect");
		return FALSE;
	}

	if (!ch->ops->is_message_pending(ch)) {
		lrmd_log(LOG_INFO, -1, "on_recieve_cmd: no pending message");
		return TRUE;
	}


	/*get the message */
	msg = msgfromIPC_noauth(ch);
	if (NULL == msg) {
		lrmd_log(LOG_ERR, -1, "on_recieve_cmd: can not recieve msg");
		return TRUE;
	}

	/*dispatch the message*/
	type = ha_msg_value(msg, F_LRM_TYPE);

	for (i=0; i<DIMOF(msg_maps); i++) {
		if (0 == strncmp(type, msg_maps[i].msg_type,
				 strlen(msg_maps[i].msg_type))) {

			/*call the handler of the message*/
			int rc = msg_maps[i].handler(client, msg);

			/*return rc to client if need*/
			if (msg_maps[i].need_return_rc) {
				send_rc_msg(ch, rc);
			}
			break;
		}
	}
	if (i == DIMOF(msg_maps)) {
		lrmd_log(LOG_INFO, 0, "on_recieve_cmd: unknown msg");
	}

	/*delete the msg*/
	ha_msg_del(msg);

	lrmd_log(LOG_INFO, -1, "on_recieve_cmd: end.");

	return TRUE;
}

void
on_remove_client (gpointer user_data)
{
	lrmd_client_t* client = NULL;

	lrmd_log(LOG_INFO, 1, "on_remove_client: start.");
	client = (lrmd_client_t*) user_data;

	g_free(client->app_name);
	g_free(client);

	lrmd_log(LOG_INFO, -1, "on_remove_client: end.");
}
gboolean
on_idle (gpointer data)
{
	GList* node;

	/* check whether some running operations finished in idle */ 
	for(node=g_list_first(rsc_list); NULL!=node; node=g_list_next(node)){
		lrmd_rsc_t* rsc = (lrmd_rsc_t*)node->data;
		if (NULL != rsc->op_list) {
			GList* first = g_list_first(rsc->op_list);
			lrmd_op_t* op = first->data;
			check_op(op);
		}
	}
	return TRUE;
}

gboolean
on_timeout_op_done(gpointer data)
{
	lrmd_op_t* op = NULL;
	lrmd_rsc_t* rsc = NULL;

	lrmd_log(LOG_INFO, 1, "on_timeout_op_done: start.");

	/*this operation is timeout*/
	op = (lrmd_op_t*)data;
	if (HA_FAIL==ha_msg_add_int(op->msg, F_LRM_OPSTATUS, LRM_OP_TIMEOUT)) {
		lrmd_log(LOG_ERR,0,
			"on_timeout_op_done: can not add opstatus to msg");
	}
	rsc = op->rsc;
	op_done(op);
	perform_op(rsc);
	lrmd_log(LOG_INFO, -1, "on_timeout_op_done: end.");
	return TRUE;
}

gboolean
on_timeout_monitor(gpointer data)
{
	lrmd_mon_t* mon = NULL;
	lrmd_op_t* op = NULL;
	int timeout = 0;

	lrmd_log(LOG_INFO, 1, "on_timeout_monitor: start.");
	mon = (lrmd_mon_t*)data;
	mon->pending_op++;
	/* create a op */ 
	op = g_new(lrmd_op_t, 1);
	op->call_id = mon->call_id;
	op->callback_id = -1;
	op->client = NULL;
	op->timeout_tag = 0;
	op->rsc = mon->rsc;
	op->mon	= mon;
	op->app_name = mon->app_name;
	op->msg = ha_msg_copy(mon->msg);
	mon->rsc->op_list = g_list_append(mon->rsc->op_list, op);

	if (HA_FAIL == ha_msg_add(op->msg, F_LRM_APP, mon->app_name)) {
		lrmd_log(LOG_ERR, 0,"on_msg_perform_op: can not add app_name.");
	}
	
	ha_msg_value_int(op->msg, F_LRM_TIMEOUT, &timeout);
	if( 0 < timeout ) {
		op->timeout_tag = g_timeout_add(timeout*1000, 
						on_timeout_op_done, op);
	}

	perform_op(mon->rsc);
	lrmd_log(LOG_INFO, -1, "on_timeout_monitor: end.");
	return TRUE;
}

/*LRM Message Handlers*/
int
on_msg_register(lrmd_client_t* client, struct ha_msg* msg)
{
	lrmd_client_t* exist = NULL;
	const char* app_name = NULL;

	lrmd_log(LOG_INFO, 1, "on_msg_register: start.");

	app_name = ha_msg_value(msg, F_LRM_APP);
	if (NULL == app_name) {
		lrmd_log(LOG_ERR, -1, "on_msg_register: app_name is null.");
		return HA_FAIL;
	}
	client->app_name = g_strdup(app_name);

	if (HA_OK != ha_msg_value_int(msg, F_LRM_PID, &client->pid)) {
		lrmd_log(LOG_ERR, -1, 
			"on_msg_register: can not find pid field.");
		return HA_FAIL;
	}

	if (HA_OK != ha_msg_value_int(msg, F_LRM_GID, &client->gid)) {
		lrmd_log(LOG_ERR, -1, 
			"on_msg_register: can not find gid field.");
		return HA_FAIL;
	}

	if (HA_OK != ha_msg_value_int(msg, F_LRM_UID, &client->uid)) {
		lrmd_log(LOG_ERR, -1, 
			"on_msg_register: can not find uid field.");
		return HA_FAIL;
	}

	exist = lookup_client(client->pid);
	if (NULL != exist) {
		client_list = g_list_remove(client_list, exist);
		on_remove_client(exist);
		lrmd_log(LOG_ERR, 0, 
			"on_msg_register: client exist, remove first.");

	}

	client_list = g_list_append (client_list, client);
	lrmd_log(LOG_INFO, -1, "on_msg_register: end.");
	return HA_OK;
}

int
on_msg_unregister(lrmd_client_t* client, struct ha_msg* msg)
{
	GList* rsc_node;
	GList* mon_node = NULL;
	GList* op_node = NULL;
	lrmd_rsc_t* rsc = NULL;

	lrmd_log(LOG_INFO, 1, "on_msg_unregister: start.");

	if (NULL == client_list || NULL == lookup_client(client->pid)) {
		lrmd_log(LOG_ERR, -1, 
			"on_msg_unregister: can not find the client.");
		return HA_FAIL;
	}
	/* remove from client_list */ 
	client_list = g_list_remove(client_list, client);
	/* remove all monitors and pending ops */ 
	for(rsc_node = g_list_first(rsc_list);
		NULL != rsc_node; rsc_node = g_list_next(rsc_node)){
		rsc = (lrmd_rsc_t*)rsc_node->data;
		/* remove monitors belong to this client */ 
		mon_node = g_list_first(rsc->mon_list);
		while (NULL != mon_node) {
			lrmd_mon_t* mon = (lrmd_mon_t*)mon_node->data;
			if (mon->client == client) {
				mon_node = g_list_next(mon_node);
				rsc->mon_list =
					 g_list_remove(rsc->mon_list, mon);
				free_mon(mon);
			}
			else {
				mon_node = g_list_next(mon_node);
			}

		}
		/* remove pending ops belong to this client */ 
		op_node = g_list_first(rsc->op_list);
		op_node = g_list_next(op_node);
		while (NULL != op_node) {
			lrmd_op_t* op = (lrmd_op_t*)op_node->data;
			if (op->client == client) {
				op_node = g_list_next(op_node);
				rsc->op_list = g_list_remove(rsc->op_list, op);
				ha_msg_del(op->msg);
				g_free(op);
			}
			else {
				op_node = g_list_next(op_node);
			}

		}
	}
	lrmd_log(LOG_INFO, -1, "on_msg_unregister: end.");
	return HA_OK;
}

int
on_msg_get_ra_types(lrmd_client_t* client, struct ha_msg* msg)
{
	struct ha_msg* ret = NULL;

	lrmd_log(LOG_INFO, 1, "on_msg_get_rsc_types: start.");

	ret = create_lrm_ret(HA_OK, 4);
	if (NULL == ret) {
		lrmd_log(LOG_ERR, -1, 
			"on_msg_get_rsc_types: can not create msg.");
		return HA_FAIL;
	}

	ha_msg_add_list(ret,F_LRM_RTYPE,ra_list);

	if (HA_OK != msg2ipcchan(ret, client->ch_cmd)) {
		lrmd_log(LOG_ERR, 0, 
			"on_msg_get_rsc_types: can not send the ret msg");
	}
	ha_msg_del(ret);

	lrmd_log(LOG_INFO, -1, "on_msg_get_rsc_types: end.");
	return HA_OK;
}
int
on_msg_get_all(lrmd_client_t* client, struct ha_msg* msg)
{
	int i = 1;
	char value[UUID_SLEN];
	char key[MAX_NAME_LEN];
	struct ha_msg* ret = NULL;
	GList* node;

	lrmd_log(LOG_INFO, 1, "on_msg_get_all: start.");

	ret = create_lrm_ret(HA_OK, g_list_length(rsc_list) + 1);
	if (NULL == ret) {
		lrmd_log(LOG_ERR, -1, "on_msg_get_all: can not create msg.");
		return HA_FAIL;
	}

	for(node=g_list_first(rsc_list); NULL!=node; node=g_list_next(node)) {
		lrmd_rsc_t* rsc = (lrmd_rsc_t*)node->data;
		uuid_unparse(rsc->id, value);
		snprintf(key,MAX_NAME_LEN,"%s%d",F_LRM_RID,i);
		ha_msg_add(ret,key,value);
		i++;
	}

	if (HA_OK != msg2ipcchan(ret, client->ch_cmd)) {
		lrmd_log(LOG_ERR, 0,"on_msg_get_all: can not send the ret msg");
	}
	ha_msg_del(ret);

	lrmd_log(LOG_INFO, -1, "on_msg_get_all: end.");
	return HA_OK;
}
int
on_msg_get_rsc(lrmd_client_t* client, struct ha_msg* msg)
{
	rsc_id_t id;
	struct ha_msg* ret = NULL;
	lrmd_rsc_t* rsc = NULL;
	lrmd_log(LOG_INFO, 1, "on_msg_get_rsc: start.");

	ha_msg_value_uuid(msg,F_LRM_RID,id);
	rsc = lookup_rsc(id);

	if (NULL == rsc) {
		lrmd_log(LOG_INFO, 0, "on_msg_get_rsc: no rsc with such id.");
		ret = create_lrm_ret(HA_FAIL, 1);
		if (NULL == ret) {
			lrmd_log(LOG_ERR, -1,
				"on_msg_get_rsc: can not create msg.");
			return HA_FAIL;
		}
	}
	else {
		ret = create_lrm_ret(HA_OK, 5);
		if (NULL == ret) {
			lrmd_log(LOG_ERR, -1, 
				"on_msg_get_rsc: can not create msg.");
			return HA_FAIL;
		}
		if (HA_FAIL == ha_msg_add_uuid(ret, F_LRM_RID, rsc->id)) {
			return HA_FAIL;
		}

		if (HA_FAIL == ha_msg_add(ret, F_LRM_RNAME, rsc->name)) {
			return HA_FAIL;
		}

		if (HA_FAIL == ha_msg_add(ret, F_LRM_RTYPE, rsc->type))	{
			return HA_FAIL;
		}

		if (NULL != rsc->params) {
			char* param_str = hash_table_to_string(rsc->params);
			if (HA_FAIL==ha_msg_add(ret, F_LRM_PARAM, param_str)){
				return HA_FAIL;
			}
		}
	}
	if (HA_OK != msg2ipcchan(ret, client->ch_cmd)) {
		lrmd_log(LOG_ERR, 0,"on_msg_get_rsc: can not send the ret msg");
	}
	ha_msg_del(ret);

	lrmd_log(LOG_INFO, -1, "on_msg_get_rsc: end.");
	return HA_OK;
}
int
on_msg_del_rsc(lrmd_client_t* client, struct ha_msg* msg)
{
	rsc_id_t id;
	lrmd_rsc_t* rsc = NULL;
	lrmd_mon_t* mon = NULL;
	GList* mon_node = NULL;
	GList* op_node = NULL;

	lrmd_log(LOG_INFO, 1, "on_msg_del_rsc: start.");

	ha_msg_value_uuid(msg,F_LRM_RID,id);
	rsc = lookup_rsc(id);

	if (NULL == rsc) {
		lrmd_log(LOG_INFO, -1, "on_msg_del_rsc: no rsc with such id.");
		return HA_FAIL;
	}
	else {
		rsc_list = g_list_remove(rsc_list, rsc);
		mon_node = g_list_first(rsc->mon_list);
		while (NULL != mon_node) {
			mon = (lrmd_mon_t*)mon_node->data;
			if (mon->client == client) {
				mon_node = g_list_next(mon_node);
				rsc->mon_list = 
					g_list_remove(rsc->mon_list, mon);
				free_mon(mon);
			}
			else {
				mon_node = g_list_next(mon_node);
			}

		}
		/* remove pending ops */ 
		op_node = g_list_first(rsc->op_list);
		if (NULL == op_node) {
			/* no ops, just remove the resource. */ 
			free_rsc(rsc);
		}
		else {
			/* the first op is running, so skip it */ 
			/* and remove others. */ 
			/* when the running op done,  */ 
			/* it will release the memory of rsc. */ 
			op_node = g_list_next(op_node);
			while (NULL != op_node) {
				lrmd_op_t* op = (lrmd_op_t*)op_node->data;
				if (op->client == client) {
					op_node = g_list_next(op_node);
					rsc->op_list =
						g_list_remove(rsc->op_list, op);
					ha_msg_del(op->msg);
					g_free(op);
				}
				else {
					op_node = g_list_next(op_node);
				}
			}
		}
	}

	lrmd_log(LOG_INFO, -1, "on_msg_del_rsc: end.");
	return HA_OK;
}

int
on_msg_add_rsc(lrmd_client_t* client, struct ha_msg* msg)
{
	GList* node;
	rsc_id_t id;
	lrmd_rsc_t* rsc = NULL;
	char* params = NULL;
	gboolean ra_type_exist = FALSE;

	lrmd_log(LOG_INFO, 1, "on_msg_add_rsc: start.");

	ha_msg_value_uuid(msg,F_LRM_RID,id);
	if (NULL != lookup_rsc(id)) {
		lrmd_log(LOG_ERR, -1,
				"on_msg_add_rsc: same id resource exists.");
		return HA_FAIL;
	}

	rsc = g_new(lrmd_rsc_t,1);
	uuid_copy(rsc->id,id);
	rsc->name = g_strdup(ha_msg_value(msg, F_LRM_RNAME));
	rsc->type = g_strdup(ha_msg_value(msg, F_LRM_RTYPE));

	for(node=g_list_first(ra_list); NULL!=node; node=g_list_next(node)){
		char* type = (char*)node->data;
		if (0 == strcmp(type, rsc->type)) {
			ra_type_exist = TRUE;
			break;
		}
	}
	if (!ra_type_exist) {
		g_free(rsc);
		lrmd_log(LOG_ERR, -1,
				"on_msg_add_rsc: ra type does not exist.");
		return HA_FAIL;
	}
	rsc->params = NULL;
	rsc->op_list = NULL;
	rsc->mon_list = NULL;
	rsc->last_op = NULL;
	params = g_strdup(ha_msg_value(msg, F_LRM_PARAM));
	if (NULL != params) {
		rsc->params = string_to_hash_table(params);
	}
/*	rsc->params = ha_msg_value_hash_table(msg, F_LRM_PARAM);*/
	rsc_list = g_list_append(rsc_list, rsc);


	lrmd_log(LOG_INFO, -1, "on_msg_add_rsc: end.");
	return HA_OK;
}

int
on_msg_perform_op(lrmd_client_t* client, struct ha_msg* msg)
{
	int timeout = 0;
	rsc_id_t id;
	lrmd_rsc_t* rsc = NULL;
	lrmd_op_t* op = NULL;
	const char* type = NULL;

	lrmd_log(LOG_INFO, 1, "on_msg_perform_op: start.");

	ha_msg_value_uuid(msg,F_LRM_RID,id);
	rsc = lookup_rsc(id);
	if (NULL == rsc) {
		lrmd_log(LOG_ERR, -1, 
			"on_msg_perform_op: no rsc with such id.");
		return HA_FAIL;
	}

	call_id++;
	if (HA_FAIL == ha_msg_add_int(msg, F_LRM_CALLID, call_id)) {
		lrmd_log(LOG_ERR, -1, "on_msg_perform_op: can not add callid.");
		return HA_FAIL;
	}

	type = ha_msg_value(msg, F_LRM_TYPE);
	/* when a flush request arrived, flush all pending ops */ 
	if (0 == strncmp(type, FLUSHOPS, strlen(FLUSHOPS))) {
		GList* node = g_list_first(rsc->op_list);
		while (NULL != node ) {
			lrmd_op_t* op = (lrmd_op_t*)node->data;
			node = g_list_next(node);
			rsc->op_list = g_list_remove(rsc->op_list, op);
			flush_op(op);
		}
	}
	else {
		op = g_new(lrmd_op_t, 1);
		op->call_id = call_id;
		op->callback_id = -1;
		op->client = client;
		op->timeout_tag = 0;
		op->rsc = rsc;
		op->mon	= NULL;
		op->app_name = client->app_name;
		op->msg = ha_msg_copy(msg);
		rsc->op_list = g_list_append(rsc->op_list, op);
		if (HA_FAIL==ha_msg_add(op->msg, F_LRM_APP, client->app_name)) {
			lrmd_log(LOG_ERR, 0, 
				"on_msg_perform_op: can not add app_name.");
		}
		ha_msg_value_int(op->msg, F_LRM_TIMEOUT, &timeout);
		if (0 < timeout ) {
			op->timeout_tag = g_timeout_add(timeout*1000, 
						on_timeout_op_done, op);
		}
		perform_op(rsc);
	}

	lrmd_log(LOG_INFO, -1, "on_msg_perform_op: end.");
	return call_id;
}