qdevice: Sending error (output) in IPC

Signed-off-by: Jan Friesse <jfriesse@redhat.com>

qdevices/Makefile.am

@@ -56,6 +56,7 @@ corosync_qdevice_SOURCES	= corosync-qdevice.c qdevice-cmap.c qdevice-instance.c
 				  msg.c msgio.c nss-sock.c tlv.c \
 				  unix-socket.c unix-socket-client.c unix-socket-client-list.c \
 				  unix-socket-ipc.c qdevice-ipc.c pr-poll-array.c dynar-simple-lex.c \
+				  dynar-str.c \
 				  qdevice-net-poll.c qdevice-net-send.c qdevice-net-votequorum.c \
 				  qdevice-net-socket.c qdevice-net-nss.c qdevice-net-msg-received.c \
 				  qdevice-net-cast-vote-timer.c qdevice-net-echo-request-timer.c \
@@ -76,8 +77,8 @@ corosync-qnetd-certutil: corosync-qnetd-certutil.sh
 	    -e 's#@''COROSYSCONFDIR@#${COROSYSCONFDIR}#g' \
 	    $< > $@
 
-TESTS				= qnetd-cluster-list.test
-check_PROGRAMS			= qnetd-cluster-list.test
+TESTS				= qnetd-cluster-list.test dynar.test
+check_PROGRAMS			= qnetd-cluster-list.test dynar.test
 
 qnetd_cluster_list_test_SOURCES	= qnetd-cluster-list.c test-qnetd-cluster-list.c \
 				    qnetd-client-list.c qnetd-client.c dynar.c node-list.c \
@@ -85,4 +86,6 @@ qnetd_cluster_list_test_SOURCES	= qnetd-cluster-list.c test-qnetd-cluster-list.c
 qnetd_cluster_list_test_CFLAGS  = $(nss_CFLAGS)
 qnetd_cluster_list_test_LDADD	= $(nss_LIBS)
 
+dynar_test_SOURCES		= test-dynar.c dynar.c dynar-str.c
+
 endif

qdevices/qdevice-ipc.c

@@ -37,6 +37,7 @@
 #include "qdevice-log.h"
 #include "unix-socket-ipc.h"
 #include "dynar-simple-lex.h"
+#include "dynar-str.h"
 
 int
 qdevice_ipc_init(struct qdevice_instance *instance)
@@ -92,6 +93,13 @@ qdevice_ipc_accept(struct qdevice_instance *instance, struct unix_socket_client
 		break;
 	}
 
+	(*res_client)->user_data = malloc(sizeof(struct qdevice_ipc_user_data));
+	if ((*res_client)->user_data == NULL) {
+		qdevice_log(LOG_ERR, "Can't alloc IPC client user data");
+		res = -1;
+		qdevice_ipc_client_disconnect(instance, *res_client);
+	}
+	memset((*res_client)->user_data, 0, sizeof(struct qdevice_ipc_user_data));
 
 	return (res);
 }
@@ -100,9 +108,34 @@ void
 qdevice_ipc_client_disconnect(struct qdevice_instance *instance, struct unix_socket_client *client)
 {
 
+	free(client->user_data);
 	unix_socket_ipc_client_disconnect(&instance->local_ipc, client);
 }
 
+int
+qdevice_ipc_send_error(struct qdevice_instance *instance, struct unix_socket_client *client,
+    const char *error_fmt, ...)
+{
+	va_list ap;
+	int res;
+
+	va_start(ap, error_fmt);
+	res = ((dynar_str_cpy(&client->send_buffer, "Error\n") == 0) &&
+	    (dynar_str_vcatf(&client->send_buffer, error_fmt, ap) > 0) &&
+	    (dynar_str_cat(&client->send_buffer, "\n") == 0));
+
+	va_end(ap);
+
+	if (res) {
+		unix_socket_client_write_buffer(client, 1);
+	} else {
+		qdevice_log(LOG_ERR, "Can't send error to client (buffer too small)");
+	}
+
+	return (res ? 0 : -1);
+}
+
+
 static void
 qdevice_ipc_parse_line(struct qdevice_instance *instance, struct unix_socket_client *client)
 {
@@ -121,8 +154,10 @@ qdevice_ipc_parse_line(struct qdevice_instance *instance, struct unix_socket_cli
 
 	str = dynar_data(token);
 	if (strcasecmp(str, "") == 0) {
-		qdevice_log(LOG_DEBUG, "IPC client error: No command specified");
-		// SEND ERROR
+		qdevice_log(LOG_DEBUG, "IPC client doesn't send command");
+		if (qdevice_ipc_send_error(instance, client, "No command specified") != 0) {
+			client->schedule_disconnect = 1;
+		}
 	} else if (strcasecmp(str, "shutdown") == 0) {
 		qdevice_log(LOG_DEBUG, "IPC client requested shutdown");
 		// Send output?
@@ -131,7 +166,9 @@ qdevice_ipc_parse_line(struct qdevice_instance *instance, struct unix_socket_cli
 		// Send output
 	} else {
 		qdevice_log(LOG_DEBUG, "IPC client sent unknown command");
-		// Send output
+		if (qdevice_ipc_send_error(instance, client, "Unknown command '%s'", str) != 0) {
+			client->schedule_disconnect = 1;
+		}
 	}
 
 	dynar_simple_lex_destroy(&lex);
@@ -155,14 +192,52 @@ qdevice_ipc_io_read(struct qdevice_instance *instance, struct unix_socket_client
 		client->schedule_disconnect = 1;
 		break;
 	case -2:
-		qdevice_log(LOG_ERR, "Can't store message from IPC client. Disconnecting client");
+		qdevice_log(LOG_ERR, "Can't store message from IPC client. Disconnecting client.");
+		client->schedule_disconnect = 1;
+		break;
+	case -3:
+		qdevice_log_err(LOG_ERR, "Can't receive message from IPC client. Disconnecting client.");
 		client->schedule_disconnect = 1;
 		break;
 	case 1:
 		/*
 		 * Full message received
 		 */
+		unix_socket_client_read_line(client, 0);
+
 		qdevice_ipc_parse_line(instance, client);
 		break;
 	}
 }
+
+void
+qdevice_ipc_io_write(struct qdevice_instance *instance, struct unix_socket_client *client)
+{
+	int res;
+
+	res = unix_socket_client_io_write(client);
+
+	switch (res) {
+	case 0:
+		/*
+		 * Partial send
+		 */
+		break;
+	case -1:
+		qdevice_log(LOG_DEBUG, "IPC client closed connection");
+		client->schedule_disconnect = 1;
+		break;
+	case -2:
+		qdevice_log_err(LOG_ERR, "Can't send message to IPC client. Disconnecting client");
+		client->schedule_disconnect = 1;
+		break;
+	case 1:
+		/*
+		 * Full message sent
+		 */
+		unix_socket_client_write_buffer(client, 0);
+		client->schedule_disconnect = 1;
+
+		break;
+	}
+}

qdevices/qdevice-ipc.h

@@ -41,6 +41,10 @@
 extern "C" {
 #endif
 
+struct qdevice_ipc_user_data {
+	void *model_data;
+};
+
 extern int		qdevice_ipc_init(struct qdevice_instance *instance);
 
 extern int		qdevice_ipc_destroy(struct qdevice_instance *instance);
@@ -54,6 +58,13 @@ extern void		qdevice_ipc_client_disconnect(struct qdevice_instance *instance,
 extern void		qdevice_ipc_io_read(struct qdevice_instance *instance,
     struct unix_socket_client *client);
 
+extern void		qdevice_ipc_io_write(struct qdevice_instance *instance,
+    struct unix_socket_client *client);
+
+extern int		qdevice_ipc_send_error(struct qdevice_instance *instance,
+    struct unix_socket_client *client, const char *error_fmt, ...)
+    __attribute__((__format__(__printf__, 3, 4)));
+
 #ifdef __cplusplus
 }
 #endif

qdevices/qdevice-net-poll.c

@@ -149,6 +149,7 @@ qdevice_net_poll_read_ipc_socket(struct qdevice_net_instance *instance)
 {
 	struct unix_socket_client *client;
 	PRFileDesc *prfd;
+	struct qdevice_ipc_user_data *user_data;
 
 	if (qdevice_ipc_accept(instance->qdevice_instance_ptr, &client) != 0) {
 		return ;
@@ -163,15 +164,10 @@ qdevice_net_poll_read_ipc_socket(struct qdevice_net_instance *instance)
 		return ;
 	}
 
-	client->user_data = (void *)prfd;
+	user_data = (struct qdevice_ipc_user_data *)client->user_data;
+	user_data->model_data = (void *)prfd;
 }
 
-static void
-qdevice_net_poll_write_ipc_client(struct qdevice_net_instance *instance, struct unix_socket_client *client)
-{
-}
-
-
 static PRPollDesc *
 qdevice_net_pr_poll_array_create(struct qdevice_net_instance *instance)
 {
@@ -180,6 +176,7 @@ qdevice_net_pr_poll_array_create(struct qdevice_net_instance *instance)
 	struct qdevice_net_poll_array_user_data *user_data;
 	struct unix_socket_client *ipc_client;
 	const struct unix_socket_client_list *ipc_client_list;
+	struct qdevice_ipc_user_data *qdevice_ipc_user_data;
 
 	poll_array = &instance->poll_array;
 	ipc_client_list = &instance->qdevice_instance_ptr->local_ipc.clients;
@@ -237,7 +234,8 @@ qdevice_net_pr_poll_array_create(struct qdevice_net_instance *instance)
 			return (NULL);
 		}
 
-		poll_desc->fd = ipc_client->user_data;
+		qdevice_ipc_user_data = (struct qdevice_ipc_user_data *)ipc_client->user_data;
+		poll_desc->fd = (PRFileDesc *)qdevice_ipc_user_data->model_data;
 		if (ipc_client->reading_line) {
 			poll_desc->in_flags |= PR_POLL_READ;
 		}
@@ -259,10 +257,12 @@ int
 qdevice_net_poll(struct qdevice_net_instance *instance)
 {
 	PRPollDesc *pfds;
+	PRFileDesc *prfd;
 	PRInt32 poll_res;
 	ssize_t i;
 	struct qdevice_net_poll_array_user_data *user_data;
 	struct unix_socket_client *ipc_client;
+	struct qdevice_ipc_user_data *qdevice_ipc_user_data;
 
 	pfds = qdevice_net_pr_poll_array_create(instance);
 	if (pfds == NULL) {
@@ -308,7 +308,7 @@ qdevice_net_poll(struct qdevice_net_instance *instance)
 					qdevice_net_poll_write_socket(instance, &pfds[i]);
 					break;
 				case QDEVICE_NET_POLL_ARRAY_USER_DATA_TYPE_IPC_CLIENT:
-					qdevice_net_poll_write_ipc_client(instance, ipc_client);
+					qdevice_ipc_io_write(instance->qdevice_instance_ptr, ipc_client);
 					break;
 				default:
 					qdevice_log(LOG_CRIT, "Unhandled write on poll descriptor %u", i);
@@ -350,6 +350,13 @@ qdevice_net_poll(struct qdevice_net_instance *instance)
 
 			if (user_data->type == QDEVICE_NET_POLL_ARRAY_USER_DATA_TYPE_IPC_CLIENT &&
 			    ipc_client->schedule_disconnect) {
+				qdevice_ipc_user_data = (struct qdevice_ipc_user_data *)ipc_client->user_data;
+				prfd = (PRFileDesc *)qdevice_ipc_user_data->model_data;
+
+				if (PR_DestroySocketPollFd(prfd) != PR_SUCCESS) {
+					qdevice_log_nss(LOG_WARNING, "Unable to destroy IPC poll socket fd");
+				}
+
 				qdevice_ipc_client_disconnect(instance->qdevice_instance_ptr, ipc_client);
 			}
 		}

qdevices/qnetd-client.c

@@ -38,7 +38,6 @@
 
 #include "qnet-config.h"
 #include "qnetd-client.h"
-#include "qnetd-client-algo-timer.h"
 
 void
 qnetd_client_init(struct qnetd_client *client, PRFileDesc *sock, PRNetAddr *addr,
@@ -61,7 +60,6 @@ void
 qnetd_client_destroy(struct qnetd_client *client)
 {
 
-	qnetd_client_algo_timer_abort(client);
 	free(client->cluster_name);
 	node_list_free(&client->last_quorum_node_list);
 	node_list_free(&client->last_membership_node_list);

qdevices/qnetd-instance.c

@@ -41,6 +41,7 @@
 #include "qnetd-log-debug.h"
 #include "qnetd-dpd-timer.h"
 #include "qnetd-poll-array-user-data.h"
+#include "qnetd-client-algo-timer.h"
 
 int
 qnetd_instance_init(struct qnetd_instance *instance, size_t max_client_receive_size,
@@ -112,6 +113,7 @@ qnetd_instance_client_disconnect(struct qnetd_instance *instance, struct qnetd_c
 	if (client->cluster != NULL) {
 		qnetd_cluster_list_del_client(&instance->clusters, client->cluster, client);
 	}
+	qnetd_client_algo_timer_abort(client);
 	qnetd_client_list_del(&instance->clients, client);
 }
 

qdevices/unix-socket-client.c

@@ -32,6 +32,7 @@
  * THE POSSIBILITY OF SUCH DAMAGE.
  */
 
+#include <errno.h>
 #include <string.h>
 
 #include "unix-socket-client.h"
@@ -78,6 +79,7 @@ unix_socket_client_write_buffer(struct unix_socket_client *client, int enabled)
  *  0 Partial read (no error)
  * -1 End of connection
  * -2 Buffer too long
+ * -3 Unhandled error
  */
 int
 unix_socket_client_io_read(struct unix_socket_client *client)
@@ -107,6 +109,53 @@ unix_socket_client_io_read(struct unix_socket_client *client)
 		res = -1;
 	}
 
+	if (readed < 0 && errno != EAGAIN && errno != EWOULDBLOCK && errno != EINTR) {
+		res = -3;
+	}
+
 exit_err:
 	return (res);
 }
+
+/*
+ *  1 All data succesfully sent
+ *  0 Partial send (no error)
+ * -1 End of connection
+ * -2 Unhandled error
+ */
+int
+unix_socket_client_io_write(struct unix_socket_client *client)
+{
+	ssize_t sent;
+	size_t to_send;
+	int res;
+
+	res = 0;
+
+	to_send = dynar_size(&client->send_buffer) - client->msg_already_sent_bytes;
+	if (to_send > UNIX_SOCKET_CLIENT_BUFFER) {
+		to_send = UNIX_SOCKET_CLIENT_BUFFER;
+	}
+
+	sent = unix_socket_write(client->socket,
+	    dynar_data(&client->send_buffer) + client->msg_already_sent_bytes,
+	    to_send);
+
+	if (sent > 0) {
+		client->msg_already_sent_bytes += sent;
+
+		if (client->msg_already_sent_bytes == dynar_size(&client->send_buffer)) {
+			return (1);
+		}
+	}
+
+	if (sent == 0) {
+		res = -1;
+	}
+
+	if (sent < 0 && errno != EAGAIN && errno != EWOULDBLOCK && errno != EINTR) {
+		res = -2;
+	}
+
+	return (res);
+}

qdevices/unix-socket-client.h

@@ -70,6 +70,8 @@ extern void		unix_socket_client_write_buffer(struct unix_socket_client *client,
 
 extern int		unix_socket_client_io_read(struct unix_socket_client *client);
 
+extern int		unix_socket_client_io_write(struct unix_socket_client *client);
+
 #ifdef __cplusplus
 }
 #endif

qdevices/unix-socket.c

@@ -157,3 +157,10 @@ unix_socket_read(int sock, void *buf, size_t len)
 
 	return (recv(sock, buf, len, 0));
 }
+
+ssize_t
+unix_socket_write(int sock, void *buf, size_t len)
+{
+
+	return (send(sock, buf, len, 0));
+}

qdevices/unix-socket.h

@@ -52,6 +52,8 @@ extern int		unix_socket_close(int sock);
 
 extern ssize_t		unix_socket_read(int sock, void *buf, size_t len);
 
+extern ssize_t		unix_socket_write(int sock, void *buf, size_t len);
+
 #ifdef __cplusplus
 }
 #endif