libipc-old/src/communication.c

#include <sys/select.h>
#include <unistd.h>

#include <stdio.h>
#include <errno.h>		// error numbers

#include <stdlib.h>
#include <string.h>

#include "ipc.h"
#include "utils.h"

// print structures
#include "message.h"

struct ipc_error ipc_server_init (struct ipc_ctx *ctx, const char *sname)
{
	T_R ((sname == NULL), IPC_ERROR_SERVER_INIT__NO_SERVER_NAME_PARAM);

	// Declaration and instanciation of the new connection (ipc_connection_info + pollfd).
	SECURE_DECLARATION (struct ipc_connection_info, srv);
	srv.type = IPC_CONNECTION_TYPE_SERVER;
	SECURE_DECLARATION(struct pollfd, pollfd);
	pollfd.events = POLLIN;

	// Get the service path.
	SECURE_BUFFER_DECLARATION (char, buf, PATH_MAX);
	TEST_IPC_RR (service_path (buf, sname), "cannot get server path");
	size_t s = strlen (buf);
	if (s > PATH_MAX)
		s = PATH_MAX;
	SECURE_BUFFER_HEAP_ALLOCATION_R (srv.spath, s + 1,, IPC_ERROR_SERVER_INIT__MALLOC);
	memcpy (srv.spath, buf, s);
	srv.spath[s] = '\0';	// to be sure

	// Socket initialisation for the service.
	TEST_IPC_RETURN_ON_ERROR (usock_init (&pollfd.fd, srv.spath));

	// Add the server to the listened file descriptors.
	// ipc_add allocate memory then copy the data of srv and pollfd in ctx.
	TEST_IPC_RR (ipc_add (ctx, &srv, &pollfd), "cannot add the server in the context");

#if 0
	printf ("server init, listen on %ld fd: ", ctx->size);
	for (size_t i = 0; i < ctx->size; i++)
		printf (" %d", ctx->pollfd[i].fd);
	printf ("\n");
#endif

	IPC_RETURN_NO_ERROR;
}

// when networkd is not working properly (or do not retrieve the service): srv->fd = 0
struct ipc_error ipc_contact_networkd (int *pfd, const char *sname)
{
	T_R ((pfd == NULL),   IPC_ERROR_CONTACT_NETWORKD__NO_FD_PARAM);
	T_R ((sname == NULL), IPC_ERROR_CONTACT_NETWORKD__NO_SERVICE_NAME_PARAM);

	char *networkvar = getenv ("IPC_NETWORK");
	if (networkvar == NULL) {
		*pfd = 0;
		IPC_RETURN_NO_ERROR;
	}
	// TODO: is there another, more interesting way to do this?
	// currently, IPC_NETWORK is shared with the network service
	// in order to route requests over any chosen protocol stack
	// ex: IPC_NETWORK="audio tor://some.example.com/audio ;pong tls://pong.example.com/pong"

	// printf ("IPC_NETWORK: %s\n", networkvar);

	SECURE_BUFFER_DECLARATION (char, columnthensname, BUFSIZ);
	columnthensname[0] = ';';
	memcpy (columnthensname + 1, sname, strlen (sname));

	if (strncmp (networkvar, sname, strlen (sname)) != 0 && strstr (networkvar, columnthensname) == NULL) {
		// printf ("sname %s not found\n", sname);
		*pfd = 0;
		IPC_RETURN_NO_ERROR;
	}

	// Get the service path.
	SECURE_BUFFER_DECLARATION (char, buf, PATH_MAX);
	TEST_IPC_RR (service_path (buf, "network"), "cannot get network service path");

	int networkdfd = 0;

	TEST_IPC_RETURN_ON_ERROR (usock_connect (&networkdfd, buf));

	SECURE_DECLARATION (struct ipc_message, msg);
	msg.type = MSG_TYPE_NETWORK_LOOKUP;
	msg.user_type = MSG_TYPE_NETWORK_LOOKUP;

	SECURE_BUFFER_DECLARATION (char, content, BUFSIZ);
	snprintf (content, BUFSIZ, "%s;%s", sname, networkvar);

	msg.length = strlen (content);
	msg.payload = content;

	TEST_IPC_RR (ipc_write_fd (networkdfd, &msg), "cannot send a message to networkd");

	struct ipc_error ret = ipc_receive_fd (networkdfd, pfd);
	if (ret.error_code == IPC_ERROR_NONE) {
		usock_close (networkdfd);
	}

	return ret;
}

// Create context, contact networkd, connects to the service.
struct ipc_error ipc_connection (struct ipc_ctx *ctx, const char *sname)
{
	T_R ((ctx == NULL),   IPC_ERROR_CONNECTION__NO_CTX);
	T_R ((sname == NULL), IPC_ERROR_CONNECTION__NO_SERVICE_NAME);

	SECURE_DECLARATION(struct ipc_connection_info, srv);
	srv.type = IPC_CONNECTION_TYPE_IPC; // Data received on the socket = messages, not new clients.
	SECURE_DECLARATION(struct pollfd, pollfd);
	pollfd.events = POLLIN;

	TEST_IPC_P (ipc_contact_networkd (&pollfd.fd, sname), "error during networkd connection");

	// if networkd did not initiate the connection
	if (pollfd.fd <= 0) {
		// gets the service path
		SECURE_BUFFER_DECLARATION (char, buf, PATH_MAX);
		TEST_IPC_RR (service_path (buf, sname), "cannot get server path");
		TEST_IPC_RETURN_ON_ERROR (usock_connect (&pollfd.fd, buf));
	}

	// Add the server to the listened file descriptors.
	TEST_IPC_RR (ipc_add (ctx, &srv, &pollfd), "cannot add the server in the context");

	IPC_RETURN_NO_ERROR;
}

struct ipc_error ipc_close_all (struct ipc_ctx *ctx)
{
	for (size_t i = 0 ; i < ctx->size ; i++) {
		TEST_IPC_P (ipc_close (ctx, i), "cannot close a connection in handle_message");
	}

	IPC_RETURN_NO_ERROR;
}

struct ipc_error ipc_close (struct ipc_ctx *ctx, uint32_t index)
{
	struct ipc_error ret = usock_close (ctx->pollfd[index].fd);

	// TODO: verify that the close was OK??
	if (ctx->cinfos[index].type == IPC_CONNECTION_TYPE_SERVER) {
		ret = usock_remove (ctx->cinfos[index].spath);
		if (ctx->cinfos[index].spath != NULL) {
			free (ctx->cinfos[index].spath);
			ctx->cinfos[index].spath = NULL;
		}
	}

	return ret;
}

// New connection from a client.
struct ipc_error ipc_accept_add (struct ipc_event *event, struct ipc_ctx *ctx, uint32_t index)
{
	T_R ((ctx == NULL),        IPC_ERROR_HANDLE_NEW_CONNECTION__NO_CINFOS_PARAM);
	T_R ((index >= ctx->size), IPC_ERROR_HANDLE_NEW_CONNECTION__INCONSISTENT_INDEX);

	// Memory reallocation.
	ipc_ctx_new_alloc (ctx);

	int server_fd  = ctx->pollfd[index].fd;
	int *client_fd = &ctx->pollfd[ctx->size -1].fd;

	TEST_IPC_RR (usock_accept (server_fd, client_fd), "cannot accept IPC connection");
	ctx->pollfd[ctx->size -1].events = POLLIN; // Tell to poll(2) to watch for incoming data from this fd.
	ctx->cinfos[ctx->size -1].type = IPC_CONNECTION_TYPE_IPC;

	// Set the event structure.
	uint32_t client_index = ctx->size - 1;
	IPC_EVENT_SET (event, IPC_EVENT_TYPE_CONNECTION, client_index, *client_fd, NULL);

	IPC_RETURN_NO_ERROR;
}

// receive then format in an ipc_message structure
struct ipc_error ipc_read (const struct ipc_ctx *ctx, uint32_t index, struct ipc_message *m)
{
	T_R ((m == NULL), IPC_ERROR_READ__NO_MESSAGE_PARAM);

	char *buf = NULL;
	size_t msize = IPC_MAX_MESSAGE_SIZE;

	printf ("reading on the fd %d\n", ctx->pollfd[index].fd);

	// On error or closed recipient, the buffer already freed.
	TEST_IPC_RETURN_ON_ERROR      (usock_recv (ctx->pollfd[index].fd, &buf, &msize));
	TEST_IPC_RETURN_ON_ERROR_FREE (ipc_message_format_read (m, buf, msize), buf);

	free (buf);

	IPC_RETURN_NO_ERROR;	// propagates ipc_message_format return
}

struct ipc_error ipc_write_fd (int fd, const struct ipc_message *m)
{
	char *buf = NULL;
	size_t msize = 0;
	ipc_message_format_write (m, &buf, &msize);

	size_t nbytes_sent = 0;
	TEST_IPC_RETURN_ON_ERROR_FREE (usock_send (fd, buf, msize, &nbytes_sent), buf);

	if (buf != NULL) {
		free (buf);
	}
	// what was sent != what should have been sent
	T_R ((nbytes_sent != msize), IPC_ERROR_WRITE_FD__NOT_ENOUGH_DATA);

	IPC_RETURN_NO_ERROR;
}

// Put the message in the list of messages to send.
struct ipc_error ipc_write (struct ipc_ctx *ctx, const struct ipc_message *m)
{
	int found = 0;
	for (size_t i = 0; i < ctx->size; i++) {
		if (ctx->pollfd[i].fd == m->fd) {
			ctx->pollfd[i].events |= POLLOUT;
			found = 1;
		}
	}

	T_R ((found == 0), IPC_ERROR_WRITE__FD_NOT_FOUND);

	// Performs a deep copy of the structure.
	return ipc_messages_add (&ctx->tx, m);
}

/**
 * Allocate memory then add a new connection to the context.
 */
struct ipc_error ipc_add ( struct ipc_ctx *ctx, struct ipc_connection_info *p, struct pollfd *pollfd)
{
	T_R ((ctx    == NULL), IPC_ERROR_ADD__NO_PARAM_CLIENTS);
	T_R ((p      == NULL), IPC_ERROR_ADD__NO_PARAM_CLIENT);
	T_R ((pollfd == NULL), IPC_ERROR_ADD__NO_PARAM_POLLFD);

	printf ("ipc add: add fd %d\n", pollfd[0].fd);

	// Memory reallocation.
	ipc_ctx_new_alloc (ctx);

	ctx->cinfos[ctx->size - 1] = *p;
	ctx->pollfd[ctx->size - 1] = *pollfd;

	IPC_RETURN_NO_ERROR;
}

struct ipc_error ipc_del (struct ipc_ctx *ctx, uint32_t index)
{
	T_R ((ctx == NULL),                                IPC_ERROR_DEL__NO_CLIENTS_PARAM);
	T_R ((ctx->cinfos == NULL || ctx->pollfd == NULL), IPC_ERROR_DEL__EMPTY_LIST);

	if (index >= ctx->size) {
		IPC_RETURN_ERROR (IPC_ERROR_DEL__CANNOT_FIND_CLIENT);
	}

	if (ctx->cinfos[index].spath != NULL)
		free (ctx->cinfos[index].spath);

	ctx->size--;

	if (ctx->size == 0) {
		// free ctx->cinfos and ctx->pollfd
		ipc_ctx_free (ctx);
		IPC_RETURN_NO_ERROR;
	}

	// The last element in the array replaces the removed one.
	ctx->cinfos[index] = ctx->cinfos[ctx->size];
	ctx->pollfd[index] = ctx->pollfd[ctx->size];

	// Reallocation of the arrays. TODO: should be optimised someday.
	ctx->cinfos = realloc (ctx->cinfos, sizeof (struct ipc_connection_info) * ctx->size);
	ctx->pollfd = realloc (ctx->pollfd, sizeof (struct pollfd             ) * ctx->size);

	if (ctx->cinfos == NULL || ctx->pollfd == NULL) {
		IPC_RETURN_ERROR (IPC_ERROR_DEL__EMPTIED_LIST);
	}

	IPC_RETURN_NO_ERROR;
}

// add an arbitrary file descriptor to read
struct ipc_error ipc_add_fd (struct ipc_ctx *ctx, int fd)
{
	T_R ((ctx == NULL), IPC_ERROR_ADD_FD__NO_PARAM_CINFOS);

	SECURE_DECLARATION (struct ipc_connection_info, cinfo);
	cinfo.type = IPC_CONNECTION_TYPE_EXTERNAL;

	SECURE_DECLARATION (struct pollfd, pollfd);
	pollfd.fd = fd;
	pollfd.events = POLLIN;

	return ipc_add (ctx, &cinfo, &pollfd);
}

// remove a connection from its file descriptor
struct ipc_error ipc_del_fd (struct ipc_ctx *ctx, int fd)
{
	T_R ((ctx == NULL),                        IPC_ERROR_DEL_FD__NO_PARAM_CINFOS);
	T_R ((ctx->cinfos == NULL || ctx->pollfd), IPC_ERROR_DEL_FD__EMPTY_LIST);

	for (size_t i = 0; i < ctx->size; i++) {
		if (ctx->pollfd[i].fd == fd) {
			return ipc_del (ctx, i);
		}
	}

	IPC_RETURN_ERROR (IPC_ERROR_DEL_FD__CANNOT_FIND_CLIENT);
}


struct ipc_error handle_writing_message (struct ipc_event *event, struct ipc_ctx *ctx, uint32_t index)
{
	int txfd = ctx->pollfd[index].fd;
	int mfd;
	struct ipc_message *m;
	for (size_t i = 0; ctx->tx.size ; i++) {
		m = &ctx->tx.messages[i];
		mfd = m->fd;
		if (txfd == mfd) {
			TEST_IPC_RR (ipc_write_fd (txfd, m), "cannot send a message to the client");

			printf ("Freeing the message\n");
			ipc_message_empty (m);
			printf ("Removing the message\n");
			ipc_messages_del (&ctx->tx, i); // remove the message indexed by i
		}
	}

	IPC_EVENT_SET (event, IPC_EVENT_TYPE_TX, index, ctx->pollfd[index].fd, NULL);
	IPC_RETURN_NO_ERROR;
}

// new message
struct ipc_error handle_new_message (struct ipc_event *event, struct ipc_ctx *ctx, int index)
{

	// If the socket is associated to another one for networkd:
	// read and write automatically and provide a new IPC_EVENT_TYPE indicating the switch.
	if (ctx->switchdb.collection != NULL) {
		int talkingfd = ctx->pollfd[index].fd;
		int correspondingfd = ipc_switching_get (&ctx->switchdb, talkingfd);
		if (correspondingfd != -1) {
			char *buf = NULL;
			size_t msize = 0;

			{   /** Some macros use "ret" as a variable name, so this is to be sure. */
				struct ipc_error ret = usock_recv (talkingfd, &buf, &msize);
				if (ret.error_code != IPC_ERROR_NONE && ret.error_code != IPC_ERROR_CLOSED_RECIPIENT) {
					if (buf != NULL)
						free (buf);
					IPC_EVENT_SET (event, IPC_EVENT_TYPE_ERROR, index, talkingfd, NULL)
					return ret;
				}
			}

			/** There is a message, send it to the corresponding fd **/
			if (msize > 0) {
				size_t nbytes_sent = 0;
				TEST_IPC_RETURN_ON_ERROR_FREE (usock_send (correspondingfd, buf, msize, &nbytes_sent), buf);

				if (nbytes_sent != msize) {
					// LOG_ERROR ("wrote not enough data from %d to fd %d", talkingfd, correspondingfd);
					IPC_EVENT_SET (event, IPC_EVENT_TYPE_ERROR, index, ctx->pollfd[index].fd, NULL);
					IPC_RETURN_NO_ERROR;	// FIXME: return something else, maybe?
				}
				// LOG_DEBUG ("received a message on fd %d => switch to fd %d", talkingfd, correspondingfd);

				if (buf != NULL)
					free (buf);

				// Everything is OK: inform networkd of a successful transfer.
				IPC_EVENT_SET (event, IPC_EVENT_TYPE_SWITCH, index, ctx->pollfd[index].fd, NULL);
				IPC_RETURN_NO_ERROR;
			} else if (msize == 0) {
				int delfd;

				delfd = ipc_switching_del (&ctx->switchdb, talkingfd);
				if (delfd >= 0) {
					close (delfd);
					ipc_del_fd (ctx, delfd);
				}

				close (talkingfd);
				ipc_del_fd (ctx, talkingfd);

#if 0
				if (delfd >= 0) {
					LOG_DEBUG ("disconnection of %d (and related fd %d)", talkingfd, delfd);
				} else {
					LOG_DEBUG ("disconnection of %d", talkingfd);
				}
#endif

				IPC_EVENT_SET (event, IPC_EVENT_TYPE_DISCONNECTION, index, ctx->pollfd[index].fd, NULL);
				IPC_RETURN_ERROR (IPC_ERROR_CLOSED_RECIPIENT);
			}
		}
	}

	// no treatment of the socket if external socket
	if (ctx->cinfos[index].type == IPC_CONNECTION_TYPE_EXTERNAL) {
		IPC_EVENT_SET (event, IPC_EVENT_TYPE_EXTRA_SOCKET, index, ctx->pollfd[index].fd, NULL);
		IPC_RETURN_NO_ERROR;
	}

	// listen to what they have to say (disconnection or message)
	// then add a client to `event`, the ipc_event structure
	SECURE_DECLARATION (struct ipc_error, ret);
	struct ipc_message *m = NULL;

	SECURE_BUFFER_HEAP_ALLOCATION_R (m, sizeof (struct ipc_message),, IPC_ERROR_HANDLE_MESSAGE__NOT_ENOUGH_MEMORY);

	// current talking client
	ret = ipc_read (ctx, index, m);
	if (ret.error_code != IPC_ERROR_NONE && ret.error_code != IPC_ERROR_CLOSED_RECIPIENT) {
		struct ipc_error rvalue = ret;	// store the final return value
		ipc_message_empty (m);
		free (m);

		// if there is a problem, just remove the client
		TEST_IPC_P (ipc_close (ctx, index), "cannot close a connection in handle_message");
		TEST_IPC_P (ipc_del   (ctx, index), "cannot delete a connection in handle_message");

		IPC_EVENT_SET (event, IPC_EVENT_TYPE_ERROR, index, ctx->pollfd[index].fd, NULL);
		return rvalue;
	}

	// disconnection: close the client then delete it from ctx
	if (ret.error_code == IPC_ERROR_CLOSED_RECIPIENT) {

		IPC_EVENT_SET (event, IPC_EVENT_TYPE_DISCONNECTION, index, ctx->pollfd[index].fd, NULL);

		TEST_IPC_P (ipc_close (ctx, index), "cannot close a connection on closed recipient in handle_message");
		TEST_IPC_P (ipc_del   (ctx, index), "cannot delete a connection on closed recipient in handle_message");

		ipc_message_empty (m);
		free (m);

		// warning: do not forget to free the ipc_client structure
		IPC_RETURN_NO_ERROR;
	}

	// The message carries the fd it was received on.
	m->fd = ctx->pollfd[index].fd;
	IPC_EVENT_SET (event, IPC_EVENT_TYPE_MESSAGE, index, ctx->pollfd[index].fd, m);
	IPC_RETURN_NO_ERROR;
}

/* timer is in ms */
struct ipc_error ipc_events_loop (struct ipc_ctx *ctx, struct ipc_event *event, int *timer)
{
	T_R ((ctx == NULL), IPC_ERROR_WAIT_EVENT__NO_CLIENTS_PARAM);
	T_R ((event  == NULL), IPC_ERROR_WAIT_EVENT__NO_EVENT_PARAM);

	IPC_EVENT_CLEAN (event);

	int32_t n;

#if 1
	printf ("%ld fd for the poll(2) syscall:", ctx->size);
	for (size_t i = 0; i < ctx->size; i++) {
		printf (" %d(%d)", ctx->pollfd[i].fd, ctx->pollfd[i].events);
	}
	printf ("\n");

	printf ("%ld messages still to write, for fd:", ctx->tx.size);
	for (size_t i = 0; i < ctx->tx.size; i++) {
		printf (" %d", ctx->tx.messages[i].fd);
	}
	printf ("\n");
#endif

	for (size_t i = 0; i < ctx->tx.size; i++) {
		for (size_t y = 0; y < ctx->size; y++) {
			if (ctx->pollfd[y].fd == ctx->tx.messages[i].fd) {
				ctx->pollfd[y].events |= POLLOUT;
				break;
			}
		}
	}

	if ((n = poll(ctx->pollfd, ctx->size, *timer)) < 0) {
		printf("select error\n");
		exit(1); // TODO FIXME
	}

	// Timeout.
	if (n == 0) {
		IPC_EVENT_SET (event, IPC_EVENT_TYPE_TIMER, 0, 0, NULL);
		IPC_RETURN_NO_ERROR;
	}

	for (size_t i = 0; i <= ctx->size; i++) {
		// Something to read or connection.
		if (ctx->pollfd[i].revents & POLLIN) {
			// In case there is something to read for the server socket: new client.
			if (ctx->cinfos[i].type == IPC_CONNECTION_TYPE_SERVER) {
				return ipc_accept_add (event, ctx, i);
			}

			return handle_new_message (event, ctx, i);
		}

		// Something can be sent.
		if (ctx->pollfd[i].revents & POLLOUT) {
			ctx->pollfd[i].events &= ~POLLOUT;
			return handle_writing_message (event, ctx, i);
		}

		// Disconnection.
		if (ctx->pollfd[i].revents & POLLHUP) {
			/** IPC_EVENT_SET: event, type, index, fd, message */
			IPC_EVENT_SET (event, IPC_EVENT_TYPE_DISCONNECTION, i, ctx->pollfd[i].fd, NULL);
			return ipc_close (ctx, i);
		}

	} /** for loop: end of the message handling */

	IPC_RETURN_NO_ERROR;
}