From 3c595ac3adf49abd305efbab4a7b11175d10cb0e Mon Sep 17 00:00:00 2001
From: Willy Tarreau <w@1wt.eu>
Date: Sun, 19 Apr 2015 09:59:31 +0200
Subject: [PATCH] MEDIUM: applet: implement a run queue for active appctx

The new function is called for each round of polling in order to call any
active appctx. For now we pick the stream interface from the appctx's
owner. At the moment there's no appctx queued yet, but we have everything
needed to queue them and remove them.
---
 include/proto/applet.h | 11 +++++++++++
 src/applet.c           | 24 +++++++++++++++++++++++-
 src/haproxy.c          |  6 ++++--
 3 files changed, 38 insertions(+), 3 deletions(-)

diff --git a/include/proto/applet.h b/include/proto/applet.h
index 4e43bbe5b..3e68b0085 100644
--- a/include/proto/applet.h
+++ b/include/proto/applet.h
@@ -31,6 +31,8 @@
 
 extern struct list applet_runq;
 
+void applet_run_active();
+
 /* Initializes all required fields for a new appctx. Note that it does the
  * minimum acceptable initialization for an appctx. This means only the
  * 3 integer states st0, st1, st2 are zeroed.
@@ -76,6 +78,15 @@ static inline void appctx_wakeup(struct appctx *appctx)
 		LIST_ADDQ(&applet_runq, &appctx->runq);
 }
 
+/* removes an applet from the list of active applets */
+static inline void appctx_pause(struct appctx *appctx)
+{
+	if (!LIST_ISEMPTY(&appctx->runq)) {
+		LIST_DEL(&appctx->runq);
+		LIST_INIT(&appctx->runq);
+	}
+}
+
 #endif /* _PROTO_APPLET_H */
 
 /*
diff --git a/src/applet.c b/src/applet.c
index ca3ead1b3..3aa67e72c 100644
--- a/src/applet.c
+++ b/src/applet.c
@@ -16,6 +16,28 @@
 #include <common/config.h>
 #include <common/mini-clist.h>
 #include <proto/applet.h>
-
+#include <proto/stream.h>
+#include <proto/stream_interface.h>
 
 struct list applet_runq = LIST_HEAD_INIT(applet_runq);
+
+void applet_run_active()
+{
+	struct appctx *curr, *back;
+	struct stream_interface *si;
+
+	list_for_each_entry_safe(curr, back, &applet_runq, runq) {
+		si = curr->owner;
+
+		/* now we'll need a buffer */
+		if (!stream_alloc_recv_buffer(si_ic(si))) {
+			si->flags |= SI_FL_WAIT_ROOM;
+			LIST_DEL(&curr->runq);
+			LIST_INIT(&curr->runq);
+			continue;
+		}
+
+		curr->applet->fct(curr);
+		/* must not dereference curr nor si now because it might have been freed */
+	}
+}
diff --git a/src/haproxy.c b/src/haproxy.c
index 1a913e37a..474179cf1 100644
--- a/src/haproxy.c
+++ b/src/haproxy.c
@@ -79,9 +79,10 @@
 #include <types/acl.h>
 #include <types/peers.h>
 
-#include <proto/auth.h>
 #include <proto/acl.h>
+#include <proto/applet.h>
 #include <proto/arg.h>
+#include <proto/auth.h>
 #include <proto/backend.h>
 #include <proto/channel.h>
 #include <proto/checks.h>
@@ -1487,12 +1488,13 @@ void run_poll_loop()
 			break;
 
 		/* expire immediately if events are pending */
-		if (fd_cache_num || run_queue || signal_queue_len)
+		if (fd_cache_num || run_queue || signal_queue_len || !LIST_ISEMPTY(&applet_runq))
 			next = now_ms;
 
 		/* The poller will ensure it returns around <next> */
 		cur_poller.poll(&cur_poller, next);
 		fd_process_cached_events();
+		applet_run_active();
 	}
 }