MAJOR: applets: Use tasks, instead of rolling our own scheduler.
There's no real reason to have a specific scheduler for applets anymore, so
nuke it and just use tasks. This comes with some benefits, the first one
being that applets cannot induce high latencies anymore since they share
nice values with other tasks. Later it will be possible to configure the
applets' nice value. The second benefit is that the applet scheduler was
not very thread-friendly, having a big lock around it in prevision of this
change. Thus applet-intensive workloads should now scale much better with
threads.
Some more improvement is possible now : some applets also use a task to
handle timers and timeouts. These ones could now be simplified to use only
one task.
diff --git a/src/applet.c b/src/applet.c
index 77f984d..d7fbb53 100644
--- a/src/applet.c
+++ b/src/applet.c
@@ -19,100 +19,36 @@
#include <proto/channel.h>
#include <proto/stream.h>
#include <proto/stream_interface.h>
+#include <proto/task.h>
unsigned int nb_applets = 0;
-unsigned long active_applets_mask = 0;
-unsigned int applets_active_queue = 0;
-__decl_hathreads(HA_SPINLOCK_T applet_active_lock); /* spin lock related to applet active queue */
-struct list applet_active_queue = LIST_HEAD_INIT(applet_active_queue);
-
-void applet_run_active()
+struct task *task_run_applet(struct task *t, void *context, unsigned short state)
{
- struct appctx *curr, *next;
- struct stream_interface *si;
- struct list applet_cur_queue = LIST_HEAD_INIT(applet_cur_queue);
- int max_processed;
+ struct appctx *app = context;
+ struct stream_interface *si = app->owner;
- max_processed = applets_active_queue;
- if (max_processed > 200)
- max_processed = 200;
-
- HA_SPIN_LOCK(APPLETS_LOCK, &applet_active_lock);
- if (!(active_applets_mask & tid_bit)) {
- HA_SPIN_UNLOCK(APPLETS_LOCK, &applet_active_lock);
- return;
+ if (app->state & APPLET_WANT_DIE) {
+ __appctx_free(app);
+ return NULL;
}
- active_applets_mask &= ~tid_bit;
- curr = LIST_NEXT(&applet_active_queue, typeof(curr), runq);
- while (&curr->runq != &applet_active_queue) {
- next = LIST_NEXT(&curr->runq, typeof(next), runq);
- if (curr->thread_mask & tid_bit) {
- LIST_DEL(&curr->runq);
- curr->state = APPLET_RUNNING;
- LIST_ADDQ(&applet_cur_queue, &curr->runq);
- applets_active_queue--;
- max_processed--;
- }
- curr = next;
- if (max_processed <= 0) {
- active_applets_mask |= tid_bit;
- break;
- }
- }
- HA_SPIN_UNLOCK(APPLETS_LOCK, &applet_active_lock);
+ /* Now we'll try to allocate the input buffer. We wake up the
+ * applet in all cases. So this is the applet responsibility to
+ * check if this buffer was allocated or not. This let a chance
+ * for applets to do some other processing if needed. */
+ if (!channel_alloc_buffer(si_ic(si), &app->buffer_wait))
+ si_applet_cant_put(si);
- /* The list is only scanned from the head. This guarantees that if any
- * applet removes another one, there is no side effect while walking
- * through the list.
+ /* We always pretend the applet can't get and doesn't want to
+ * put, it's up to it to change this if needed. This ensures
+ * that one applet which ignores any event will not spin.
*/
- while (!LIST_ISEMPTY(&applet_cur_queue)) {
- curr = LIST_ELEM(applet_cur_queue.n, typeof(curr), runq);
- si = curr->owner;
+ si_applet_cant_get(si);
+ si_applet_stop_put(si);
- /* Now we'll try to allocate the input buffer. We wake up the
- * applet in all cases. So this is the applet responsibility to
- * check if this buffer was allocated or not. This let a chance
- * for applets to do some other processing if needed. */
- if (!channel_alloc_buffer(si_ic(si), &curr->buffer_wait))
- si_applet_cant_put(si);
-
- /* We always pretend the applet can't get and doesn't want to
- * put, it's up to it to change this if needed. This ensures
- * that one applet which ignores any event will not spin.
- */
- si_applet_cant_get(si);
- si_applet_stop_put(si);
-
- curr->applet->fct(curr);
- si_applet_wake_cb(si);
- channel_release_buffer(si_ic(si), &curr->buffer_wait);
-
- if (applet_cur_queue.n == &curr->runq) {
- /* curr was left in the list, move it back to the active list */
- LIST_DEL(&curr->runq);
- LIST_INIT(&curr->runq);
- HA_SPIN_LOCK(APPLETS_LOCK, &applet_active_lock);
- if (curr->state & APPLET_WANT_DIE) {
- curr->state = APPLET_SLEEPING;
- __appctx_free(curr);
- }
- else {
- if (curr->state & APPLET_WOKEN_UP) {
- curr->state = APPLET_SLEEPING;
- __appctx_wakeup(curr);
- }
- else {
- curr->state = APPLET_SLEEPING;
- }
- }
- HA_SPIN_UNLOCK(APPLETS_LOCK, &applet_active_lock);
- }
- }
+ app->applet->fct(app);
+ si_applet_wake_cb(si);
+ channel_release_buffer(si_ic(si), &app->buffer_wait);
+ return t;
}
-__attribute__((constructor))
-static void __applet_init(void)
-{
- HA_SPIN_INIT(&applet_active_lock);
-}
diff --git a/src/cfgparse.c b/src/cfgparse.c
index 023973f..3d224b9 100644
--- a/src/cfgparse.c
+++ b/src/cfgparse.c
@@ -84,6 +84,7 @@
#include <proto/stick_table.h>
#include <proto/task.h>
#include <proto/tcp_rules.h>
+#include <proto/connection.h>
/* This is the SSLv3 CLIENT HELLO packet used in conjunction with the
diff --git a/src/cli.c b/src/cli.c
index 4adf684..c34078f 100644
--- a/src/cli.c
+++ b/src/cli.c
@@ -953,7 +953,6 @@
chunk_appendf(&trash, "\nloops:"); for (thr = 0; thr < global.nbthread; thr++) chunk_appendf(&trash, " %u", activity[thr].loops);
chunk_appendf(&trash, "\nwake_cache:"); for (thr = 0; thr < global.nbthread; thr++) chunk_appendf(&trash, " %u", activity[thr].wake_cache);
chunk_appendf(&trash, "\nwake_tasks:"); for (thr = 0; thr < global.nbthread; thr++) chunk_appendf(&trash, " %u", activity[thr].wake_tasks);
- chunk_appendf(&trash, "\nwake_applets:"); for (thr = 0; thr < global.nbthread; thr++) chunk_appendf(&trash, " %u", activity[thr].wake_applets);
chunk_appendf(&trash, "\nwake_signal:"); for (thr = 0; thr < global.nbthread; thr++) chunk_appendf(&trash, " %u", activity[thr].wake_signal);
chunk_appendf(&trash, "\npoll_exp:"); for (thr = 0; thr < global.nbthread; thr++) chunk_appendf(&trash, " %u", activity[thr].poll_exp);
chunk_appendf(&trash, "\npoll_drop:"); for (thr = 0; thr < global.nbthread; thr++) chunk_appendf(&trash, " %u", activity[thr].poll_drop);
diff --git a/src/flt_spoe.c b/src/flt_spoe.c
index 856050d..c256303 100644
--- a/src/flt_spoe.c
+++ b/src/flt_spoe.c
@@ -2840,8 +2840,7 @@
/* Release the buffer if needed */
if ((*buf)->size) {
b_free(buf);
- offer_buffers(buffer_wait->target,
- tasks_run_queue + applets_active_queue);
+ offer_buffers(buffer_wait->target, tasks_run_queue);
}
}
diff --git a/src/haproxy.c b/src/haproxy.c
index 4e61e40..766758f 100644
--- a/src/haproxy.c
+++ b/src/haproxy.c
@@ -90,7 +90,6 @@
#include <types/peers.h>
#include <proto/acl.h>
-#include <proto/applet.h>
#include <proto/arg.h>
#include <proto/auth.h>
#include <proto/backend.h>
@@ -2419,8 +2418,6 @@
activity[tid].wake_cache++;
else if (active_tasks_mask & tid_bit)
activity[tid].wake_tasks++;
- else if (active_applets_mask & tid_bit)
- activity[tid].wake_applets++;
else if (signal_queue_len)
activity[tid].wake_signal++;
else
@@ -2429,7 +2426,6 @@
/* The poller will ensure it returns around <next> */
cur_poller.poll(&cur_poller, exp);
fd_process_cached_events();
- applet_run_active();
/* Synchronize all polling loops */
diff --git a/src/mux_h2.c b/src/mux_h2.c
index 9c7b828..57b1722 100644
--- a/src/mux_h2.c
+++ b/src/mux_h2.c
@@ -17,7 +17,6 @@
#include <common/hpack-enc.h>
#include <common/hpack-tbl.h>
#include <common/net_helper.h>
-#include <proto/applet.h>
#include <proto/connection.h>
#include <proto/h1.h>
#include <proto/stream.h>
@@ -303,8 +302,7 @@
{
if ((*bptr)->size) {
b_free(bptr);
- offer_buffers(h2c->buf_wait.target,
- tasks_run_queue + applets_active_queue);
+ offer_buffers(h2c->buf_wait.target, tasks_run_queue);
}
}
diff --git a/src/stream.c b/src/stream.c
index 3ea8953..7ad84e9 100644
--- a/src/stream.c
+++ b/src/stream.c
@@ -339,7 +339,7 @@
if (s->req.buf->size || s->res.buf->size) {
b_drop(&s->req.buf);
b_drop(&s->res.buf);
- offer_buffers(NULL, tasks_run_queue + applets_active_queue);
+ offer_buffers(NULL, tasks_run_queue);
}
hlua_ctx_destroy(s->hlua);
@@ -469,7 +469,7 @@
* someone waiting, we can wake up a waiter and offer them.
*/
if (offer)
- offer_buffers(s, tasks_run_queue + applets_active_queue);
+ offer_buffers(s, tasks_run_queue);
}
/* perform minimal intializations, report 0 in case of error, 1 if OK. */