fix(ipc): handle multiple waiters

Signed-off-by: Marco Casaroli <marco.casaroli@gmail.com>

Author: casaroli

src/samples/supervisor/posix/client/client.c

@@ -117,6 +117,10 @@ struct ocre_container *ocre_context_create_container(struct ocre_context *contex
 	uint8_t payload[LARGE_PAYLOAD_SIZE];
 	bool success;
 
+	if (!context) {
+		return NULL;
+	}
+
 	/* Encode the request */
 	ZCBOR_STATE_E(encoding_state, 0, payload, sizeof(payload), 0);
 

src/samples/supervisor/posix/server/handlers.c

@@ -36,18 +36,6 @@
 #define MAX_CONTAINERS	   64
 #define MAX_STRING_ARRAY   32
 
-/* Global ocre context for the server */
-// static struct ocre_context *ctx = NULL;
-
-// static void print_hex(const char *label, const uint8_t *data, size_t len)
-// {
-// 	printf("HEX %s: ", label);
-// 	for (size_t i = 0; i < len; i++) {
-// 		printf("%02x ", data[i]);
-// 	}
-// 	printf("\n");
-// }
-
 static int handle_context_create_container(struct ocre_context *ctx, zcbor_state_t *dec_state, uint8_t *tx_buf,
 					   size_t tx_buf_size)
 {

src/samples/supervisor/posix/server/waiters.c

@@ -5,11 +5,13 @@
  * SPDX-License-Identifier: Apache-2.0
  */
 
+#include <pthread.h>
 #include <stdlib.h>
 #include <string.h>
 #include <unistd.h>
 #include <stdio.h>
 #include <stdint.h>
+#include <poll.h>
 
 #include <sys/socket.h>
 
@@ -28,19 +30,28 @@ struct client {
 };
 
 struct waiter {
+	pthread_mutex_t mutex;
+	bool finished;
+	int exit_status;
+	int result;
 	struct ocre_container *container;
-	pthread_t thread;
+	pthread_t wait_thread;
+	pthread_t socket_thread;
 	struct client *clients;
 	struct waiter *next;
 };
 
 static struct waiter *waiters = NULL;
 
-static void *container_wait_thread(void *arg)
+static pthread_mutex_t mutex = PTHREAD_MUTEX_INITIALIZER;
+
+static void *wait_thread(void *arg)
 {
 	uint8_t tx_buf[TX_BUFFER_SIZE];
 	struct waiter *waiter = (struct waiter *)arg;
 
+	fprintf(stderr, "Wait thread started\n");
+
 	/* Call the actual function */
 	waiter->result = ocre_container_wait(waiter->container, &waiter->exit_status);
 
@@ -82,6 +93,8 @@ static void *container_wait_thread(void *arg)
 	struct client *client = NULL;
 	struct client *elt = NULL;
 
+	pthread_mutex_lock(&waiter->mutex);
+
 	LL_FOREACH_SAFE(waiter->clients, client, elt)
 	{
 		if (send(client->socket, tx_buf, response_len, 0) < 0) {
@@ -97,7 +110,93 @@ static void *container_wait_thread(void *arg)
 		LL_DELETE(waiter->clients, client);
 	}
 
-	LL_DELETE(waiters, waiter);
+	fprintf(stderr, "WAIT FOR SOCKET THREAD TO FINISH\n");
+
+	waiter->finished = true;
+
+	pthread_mutex_unlock(&waiter->mutex);
+
+	// if (!waiter->finished) {
+	/* Wait for the socket thread to finish */
+	pthread_join(waiter->socket_thread, NULL);
+	// }
+
+	fprintf(stderr, "SOCKET THREAD FINISHED\n");
+
+	fprintf(stderr, "returned\n");
+
+	return NULL;
+}
+
+static void *socket_thread(void *arg)
+{
+	struct client *client = NULL;
+	struct waiter *waiter = (struct waiter *)arg;
+
+	while (true) {
+		int count = 0;
+		pthread_mutex_lock(&waiter->mutex);
+
+		if (waiter->finished) {
+			pthread_mutex_unlock(&waiter->mutex);
+			return NULL;
+		};
+
+		LL_COUNT(waiter->clients, client, count);
+
+		fprintf(stderr, "Number of clients: %d.\n", count);
+
+		uint8_t rx_buf[RX_BUFFER_SIZE];
+
+		struct pollfd *pfds = malloc(sizeof(struct pollfd) * count);
+		if (!pfds) {
+			perror("malloc");
+			pthread_mutex_unlock(&waiter->mutex);
+			return NULL;
+		}
+
+		int i = 0;
+		LL_FOREACH(waiter->clients, client)
+		{
+			pfds[i].fd = client->socket;
+			pfds[i].events = POLLIN;
+			i++;
+		}
+
+		pthread_mutex_unlock(&waiter->mutex);
+
+		int ret = poll(pfds, count, 2500);
+		if (ret < 0) {
+			perror("poll");
+			return NULL;
+		}
+
+		for (i = 0; i < count; i++) {
+			if (pfds[i].revents & POLLIN) {
+				fprintf(stderr, "poll in from %d\n", pfds[i].fd);
+				ssize_t n = recv(pfds[i].fd, rx_buf, sizeof(rx_buf), 0);
+				if (n <= 0) {
+					if (n < 0)
+						perror("recv");
+					else
+						printf("Client disconnected\n");
+
+					close(pfds[i].fd);
+
+					pthread_mutex_lock(&waiter->mutex);
+
+					pthread_mutex_unlock(&waiter->mutex);
+					break;
+				}
+
+				// if (recv(pfds[i].fd, rx_buf, RX_BUFFER_SIZE, 0) < 0) {
+				// 	perror("recv");
+				// }
+			}
+		}
+
+		free(pfds);
+	}
 
 	return NULL;
 }
@@ -125,9 +224,26 @@ static struct waiter *waiter_get_or_new(struct ocre_container *container)
 
 	waiter->container = container;
 
-	int rc = pthread_create(&waiter->thread, NULL, container_wait_thread, waiter);
+	int rc = pthread_mutex_init(&waiter->mutex, NULL);
 	if (rc) {
-		fprintf(stderr, "Failed to create thread: rc=%d", rc);
+		fprintf(stderr, "Failed to initialize mutex: rc=%d", rc);
+		free(waiter);
+		return NULL;
+	}
+
+	pthread_mutex_lock(&waiter->mutex);
+
+	fprintf(stderr, "Creating socket thread for container %p\n", container);
+	rc = pthread_create(&waiter->socket_thread, NULL, socket_thread, waiter);
+	if (rc) {
+		fprintf(stderr, "Failed to create socket thread: rc=%d", rc);
+		free(waiter);
+		return NULL;
+	}
+
+	rc = pthread_create(&waiter->wait_thread, NULL, wait_thread, waiter);
+	if (rc) {
+		fprintf(stderr, "Failed to create waiter thread: rc=%d", rc);
 		free(waiter);
 		return NULL;
 	}
@@ -153,8 +269,12 @@ static int waiter_add_client(struct waiter *waiter, int socket)
 
 	client->socket = socket;
 
+	pthread_mutex_lock(&waiter->mutex);
+
 	LL_APPEND(waiter->clients, client);
 
+	pthread_mutex_unlock(&waiter->mutex);
+
 	return 0;
 }