use upoll for event loop

also add '#pragma once' in headers

Change-Id: I90cc2d53ec60d4d1e66cf0f229109621e4019864
Signed-off-by: José Bollo <jose.bollo@iot.bzh>

16 files changed, 174 insertions, 47 deletions

diff --git a/src/afb-api-so.c b/src/afb-api-so.c
index 2616e0bc..e31216e0 100644
--- a/src/afb-api-so.c
+++ b/src/afb-api-so.c
@@ -56,21 +56,15 @@ static int api_timeout = 15;
 
 static const char plugin_register_function[] = "pluginRegister";
 
-static const struct afb_poll_itf upoll_itf = {
+static const struct afb_pollitf upollitf = {
+	.wait = (void*)upoll_wait,
+	.open = (void*)upoll_open,
 	.on_readable = (void*)upoll_on_readable,
 	.on_writable = (void*)upoll_on_writable,
 	.on_hangup = (void*)upoll_on_hangup,
 	.close = (void*)upoll_close
 };
 
-static struct afb_poll itf_poll_open(int fd, void *closure)
-{
-	struct afb_poll result;
-	result.data = upoll_open(fd, closure);
-	result.itf = result.data ? &upoll_itf : NULL;
-	return result;
-}
-
 static void free_context(struct api_so_desc *desc, void *context)
 {
 	void (*cb)(void*);
@@ -198,7 +192,8 @@ int afb_api_so_add_plugin(const char *path)
 	/* init the interface */
 	desc->interface.verbosity = 0;
 	desc->interface.mode = AFB_MODE_LOCAL;
-	desc->interface.poll_open = itf_poll_open;
+	desc->interface.pollitf = &upollitf;
+	desc->interface.pollclosure = NULL;
 
 	/* init the plugin */
 	desc->plugin = pluginRegisterFct(&desc->interface);
diff --git a/src/afb-api-so.h b/src/afb-api-so.h
index 9c2a570b..fe4d76b0 100644
--- a/src/afb-api-so.h
+++ b/src/afb-api-so.h
@@ -16,6 +16,8 @@
  */
 
 
+#pragma once
+
 extern int afb_api_so_add_plugin(const char *path);
 
 extern int afb_api_so_add_directory(const char *path);
diff --git a/src/afb-apis.h b/src/afb-apis.h
index 97778b64..59c7008f 100644
--- a/src/afb-apis.h
+++ b/src/afb-apis.h
@@ -15,6 +15,8 @@
  * limitations under the License.
  */
 
+#pragma once
+
 struct afb_req;
 struct AFB_clientCtx;
 
diff --git a/src/afb-config.h b/src/afb-config.h
index bd8d07a0..87691af0 100644
--- a/src/afb-config.h
+++ b/src/afb-config.h
@@ -21,6 +21,8 @@
 #ifndef LOCAL_DEF_H
 #define LOCAL_DEF_H
 
+#pragma once
+
 /* other definitions --------------------------------------------------- */
 
 // Note: because of a bug in libmagic MAGIC_DB NULL should not be used for default
diff --git a/src/afb-hreq.h b/src/afb-hreq.h
index 5c09e4e6..aa2cf117 100644
--- a/src/afb-hreq.h
+++ b/src/afb-hreq.h
@@ -15,7 +15,8 @@
  * limitations under the License.
  */
 
-struct AFB_session;
+#pragma once
+
 struct AFB_clientCtx;
 struct json_object;
 
diff --git a/src/afb-hsrv.h b/src/afb-hsrv.h
index c22f57ee..b422df47 100644
--- a/src/afb-hsrv.h
+++ b/src/afb-hsrv.h
@@ -16,6 +16,8 @@
  limitations under the License.
 */
 
+#pragma once
+
 struct afb_hsrv;
 struct afb_hreq;
 
diff --git a/src/afb-hswitch.h b/src/afb-hswitch.h
index f18b8b14..f70a0bf8 100644
--- a/src/afb-hswitch.h
+++ b/src/afb-hswitch.h
@@ -16,6 +16,8 @@
  * limitations under the License.
  */
 
+#pragma once
+
 struct afb_hreq;
 extern int afb_hswitch_apis(struct afb_hreq *hreq, void *data);
 extern int afb_hswitch_one_page_api_redirect(struct afb_hreq *hreq, void *data);
diff --git a/src/afb-method.h b/src/afb-method.h
index fde9dbfc..1411d29f 100644
--- a/src/afb-method.h
+++ b/src/afb-method.h
@@ -16,6 +16,8 @@
  */
 
 
+#pragma once
+
 enum afb_method {
 	afb_method_none = 0,
 	afb_method_get = 1,
diff --git a/src/afb-websock.h b/src/afb-websock.h
index 4a0a3e55..646e98f8 100644
--- a/src/afb-websock.h
+++ b/src/afb-websock.h
@@ -15,6 +15,8 @@
  * limitations under the License.
  */
 
+#pragma once
+
 struct afb_hreq;
 extern int afb_websock_check_upgrade(struct afb_hreq *hreq);
 
diff --git a/src/afb-ws-json.h b/src/afb-ws-json.h
index 2d3cfd75..c9328634 100644
--- a/src/afb-ws-json.h
+++ b/src/afb-ws-json.h
@@ -15,6 +15,8 @@
  * limitations under the License.
  */
 
+#pragma once
+
 struct afb_ws_json;
 struct AFB_clientCtx;
 
diff --git a/src/afb-ws.h b/src/afb-ws.h
index a8af90ac..1faec65c 100644
--- a/src/afb-ws.h
+++ b/src/afb-ws.h
@@ -15,6 +15,8 @@
  * limitations under the License.
  */
 
+#pragma once
+
 struct afb_ws;
 
 struct afb_ws_itf
diff --git a/src/session.h b/src/session.h
index c11921b6..0a61612d 100644
--- a/src/session.h
+++ b/src/session.h
@@ -16,6 +16,8 @@
  */
 // User Client Session Context
 
+#pragma once
+
 struct AFB_clientCtx
 {
 	time_t expiration;    // expiration time of the token
diff --git a/src/utils-upoll.c b/src/utils-upoll.c
index 6db2246d..d72a0e8b 100644
--- a/src/utils-upoll.c
+++ b/src/utils-upoll.c
@@ -25,9 +25,11 @@
 #include "utils-upoll.h"
 
 
+struct upollfd;
+
 struct upoll
 {
-	int fd;
+	struct upollfd *fd;
 	void (*read)(void *);
 	void (*write)(void *);
 	void (*hangup)(void *);
@@ -35,25 +37,67 @@ struct upoll
 	struct upoll *next;
 };
 
+struct upollfd
+{
+	int fd;
+	uint32_t events;
+	struct upollfd *next;
+	struct upoll *head;
+};
+
 static int pollfd = 0;
-static struct upoll *head = NULL;
+static struct upollfd *head = NULL;
 static pthread_mutex_t mutex = PTHREAD_MUTEX_INITIALIZER;
 
-int upoll_is_valid(struct upoll *upoll)
+static int update(struct upollfd *ufd)
 {
-	struct upoll *it = head;
-	while (it != NULL) {
-		if (it == upoll)
-			return 1;
-		it = it->next;
+	int rc;
+	struct upoll *u;
+	struct epoll_event e;
+	uint32_t events;
+	struct upollfd **prv;
+
+	events = 0;
+	pthread_mutex_lock(&mutex);
+	u = ufd->head;
+	if (u == NULL) {
+		/* no more watchers */
+		prv = &head;
+		while(*prv) {
+			if (*prv == ufd) {
+				*prv = ufd->next;
+				break;
+			}
+			prv = &(*prv)->next;
+		}
+		pthread_mutex_unlock(&mutex);
+		epoll_ctl(pollfd, EPOLL_CTL_DEL, ufd->fd, NULL);
+		free(ufd);
+		return 0;
 	}
-	return 0;
+	/* compute the events for the watchers */
+	while (u != NULL) {
+		if (u->read != NULL)
+			events |= EPOLLIN;
+		if (u->write != NULL)
+			events |= EPOLLOUT;
+		u = u->next;
+	}
+	pthread_mutex_unlock(&mutex);
+	if (ufd->events == events)
+		return 0;
+	e.events = events;
+	e.data.ptr = ufd;
+	rc = epoll_ctl(pollfd, EPOLL_CTL_MOD, ufd->fd, &e);
+	if (rc == 0)
+		ufd->events = events;
+	return rc;
 }
 
-struct upoll *upoll_open(int fd, void *closure)
+static struct upollfd *get_fd(int fd)
 {
 	struct epoll_event e;
-	struct upoll *result;
+	struct upollfd *result;
 	int rc;
 
 	/* opens the epoll stream */
@@ -69,6 +113,14 @@ struct upoll *upoll_open(int fd, void *closure)
 		}
 	}
 
+	/* search */
+	result = head;
+	while (result != NULL) {
+		if (result->fd == fd)
+			return result;
+		result = result->next;
+	}
+
 	/* allocates */
 	result = calloc(1, sizeof *result);
 	if (result == NULL)
@@ -76,7 +128,6 @@ struct upoll *upoll_open(int fd, void *closure)
 
 	/* init */
 	result->fd = fd;
-	result->closure = closure;
 	pthread_mutex_lock(&mutex);
 	result->next = head;
 	head = result;
@@ -91,18 +142,52 @@ struct upoll *upoll_open(int fd, void *closure)
 
 	/* revert on error */
 	rc = errno;
-	upoll_close(result);
+	update(result);
 	errno = rc;
 	return NULL;
 }
 
-static int update(struct upoll *upoll)
+int upoll_is_valid(struct upoll *upoll)
 {
-	struct epoll_event e;
-	e.events = (uint32_t)((upoll->read != NULL ? EPOLLIN : 0 )
-		 | (upoll->write != NULL ? EPOLLOUT : 0));
-	e.data.ptr = upoll;
-	return epoll_ctl(pollfd, EPOLL_CTL_MOD, upoll->fd, &e);
+	struct upollfd *itfd = head;
+	struct upoll *it;
+	while (itfd != NULL) {
+		it = itfd->head;
+		while (it != NULL) {
+			if (it == upoll)
+				return 1;
+			it = it->next;
+		}
+		itfd = itfd->next;
+	}
+	return 0;
+}
+
+struct upoll *upoll_open(int fd, void *closure)
+{
+	struct upollfd *ufd;
+	struct upoll *result;
+
+	/* allocates */
+	result = calloc(1, sizeof *result);
+	if (result == NULL)
+		return NULL;
+
+	/* get for fd */
+	ufd = get_fd(fd);
+	if (ufd == NULL) {
+		free(result);
+		return NULL;
+	}
+
+	/* init */
+	result->fd = ufd;
+	result->closure = closure;
+	pthread_mutex_lock(&mutex);
+	result->next = ufd->head;
+	ufd->head = result;
+	pthread_mutex_unlock(&mutex);
+	return result;
 }
 
 int upoll_on_readable(struct upoll *upoll, void (*process)(void *))
@@ -111,7 +196,7 @@ int upoll_on_readable(struct upoll *upoll, void (*process)(void *))
 	assert(upoll_is_valid(upoll));
 
 	upoll->read = process;
-	return update(upoll);
+	return update(upoll->fd);
 }
 
 int upoll_on_writable(struct upoll *upoll, void (*process)(void *))
@@ -120,7 +205,7 @@ int upoll_on_writable(struct upoll *upoll, void (*process)(void *))
 	assert(upoll_is_valid(upoll));
 
 	upoll->write = process;
-	return update(upoll);
+	return update(upoll->fd);
 }
 
 void upoll_on_hangup(struct upoll *upoll, void (*process)(void *))
@@ -134,38 +219,56 @@ void upoll_on_hangup(struct upoll *upoll, void (*process)(void *))
 void upoll_close(struct upoll *upoll)
 {
 	struct upoll **it;
+	struct upollfd *ufd;
 
 	assert(pollfd != 0);
 	assert(upoll_is_valid(upoll));
 
-	epoll_ctl(pollfd, EPOLL_CTL_DEL, upoll->fd, NULL);
+	ufd = upoll->fd;
 	pthread_mutex_lock(&mutex);
-	it = &head;
+	it = &ufd->head;
 	while (*it != upoll)
 		it = &(*it)->next;
 	*it = upoll->next;
 	pthread_mutex_unlock(&mutex);
 	free(upoll);
+	update(ufd);
 }
 
-void upoll_wait(int timeout)
+int upoll_wait(int timeout)
 {
 	int rc;
 	struct epoll_event e;
-	struct upoll *upoll;
+	struct upollfd *ufd;
+	struct upoll *u;
 
-	if (pollfd == 0)
-		return;
+	if (pollfd == 0) {
+		errno = ECANCELED;
+		return -1;
+	}
 
-	rc = epoll_wait(pollfd, &e, 1, timeout);
+	do {
+		rc = epoll_wait(pollfd, &e, 1, timeout);
+	} while (rc < 0 && errno == EINTR);
 	if (rc == 1) {
-		upoll = e.data.ptr;
-		if ((e.events & EPOLLIN) && upoll->read)
-			upoll->read(upoll->closure);
-		if ((e.events & EPOLLOUT) && upoll->write)
-			upoll->write(upoll->closure);
-		if ((e.events & EPOLLHUP) && upoll->hangup)
-			upoll->hangup(upoll->closure);
+		ufd = e.data.ptr;
+		u = ufd->head;
+		while (u != NULL) {
+			if ((e.events & EPOLLIN) && u->read) {
+				u->read(u->closure);
+				break;
+			}
+			if ((e.events & EPOLLOUT) && u->write) {
+				u->write(u->closure);
+				break;
+			}
+			if ((e.events & EPOLLHUP) && u->hangup) {
+				u->hangup(u->closure);
+				break;
+			}
+			u = u->next;
+		}
 	}
+	return rc < 0 ? rc : 0;
 }
 
diff --git a/src/utils-upoll.h b/src/utils-upoll.h
index 705fbc36..56692d3f 100644
--- a/src/utils-upoll.h
+++ b/src/utils-upoll.h
@@ -15,6 +15,8 @@
  * limitations under the License.
  */
 
+#pragma once
+
 struct upoll;
 
 extern int upoll_is_valid(struct upoll *upoll);
@@ -28,5 +30,7 @@ extern void upoll_on_hangup(struct upoll *upoll, void (*process)(void *closure))
 
 extern void upoll_close(struct upoll *upoll);
 
-extern void upoll_wait(int timeout);
+extern int upoll_wait(int timeout);
+
+
 
diff --git a/src/verbose.h b/src/verbose.h
index 09254b22..25bd85b0 100644
--- a/src/verbose.h
+++ b/src/verbose.h
@@ -16,6 +16,8 @@
  limitations under the License.
 */
 
+#pragma once
+
 #if !defined(NDEBUG)
 #include <syslog.h>
 extern int verbosity;
diff --git a/src/websock.h b/src/websock.h
index c40a364c..7785a0c0 100644
--- a/src/websock.h
+++ b/src/websock.h
@@ -21,6 +21,8 @@
  *   Copyright 2010-2012 self.disconnect (APACHE-2)
  */
 
+#pragma once
+
 struct iovec;
 
 #define WEBSOCKET_CODE_OK                1000