lrmd.c

在LINUX下实现HA的源代码

int
on_msg_set_monitor(lrmd_client_t* client, struct ha_msg* msg)
{
	rsc_id_t id;
	lrmd_rsc_t* rsc = NULL;
	mon_mode_t mode;

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

	ha_msg_value_uuid(msg,F_LRM_RID,id);
	rsc = lookup_rsc(id);
	if (NULL == rsc) {
		lrmd_log(LOG_ERR, -1, 
			"on_msg_set_monitor: 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_set_monitor: can not add callid.");
		return HA_FAIL;
	}

	/* if the monitor mode is clear, remove all monitors on the resource. */ 
	if (HA_FAIL == ha_msg_value_int(msg, F_LRM_MONMODE, (int*)&mode)) {
		lrmd_log(LOG_ERR, -1, 
			"on_msg_set_monitor: can not get monitor mode.");
		return HA_FAIL;
	}
	if (LRM_MONITOR_CLEAR == mode) {
		GList* first = g_list_first(rsc->mon_list);
		while (NULL != first) {
			lrmd_mon_t* mon = (lrmd_mon_t*)first->data;
			rsc->mon_list = g_list_remove(rsc->mon_list, mon);
			free_mon(mon);
			first = g_list_first(rsc->mon_list);
		}
	}
	else {
	/* otherwise, create a mon object */ 
		lrmd_mon_t*	mon = g_new(lrmd_mon_t, 1);
		mon->mode = mode;
		mon->rsc = rsc;
		mon->call_id = call_id;
		mon->client = client;
		mon->app_name = client->app_name;
		mon->pending_op = 0;
		mon->is_deleted = FALSE;
		mon->last_status = -1;
		if (HA_FAIL == ha_msg_value_int(msg, F_LRM_MONINTVL, 
						&mon->interval)) {
			g_free(mon);
			lrmd_log(LOG_ERR, -1,
				"on_msg_set_monitor: can not get interval.");
			return HA_FAIL;
		}
		if (0 >= mon->interval) {
			g_free(mon);
			lrmd_log(LOG_ERR, -1, 
				"on_msg_set_monitor: interal less 1 second.");
			return HA_FAIL;

		}
		if (HA_FAIL == ha_msg_value_int(msg, F_LRM_MONTGT, 
						&mon->target)) {
			g_free(mon);
			lrmd_log(LOG_ERR, -1,
				"on_msg_set_monitor: can not get target.");
			return HA_FAIL;
		}
		mon->msg = ha_msg_copy(msg);
		/* add a time GSource to g_loop */ 
		mon->timeout_tag = g_timeout_add(mon->interval*1000,
						 on_timeout_monitor, mon);
		/* insert the monitor to the list of resource */ 
		rsc->mon_list = g_list_append(rsc->mon_list, mon);
	}

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

int
on_msg_get_state(lrmd_client_t* client, struct ha_msg* msg)
{
	rsc_id_t id;
	lrmd_rsc_t* rsc = NULL;
	GList* node;
	lrmd_op_t* op = NULL;
	struct ha_msg* op_msg = NULL;
	struct ha_msg* ret = NULL;

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

	ha_msg_value_uuid(msg,F_LRM_RID,id);
	rsc = lookup_rsc(id);
	if (NULL == rsc) {
		lrmd_log(LOG_ERR, -1, "on_msg_get_state: no rsc with such id.");
		send_rc_msg(client->ch_cmd, HA_FAIL);
		return HA_FAIL;
	}
	if ( NULL == rsc->op_list )
	{
		if (NULL != rsc->last_op) {
			ret = op_to_msg(rsc->last_op);
		}
		if (NULL == ret) {
			ret = ha_msg_new(5);
		}

		if (HA_FAIL == ha_msg_add_int(ret, F_LRM_STATE, LRM_RSC_IDLE)) {
			lrmd_log(LOG_ERR, -1, 
				"on_msg_get_state: can not add state to msg.");
			ha_msg_del(ret);
			return HA_FAIL;
		}
		if (HA_OK != msg2ipcchan(ret, client->ch_cmd)) {
			lrmd_log(LOG_ERR, 0, 
				"on_msg_get_state: can not send the ret msg");
		}
		ha_msg_del(ret);

	}
	else {
		int op_count = 0;
		struct ha_msg* ret = ha_msg_new(5);

		if (HA_FAIL == ha_msg_add_int(ret, F_LRM_STATE, LRM_RSC_BUSY)) {
			lrmd_log(LOG_ERR, -1, 
				"on_msg_get_state: can not add state to msg.");
			ha_msg_del(ret);
			return HA_FAIL;
		}
		op_count = g_list_length(rsc->op_list);
		if (HA_FAIL == ha_msg_add_int(ret, F_LRM_OPCNT, op_count)) {
			lrmd_log(LOG_ERR, -1, 
				"on_msg_get_state: can not add state count.");
			ha_msg_del(ret);
			return HA_FAIL;
		}
		if (HA_OK != msg2ipcchan(ret, client->ch_cmd)) {
			lrmd_log(LOG_ERR, -1, 
				"on_msg_get_state: can not send the ret msg");
			ha_msg_del(ret);
			return HA_FAIL;
		}
		ha_msg_del(ret);

		for(node = g_list_first(rsc->op_list);
			NULL != node; node = g_list_next(node)){

			op = (lrmd_op_t*)node->data;
			op_msg = op_to_msg(op);
			if ( NULL == op_msg  ) {
				lrmd_log(LOG_ERR, 0, 
					"on_msg_get_state: can not add op.");
				ha_msg_del(op_msg);
				continue;
			}
			if (HA_OK != msg2ipcchan(op_msg, client->ch_cmd)) {
				lrmd_log(LOG_ERR, 0,
					"on_msg_get_state: can not send msg");
			}
			ha_msg_del(op_msg);
		}
	}
	lrmd_log(LOG_INFO, -1, "on_msg_get_state: end.");
	return HA_OK;
}
/* ********************** op functions ********************** */
int
check_op(lrmd_op_t* op)
{
	int rc;
	char* data=NULL;
	struct RAExecOps * RAExec = NULL;
	int ret = NULL;
	lrmd_rsc_t* rsc = NULL;

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

	RAExec = g_hash_table_lookup(RAExecFuncs,op->rsc->type);

	if (NULL == RAExec) {
		lrmd_log(LOG_ERR,-1,"check_op: can not find RAExec");
		return HA_FAIL;
	}
	if ( 0 > op->callback_id ) {
		if (HA_FAIL == ha_msg_add_int(op->msg, F_LRM_OPSTATUS,
						LRM_OP_ERROR)) {
			lrmd_log(LOG_ERR,-1,
				"check_op: can not add opstatus to msg");
			return HA_FAIL;
		}
		rsc = op->rsc;
		op_done(op);
		perform_op(rsc);
		lrmd_log(LOG_INFO, -1, "check_op: end.");
		return HA_OK;
	}
	ret = check_ra_rc(RAExec,op->callback_id, &rc, &data);
	if ( 0 < ret ) {
		if (HA_FAIL == ha_msg_add_int(op->msg, F_LRM_RC, rc)) {
			lrmd_log(LOG_ERR,-1,"check_op: can not add rc to msg");
			return HA_FAIL;
		}
		if (HA_FAIL == ha_msg_add_int(op->msg, F_LRM_OPSTATUS, 
						LRM_OP_DONE)) {
			lrmd_log(LOG_ERR,-1,
				"check_op: can not add opstatus to msg");
			return HA_FAIL;
		}
		if (NULL != data) {
			int ret = ha_msg_addbin(op->msg, F_LRM_DATA,data, 
						strlen(data));
			if (HA_FAIL == ret) {
				lrmd_log(LOG_ERR,-1,
					"check_op: can not add data to msg");
				return HA_FAIL;
			}
		}
		rsc = op->rsc;
		op_done(op);
		perform_op(rsc);
	}
	lrmd_log(LOG_INFO, -1, "check_op: end.");
	return HA_OK;
}
int
op_done(lrmd_op_t* op)
{
	lrmd_mon_t* mon = NULL;

	lrmd_log(LOG_INFO, 1, "op_done: start.");
	/*  we should check if the resource exists. */
	if (NULL == g_list_find(rsc_list, op->rsc)) {
		if( op->timeout_tag > 0 ) {
			g_source_remove(op->timeout_tag);
		}
		/* delete the op */
		ha_msg_del(op->msg);
		g_free(op);

		lrmd_log(LOG_INFO,-1,
			"op_done: the resource of this op does not exists");
		return HA_FAIL;

	}

	/* if the op is create by client */
	if (NULL != op->client) {
		/* send the result to client */
		lrmd_log(LOG_INFO, 0, "op_done: a normal op done.");
		/* we have to check whether the client still exists */
		/* for the client may signoff during the op running. */
		if (NULL != g_list_find(client_list, op->client)) {
			/* the client still exists */
			if (NULL == op->client->ch_cbk) {
				lrmd_log(LOG_ERR, 0, 
					"op_done: client->ch_cbk is null");
			}
			else
			if (HA_OK != msg2ipcchan(op->msg, op->client->ch_cbk)) {
				lrmd_log(LOG_ERR, 0, 
					"op_done: can not send the ret msg");
			}

		}
		/* release the old last_op */
		if (NULL != op->rsc->last_op) {
			ha_msg_del(op->rsc->last_op->msg);
			g_free(op->rsc->last_op);
		}
		/* remove the op from op_list and assign to last_op */
		op->rsc->op_list = g_list_remove(op->rsc->op_list,op);
		op->rsc->last_op = op;
		if( op->timeout_tag > 0 ) {
			g_source_remove(op->timeout_tag);
		}

	}
	else {
	/* if the op is created by monitor */
		lrmd_log(LOG_INFO, 0, "op_done: a monitor op done.");

		mon = op->mon;
		mon->pending_op--;
		if (!mon->is_deleted) {
			/* check status */
			op_status_t status = LRM_OP_ERROR;
			gboolean need_send = FALSE;
			int rc = -1;

			ha_msg_value_int(op->msg,F_LRM_OPSTATUS,(int*)&status);
			ha_msg_value_int(op->msg, F_LRM_RC, &rc);

			if (LRM_OP_TIMEOUT == status||LRM_OP_ERROR == status) {
				need_send = TRUE;
			}
			else
			if (LRM_OP_DONE == status) {
				if ((LRM_MONITOR_SET == mon->mode &&
				     rc == mon->target &&
				     mon->last_status != rc) ||
				    (LRM_MONITOR_CHANGE == mon->mode &&
				     rc != mon->last_status)) {
					need_send = TRUE;
				}
				mon->last_status = rc;
			}
			/* send monitor msg to client */
			if (need_send) {
				if (NULL == mon->client->ch_cbk) {
					lrmd_log(LOG_ERR, 0, 
						"op_done: ch_cbk is null");
				}
				else
				if (HA_OK != msg2ipcchan(op->msg, 
						mon->client->ch_cbk)) {
					lrmd_log(LOG_ERR, 0, 
						"op_done: can not send msg");
				}
			}

		}
		else {
			/* delete the monitor */
			if (0 == mon->pending_op) {
				ha_msg_del(mon->msg);
				g_free(mon);
			}
		}
		/* remove timeout source */
		if( op->timeout_tag > 0 ) {
			g_source_remove(op->timeout_tag);
		}
		/* delete the op */
		op->rsc->op_list = g_list_remove(op->rsc->op_list,op);
		ha_msg_del(op->msg);
		g_free(op);
	}
	lrmd_log(LOG_INFO, -1, "op_done: end.");
	return HA_OK;
}
int
flush_op(lrmd_op_t* op)
{
	lrmd_log(LOG_INFO, 1, "flush_op: start.");
	if (HA_FAIL == ha_msg_add_int(op->msg, F_LRM_RC, HA_FAIL)) {
		lrmd_log(LOG_ERR,-1,"flush_op: can not add rc ");
		return HA_FAIL;
	}

	if (HA_FAIL==ha_msg_add_int(op->msg,F_LRM_OPSTATUS,LRM_OP_CANCELLED)) {
		lrmd_log(LOG_ERR,-1,"flush_op: can not add op status");
		return HA_FAIL;
	}

	op_done(op);

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

int
perform_op(lrmd_rsc_t* rsc)
{
	struct RAExecOps * RAExec = NULL;
	GList* first = NULL;
	lrmd_op_t* op = NULL;

	lrmd_log(LOG_INFO, 1, "perform_op: start.");
	if (NULL == g_list_find(rsc_list, rsc)) {

		lrmd_log(LOG_INFO,-1,
			"op_done: the resource of this op does not exists");
		return HA_FAIL;

	}
	if (NULL == rsc->op_list) {
		lrmd_log(LOG_INFO,-1,"perform_op: no op to perform");
		return HA_OK;
	}
	first = g_list_first(rsc->op_list);
	op = first->data;
	if (-1 != op->callback_id )	{
		lrmd_log(LOG_INFO,-1,"perform_op: current op is performing");
		return HA_OK;
	}
	RAExec = g_hash_table_lookup(RAExecFuncs,op->rsc->type);
	if (NULL == RAExec) {
		lrmd_log(LOG_ERR,-1,"check_op: can not find RAExec");
		return HA_FAIL;
	}
	op->callback_id = perform_ra_op(RAExec, op);

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

struct ha_msg*
op_to_msg(lrmd_op_t* op)
{
	struct ha_msg* msg = NULL;

	lrmd_log(LOG_INFO, 1, "op_to_msg: start.");
	msg = ha_msg_copy(op->msg);
	if (NULL == msg) {
		lrmd_log(LOG_ERR,-1,"op_to_msg: can not copy the msg");
		return NULL;
	}
	if (HA_FAIL == ha_msg_add_int(msg, F_LRM_CALLID, op->call_id)) {
		ha_msg_del(msg);
		lrmd_log(LOG_ERR,-1,"op_to_msg: can not add call_id");
		return NULL;
	}
	if (HA_FAIL == ha_msg_add(msg, F_LRM_APP, op->app_name)) {
		ha_msg_del(msg);
		lrmd_log(LOG_ERR,-1,"op_to_msg: can not add app_name");
		return NULL;
	}
	lrmd_log(LOG_INFO, -1, "op_to_msg: end.");
	return msg;
}

/* ***************************** RA wrap funcs ***************************** */
int
check_ra_rc(struct RAExecOps * RAExec ,int callback_id, int* rc, char** data)
{
	int ret = 0;

	lrmd_log(LOG_INFO, 1, "check_ra_rc: start.");
	ret = RAExec->post_query_result(callback_id, rc, data);
	*rc = *rc / 256;

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

	return ret;
}
int
perform_ra_op(struct RAExecOps * RAExec, lrmd_op_t* op)
{
	int key, ret;
	GHashTable* params_table = NULL;
	char* params = NULL;
	char* rsc_name = NULL;
	const char* op_type = NULL;
	const char* temp_params = NULL;

	lrmd_log(LOG_INFO, 1, "perform_ra_op: start.");
	rsc_name = op->rsc->name;
	op_type = ha_msg_value(op->msg, F_LRM_OP);
	temp_params = ha_msg_value(op->msg, F_LRM_PARAM);
	if (NULL != temp_params) {
		params = g_strdup(temp_params);
		params_table = string_to_hash_table(params);
	}
	ret=RAExec->execra(rsc_name, op_type, params_table, NULL, TRUE, &key);

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

	return ret == 0 ? key : ret;
}
/* ***************************** Util Functions ***************************** */
int
send_rc_msg (IPC_Channel* ch, int rc)
{
	struct ha_msg* ret = NULL;

	lrmd_log(LOG_INFO, 1, "send_rc_msg: start.");
	ret = create_lrm_ret(rc, 1);
	if (NULL == ret) {
		lrmd_log(LOG_ERR, -1, "send_rc_msg: can not create ret msg");
		return HA_FAIL;
	}

	if (HA_OK != msg2ipcchan(ret, ch)) {
		lrmd_log(LOG_ERR, 0, "send_rc_msg: can not send the ret msg");
	}
	ha_msg_del(ret);
	lrmd_log(LOG_INFO, -1, "send_rc_msg: end.");
	return HA_OK;
}

lrmd_client_t*
lookup_client (pid_t pid)
{
	GList* node;

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

	for(node = g_list_first(client_list);
		NULL != node; node = g_list_next(node)){
		lrmd_client_t* client = (lrmd_client_t*)node->data;
		if (pid == client->pid) {
			lrmd_log(LOG_INFO, -1, "lookup_client: end.");
			return client;
		}
	}

	lrmd_log(LOG_INFO, -1, "lookup_client: end.");
	return NULL;
}

lrmd_rsc_t*
lookup_rsc (rsc_id_t rid)
{
	GList* node;

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

	for(node=g_list_first(rsc_list); NULL!=node; node=g_list_next(node)){
		lrmd_rsc_t* rsc = (lrmd_rsc_t*)node->data;
		if (0 == uuid_compare(rid,rsc->id)) {
			lrmd_log(LOG_INFO, -1, "lookup_rsc: end.");
			return rsc;
		}
	}

	lrmd_log(LOG_INFO, -1, "lookup_rsc: end.");
	return NULL;
}

void
free_rsc(lrmd_rsc_t* rsc)
{
	g_free(rsc->name);
	g_free(rsc->type);
	if (NULL != rsc->params) {
		free_hash_table(rsc->params);
	}
	g_free(rsc);
}

void
free_mon(lrmd_mon_t* mon)
{
	if (mon->timeout_tag > 0 ) {
		g_source_remove(mon->timeout_tag);
	}
	/* if there is no status op is pending, just release it. */
	if (!mon->pending_op) {
		ha_msg_del(mon->msg);
		g_free(mon);
	}
	else {
		/*  the op stores this pointer so let the op done routine release */
		/*  the memory */
		mon->is_deleted = TRUE;
	}
}
void
lrmd_log (int priority, int level, const char* fmt)
{
	cl_log(priority, "%s",fmt);
}