Fix a bunch of errors in node get and async node get.

(Logical change 1.87)


git-svn-id: http://svn.fedorahosted.org/svn/corosync/trunk@317 fd59a12c-fef9-0310-b244-a6a79926bd2f

exec/clm.c

@@ -118,6 +118,9 @@ static int message_handler_req_clm_trackstop (struct conn_info *conn_info,
 static int message_handler_req_clm_nodeget (struct conn_info *conn_info,
 	void *message);
 
+static int message_handler_req_clm_nodegetasync (struct conn_info *conn_info,
+	void *message);
+
 static int clm_exit_fn (struct conn_info *conn_info);
 
 struct libais_handler clm_libais_handlers[] =
@@ -145,6 +148,12 @@ struct libais_handler clm_libais_handlers[] =
 		.response_size				= sizeof (struct res_clm_nodeget),
 		.response_id				= MESSAGE_RES_CLM_NODEGET, // TODO RESPONSE
 		.gmi_prio					= GMI_PRIO_RECOVERY
+	},
+	{ /* 4 */
+		.libais_handler_fn			= message_handler_req_clm_nodegetasync,
+		.response_size				= sizeof (struct res_clm_nodegetasync),
+		.response_id				= MESSAGE_RES_CLM_NODEGETCALLBACK, // TODO RESPONSE
+		.gmi_prio					= GMI_PRIO_RECOVERY
 	}
 };
 
@@ -541,7 +550,6 @@ static int message_handler_req_clm_nodeget (struct conn_info *conn_info, void *m
 	res_clm_nodeget.header.id = MESSAGE_RES_CLM_NODEGET;
 	res_clm_nodeget.header.error = SA_OK;
 	res_clm_nodeget.invocation = req_clm_nodeget->invocation;
-	res_clm_nodeget.clusterNodeAddress = req_clm_nodeget->clusterNodeAddress;
 	res_clm_nodeget.valid = valid;
 	if (valid) {
 		memcpy (&res_clm_nodeget.clusterNode, clusterNode, sizeof (SaClmClusterNodeT));
@@ -550,3 +558,55 @@ static int message_handler_req_clm_nodeget (struct conn_info *conn_info, void *m
 
 	return (0);
 }
+
+static int message_handler_req_clm_nodegetasync (struct conn_info *conn_info, void *message)
+{
+	struct req_clm_nodegetasync *req_clm_nodegetasync = (struct req_clm_nodegetasync *)message;
+	struct res_clm_nodegetasync res_clm_nodegetasync;
+	struct res_clm_nodegetcallback res_clm_nodegetcallback;
+	SaClmClusterNodeT *clusterNode = 0;
+	int valid = 0;
+	int i;
+
+	log_printf (LOG_LEVEL_DEBUG, "nodeget: trying to find node %x\n", (int)req_clm_nodegetasync->nodeId);
+
+	if (req_clm_nodegetasync->nodeId == SA_CLM_LOCAL_NODE_ID) {
+		clusterNode = &clusterNodes[0];
+		valid = 1;
+	} else 
+	for (i = 0; i < clusterNodeEntries; i++) {
+		if (clusterNodes[i].nodeId == req_clm_nodegetasync->nodeId) {
+			log_printf (LOG_LEVEL_DEBUG, "found host that matches one desired in nodeget.\n");
+			clusterNode = &clusterNodes[i];
+			valid = 1;
+			break;
+		}
+	}
+
+	/*
+	 * Respond to library request
+	 */
+	res_clm_nodegetasync.header.size = sizeof (struct res_clm_nodegetasync);
+	res_clm_nodegetasync.header.id = MESSAGE_RES_CLM_NODEGETASYNC;
+	res_clm_nodegetasync.header.error = SA_OK;
+	libais_send_response (conn_info, &res_clm_nodegetasync,
+		sizeof (struct res_clm_nodegetasync));
+
+	/*
+	 * Send async response
+	 */
+	res_clm_nodegetcallback.header.size = sizeof (struct res_clm_nodegetcallback);
+	res_clm_nodegetcallback.header.id = MESSAGE_RES_CLM_NODEGETCALLBACK;
+	res_clm_nodegetcallback.header.error = SA_OK;
+	res_clm_nodegetcallback.invocation = req_clm_nodegetasync->invocation;
+	res_clm_nodegetcallback.clusterNodeAddress = req_clm_nodegetasync->clusterNodeAddress;
+	res_clm_nodegetcallback.valid = valid;
+	if (valid) {
+		memcpy (&res_clm_nodegetcallback.clusterNode, clusterNode,
+			sizeof (SaClmClusterNodeT));
+	}
+	libais_send_response (conn_info, &res_clm_nodegetcallback,
+		sizeof (struct res_clm_nodegetcallback));
+
+	return (0);
+}

include/ais_msg.h

@@ -85,15 +85,16 @@ enum res_lib_evs_types {
 enum req_clm_types {
 	MESSAGE_REQ_CLM_TRACKSTART = 1,
 	MESSAGE_REQ_CLM_TRACKSTOP,
-	MESSAGE_REQ_CLM_NODEGET
+	MESSAGE_REQ_CLM_NODEGET,
+	MESSAGE_REQ_CLM_NODEGETASYNC
 };
 
 enum res_clm_types {
-	
 	MESSAGE_RES_CLM_TRACKCALLBACK = 1,
 	MESSAGE_RES_CLM_TRACKSTART,
 	MESSAGE_RES_CLM_TRACKSTOP,
 	MESSAGE_RES_CLM_NODEGET,
+	MESSAGE_RES_CLM_NODEGETASYNC,
 	MESSAGE_RES_CLM_NODEGETCALLBACK
 };
 
@@ -826,11 +827,21 @@ struct req_clm_nodeget {
 struct res_clm_nodeget {
 	struct res_header header;
 	SaInvocationT invocation;
-	SaClmClusterNodeT *clusterNodeAddress;
 	SaClmClusterNodeT clusterNode;
 	int valid;
 };
 
+struct req_clm_nodegetasync {
+	struct req_header header;
+	SaClmClusterNodeT *clusterNodeAddress;
+	SaInvocationT invocation;
+	SaClmNodeIdT nodeId;
+};
+
+struct res_clm_nodegetasync {
+	struct res_header header;
+};
+
 struct res_clm_nodegetcallback {
 	struct res_header header;
 	SaInvocationT invocation;

lib/clm.c

@@ -49,13 +49,14 @@
 #include "../include/ais_msg.h"
 #include "util.h"
 
-struct message_overlay {
-	struct req_header header;
+struct res_overlay {
+	struct res_header header;
 	char data[4096];
 };
 
 struct clmInstance {
 	int fd;
+	struct queue inq;
 	SaClmCallbacksT callbacks;
 	int finalize;
 	pthread_mutex_t mutex;
@@ -121,7 +122,16 @@ saClmInitialize (
 	}
 
 	clmInstance->fd = -1;
-	
+
+	/*
+	 * An inq is needed to store async messages while waiting for a
+	 * sync response
+	 */
+	error = saQueueInit (&clmInstance->inq, 512, sizeof (void *));
+	if (error != SA_OK) {
+		goto error_put_destroy;
+	}
+
 	error = saServiceConnect (&clmInstance->fd, MESSAGE_REQ_CLM_INIT);
 	if (error != SA_OK) {
 		goto error_put_destroy;
@@ -177,7 +187,10 @@ saClmDispatch (
 	struct res_clm_trackcallback *res_clm_trackcallback;
 	struct res_clm_nodegetcallback *res_clm_nodegetcallback;
 	SaClmCallbacksT callbacks;
-	struct message_overlay dispatch_data;
+	struct res_overlay dispatch_data;
+	int empty;
+	struct res_header **queue_msg;
+	struct res_header *msg;
 
 	error = saHandleInstanceGet (&clmHandleDatabase, *clmHandle, (void *)&clmInstance);
 	if (error != SA_OK) {
@@ -225,33 +238,42 @@ saClmDispatch (
 			continue; /* next poll */
 		}
 
-		/*
-		 * Read header
-		 */
-		error = saRecvRetry (clmInstance->fd, &dispatch_data.header,
-			sizeof (struct req_header), MSG_WAITALL | MSG_NOSIGNAL);
-		if (error != SA_OK) {
-			goto error_unlock;
-		}
-
-		/*
-		 * Read data payload
-		 */
-		if (dispatch_data.header.size > sizeof (struct req_header)) {
-			error = saRecvRetry (clmInstance->fd, &dispatch_data.data,
-				dispatch_data.header.size - sizeof (struct req_header), MSG_WAITALL | MSG_NOSIGNAL);
+		saQueueIsEmpty(&clmInstance->inq, &empty);
+		if (empty == 0) {
+			/*
+			 * Queue is not empty, read data from queue
+			 */
+			saQueueItemGet (&clmInstance->inq, (void *)&queue_msg);
+			msg = *queue_msg;
+			memcpy (&dispatch_data, msg, msg->size);
+			saQueueItemRemove (&clmInstance->inq);
+			free (msg);
+		} else {
+			/*
+			 * Queue empty, read response from socket
+			 */
+			error = saRecvRetry (clmInstance->fd, &dispatch_data.header,
+				sizeof (struct res_header), MSG_WAITALL | MSG_NOSIGNAL);
 			if (error != SA_OK) {
 				goto error_unlock;
 			}
+			if (dispatch_data.header.size > sizeof (struct res_header)) {
+				error = saRecvRetry (clmInstance->fd, &dispatch_data.data,
+					dispatch_data.header.size - sizeof (struct res_header),
+					MSG_WAITALL | MSG_NOSIGNAL);
+				if (error != SA_OK) {
+					goto error_unlock;
+				}
+			}
 		}
 		/*
 		 * Make copy of callbacks, message data, unlock instance, and call callback
 		 * A risk of this dispatch method is that the callback routines may
-		 * operate at the same time that clmFinalize has been called.
-		*/
+		 * operate at the same time that clmFinalize has been called in another thread.
+		 */
 		memcpy (&callbacks, &clmInstance->callbacks, sizeof (SaClmCallbacksT));
-
 		pthread_mutex_unlock (&clmInstance->mutex);
+
 		/*
 		 * Dispatch incoming message
 		 */
@@ -274,7 +296,8 @@ saClmDispatch (
 			res_clm_nodegetcallback = (struct res_clm_nodegetcallback *)&dispatch_data;
 
 			memcpy (res_clm_nodegetcallback->clusterNodeAddress,
-				&res_clm_nodegetcallback->clusterNode, sizeof (SaClmClusterNodeT));
+				&res_clm_nodegetcallback->clusterNode,
+				sizeof (SaClmClusterNodeT));
 
 			callbacks.saClmClusterNodeGetCallback (
 				res_clm_nodegetcallback->invocation,
@@ -412,7 +435,6 @@ saClmClusterNodeGet (
 	int fd;
 	struct req_clm_nodeget req_clm_nodeget;
 	struct res_clm_nodeget res_clm_nodeget;
-	struct message_overlay message;
 	SaErrorT error = SA_OK;
 	struct timeval select_timeout;
 	fd_set read_fds;
@@ -454,21 +476,21 @@ saClmClusterNodeGet (
 		goto error_close;
 	}
 
-	error = saRecvRetry (fd, &message.header, sizeof (struct req_header), MSG_WAITALL | MSG_NOSIGNAL);
+	error = saRecvRetry (fd, &res_clm_nodeget, sizeof (struct res_clm_nodeget),
+		MSG_WAITALL | MSG_NOSIGNAL);
 	if (error != SA_OK) {
 		goto error_close;
 	}
 
-	error = saRecvRetry (fd, &message.data, message.header.size - sizeof (struct req_header), MSG_WAITALL | MSG_NOSIGNAL);
-	if (error != SA_OK) {
-		goto error_close;
-	}
+	error = res_clm_nodeget.header.error;
 
 	memcpy (clusterNode, &res_clm_nodeget.clusterNode, sizeof (SaClmClusterNodeT));
 
+error_noclose:
+	return (error);
+
 error_close:
 	close (fd);
-error_noclose:
 	return (error);
 }
 
@@ -480,14 +502,15 @@ saClmClusterNodeGetAsync (
 	SaClmClusterNodeT *clusterNode)
 {
 	struct clmInstance *clmInstance;
-	struct req_clm_nodeget req_clm_nodeget;
+	struct req_clm_nodegetasync req_clm_nodegetasync;
+	struct res_clm_nodegetasync res_clm_nodegetasync;
 	SaErrorT error = SA_OK;
 
-	req_clm_nodeget.header.size = sizeof (struct req_clm_nodeget);
-	req_clm_nodeget.header.id = MESSAGE_REQ_CLM_NODEGET;
-	memcpy (&req_clm_nodeget.invocation, &invocation, sizeof (SaInvocationT));
-	memcpy (&req_clm_nodeget.nodeId, &nodeId, sizeof (SaClmNodeIdT));
-	req_clm_nodeget.clusterNodeAddress = clusterNode;
+	req_clm_nodegetasync.header.size = sizeof (struct req_clm_nodeget);
+	req_clm_nodegetasync.header.id = MESSAGE_REQ_CLM_NODEGETASYNC;
+	memcpy (&req_clm_nodegetasync.invocation, &invocation, sizeof (SaInvocationT));
+	memcpy (&req_clm_nodegetasync.nodeId, &nodeId, sizeof (SaClmNodeIdT));
+	req_clm_nodegetasync.clusterNodeAddress = clusterNode;
 
 	error = saHandleInstanceGet (&clmHandleDatabase, *clmHandle, (void *)&clmInstance);
 	if (error != SA_OK) {
@@ -496,11 +519,20 @@ saClmClusterNodeGetAsync (
 
 	pthread_mutex_lock (&clmInstance->mutex);
 
-	error = saSendRetry (clmInstance->fd, &req_clm_nodeget,
-		sizeof (struct req_clm_nodeget), MSG_NOSIGNAL);
+	error = saSendRetry (clmInstance->fd, &req_clm_nodegetasync,
+		sizeof (struct req_clm_nodegetasync), MSG_NOSIGNAL);
 
 	pthread_mutex_unlock (&clmInstance->mutex);
 
+	error = saRecvQueue (clmInstance->fd, &res_clm_nodegetasync,
+		&clmInstance->inq, MESSAGE_RES_CLM_NODEGETASYNC);
+	if (error != SA_OK) {
+		goto error_exit;
+	}
+
+	error = res_clm_nodegetasync.header.error;
+
+error_exit:
 	saHandleInstancePut (&clmHandleDatabase, *clmHandle);
 
 	return (error);