Common daemon code: Implement basic socket-based communication infrastructure

(both client and server side). The server handles each connection in a separate
thread.

daemons/common/daemon-client.c

@@ -0,0 +1,54 @@
+#include "daemon-client.h"
+#include "daemon-shared.h"
+#include <sys/un.h>
+#include <sys/socket.h>
+#include <string.h>
+#include <stdio.h>
+#include <assert.h>
+
+daemon_handle daemon_open(daemon_info i) {
+	daemon_handle h;
+	struct sockaddr_un sockaddr;
+	if ((h.socket_fd = socket(PF_UNIX, SOCK_STREAM | SOCK_NONBLOCK, 0)) < 0) {
+		perror("socket");
+		goto error;
+	}
+	memset(&sockaddr, 0, sizeof(sockaddr));
+	fprintf(stderr, "connecting to %s\n", i.socket);
+	strcpy(sockaddr.sun_path, i.socket);
+	sockaddr.sun_family = AF_UNIX;
+	if (connect(h.socket_fd,(struct sockaddr *) &sockaddr, sizeof(sockaddr))) {
+		perror("connect");
+		goto error;
+	}
+	h.protocol = 0;
+	return h;
+error:
+	if (h.socket_fd >= 0)
+		close(h.socket_fd);
+	h.socket_fd = -1;
+	return h;
+}
+
+daemon_reply daemon_send(daemon_handle h, daemon_request rq)
+{
+	daemon_reply reply;
+	assert(h.socket_fd >= 0);
+
+	if (!rq.buffer) {
+		/* TODO: build the buffer from rq.cft */
+	}
+
+	assert(rq.buffer);
+	write_buffer(h.socket_fd, rq.buffer, strlen(rq.buffer));
+
+	if (read_buffer(h.socket_fd, &reply.buffer)) {
+		/* TODO: parse reply.buffer into reply.cft */
+	} else
+		reply.error = 1;
+
+	return reply;
+}
+
+void daemon_close(daemon_handle h) {
+}

daemons/common/daemon-client.h

@@ -12,6 +12,8 @@
  * Inc., 59 Temple Place, Suite 330, Boston, MA  02111-1307  USA
  */
 
+#include "config.h" // should become part of libdevmapper later
+
 #ifndef _LVM_DAEMON_COMMON_CLIENT_H
 #define _LVM_DAEMON_COMMON_CLIENT_H
 
@@ -28,13 +30,14 @@ typedef struct {
 } daemon_info;
 
 typedef struct {
-	char *request;
+	char *buffer;
+	struct config_node *cft;
 } daemon_request;
 
 typedef struct {
 	int error; /* 0 for success */
-	char *reply; /* textual reply */
-	struct config_tree *cft; /* parsed reply, if available */
+	char *buffer; /* textual reply */
+	struct config_node *cft; /* parsed reply, if available */
 } daemon_reply;
 
 /*

daemons/common/daemon-server.c

@@ -107,8 +107,9 @@ static int _open_socket(daemon_state s)
 		fprintf(stderr, "setting CLOEXEC on socket fd %d failed: %s\n", fd, strerror(errno));
 	fcntl(fd, F_SETFL, fcntl(fd, F_GETFL, 0) | O_NONBLOCK);
 
+	fprintf(stderr, "creating %s\n", s.socket_path);
 	memset(&sockaddr, 0, sizeof(sockaddr));
-	memcpy(sockaddr.sun_path, s.socket_path, strlen(s.socket_path));
+	strcpy(sockaddr.sun_path, s.socket_path);
 	sockaddr.sun_family = AF_UNIX;
 
 	if (bind(fd, (struct sockaddr *) &sockaddr, sizeof(sockaddr))) {
@@ -128,6 +129,7 @@ out:
 error:
 	if (fd >= 0) {
 		close(fd);
+		unlink(s.socket_path);
 		fd = -1;
 	}
 	goto out;
@@ -198,6 +200,60 @@ static void _daemonise(void)
 	setsid();
 }
 
+struct thread_baton {
+	daemon_state s;
+	client_handle client;
+};
+
+void *client_thread(void *baton)
+{
+	struct thread_baton *b = baton;
+	request req;
+	while (1) {
+		if (!read_buffer(b->client.socket_fd, &req.buffer))
+			goto fail;
+
+		/* TODO parse the buffer into req.cft */
+		response res = b->s.handler(b->s, b->client, req);
+
+		if (!res.buffer) {
+			/* TODO fill in the buffer from res.cft */
+		}
+
+		write_buffer(b->client.socket_fd, res.buffer, strlen(res.buffer));
+
+		free(res.buffer);
+		free(req.buffer);
+	}
+fail:
+	/* TODO what should we really do here? */
+	return NULL;
+}
+
+int handle_connect(daemon_state s)
+{
+	struct sockaddr_un sockaddr;
+	client_handle client;
+	socklen_t sl = sizeof(sockaddr);
+	int client_fd = accept(s.socket_fd, (struct sockaddr *) &sockaddr, &sl);
+	if (client_fd < 0)
+		return 0;
+
+	struct thread_baton *baton = malloc(sizeof(struct thread_baton));
+	if (!baton)
+		return 0;
+
+	client.socket_fd = client_fd;
+	client.read_buf = 0;
+	client.private = 0;
+	baton->s = s;
+	baton->client = client;
+
+	if (pthread_create(&baton->client.thread_id, NULL, client_thread, baton))
+		return 0;
+	return 1;
+}
+
 void daemon_start(daemon_state s)
 {
 	int failed = 0;
@@ -230,6 +286,8 @@ void daemon_start(daemon_state s)
 	signal(SIGINT, &_exit_handler);
 	signal(SIGHUP, &_exit_handler);
 	signal(SIGQUIT, &_exit_handler);
+	signal(SIGTERM, &_exit_handler);
+	signal(SIGPIPE, SIG_IGN);
 
 #ifdef linux
 	if (s.avoid_oom && !_set_oom_adj(OOM_DISABLE) && !_set_oom_adj(OOM_ADJUST_MIN))
@@ -247,9 +305,20 @@ void daemon_start(daemon_state s)
 		kill(getppid(), SIGTERM);
 
 	while (!_shutdown_requested && !failed) {
-		/* TODO: do work */
+		int status;
+		fd_set in;
+		FD_ZERO(&in);
+		FD_SET(s.socket_fd, &in);
+		if (select(FD_SETSIZE, &in, NULL, NULL, NULL) < 0 && errno != EINTR)
+			perror("select error");
+		if (FD_ISSET(s.socket_fd, &in))
+			if (!handle_connect(s))
+				syslog(LOG_ERR, "Failed to handle a client connection.");
 	}
 
+	if (s.socket_fd >= 0)
+		unlink(s.socket_path);
+
 	syslog(LOG_NOTICE, "%s shutting down", s.name);
 	closelog();
 	remove_lockfile(s.pidfile);

daemons/common/daemon-server.h

@@ -13,6 +13,7 @@
  */
 
 #include "daemon-client.h"
+#include "config.h" // XXX will be in libdevmapper.h later
 
 #ifndef _LVM_DAEMON_COMMON_SERVER_H
 #define _LVM_DAEMON_COMMON_SERVER_H
@@ -25,12 +26,14 @@ typedef struct {
 } client_handle;
 
 typedef struct {
-	struct config_tree *cft;
+	struct config_node *cft;
+	char *buffer;
 } request;
 
 typedef struct {
 	int error;
-	struct config_tree *cft;
+	struct config_node *cft;
+	char *buffer;
 } response;
 
 struct daemon_state;

daemons/common/daemon-shared.c

@@ -0,0 +1,68 @@
+#include <errno.h>
+#include <stdio.h>
+#include <malloc.h>
+#include <string.h>
+
+/*
+ * Read a single message from a (socket) filedescriptor. Messages are delimited
+ * by blank lines. This call will block until all of a message is received. The
+ * memory will be allocated from heap. Upon error, all memory is freed and the
+ * buffer pointer is set to NULL.
+ */
+int read_buffer(int fd, char **buffer) {
+	int bytes = 0;
+	int buffersize = 32;
+	*buffer = malloc(buffersize + 1);
+
+	while (1) {
+		int result = read(fd, (*buffer) + bytes, buffersize - bytes);
+		if (result > 0)
+			bytes += result;
+		if (result == 0)
+			goto fail; /* we should never encounter EOF here */
+		if (result < 0 && errno != EAGAIN && errno != EWOULDBLOCK)
+			goto fail;
+
+		if (bytes == buffersize) {
+			buffersize += 1024;
+			char *new = realloc(*buffer, buffersize + 1);
+			if (new)
+				*buffer = new;
+			else
+				goto fail;
+		} else {
+			(*buffer)[bytes] = 0;
+			char *end;
+			if ((end = strstr((*buffer) + bytes - 2, "\n\n"))) {
+				*end = 0;
+				break; /* success, we have the full message now */
+			}
+			/* TODO call select here if we encountered EAGAIN/EWOULDBLOCK */
+		}
+	}
+	return 1;
+fail:
+	free(*buffer);
+	*buffer = NULL;
+	return 0;
+}
+
+/*
+ * Write a buffer to a filedescriptor. Keep trying. Blocks (even on
+ * SOCK_NONBLOCK) until all of the write went through.
+ *
+ * TODO use select on EWOULDBLOCK/EAGAIN to avoid useless spinning
+ */
+int write_buffer(int fd, char *buffer, int length) {
+	int written = 0;
+	while (1) {
+		int result = write(fd, buffer + written, length - written);
+		if (result > 0)
+			written += result;
+		if (result < 0 && errno != EWOULDBLOCK && errno != EAGAIN)
+			break; /* too bad */
+		if (written == length)
+			return 1; /* done */
+	}
+	return 0;
+}

daemons/common/daemon-shared.h

@@ -0,0 +1,2 @@
+int read_buffer(int fd, char **buffer);
+int write_buffer(int fd, char *buffer, int length);