MEDIUM: queue: use tasklet_instant_wakeup() to wake tasks

It's long been known that queues didn't scale with threads for various
reasons ranging from the cost of the queue lock to the cost of the
massive amount of inter-thread wakeups.

But some recent reports showing deplorable perfs with threads used at
100% CPU helped us notice that the two elements above add on top of
each other:
  - with plenty of inter-thread wakeups, the scheduler takes a lot of
    time to dequeue pending tasks from the shared queue ;
  - the lock held by the scheduler to do this slows down subsequent
    task_wakeup() calls from the the queue that are made under the
    queue's lock
  - the queue's lock slows down addition of new requests to the queue
    and adds up to the number of needed queue entries for a steady
    traffic.

But the cost of the share queue has no reason for being paid because
it had already been paid when process_stream() added the request to
the queue. As such an instant wakeup is perfectly fit for this.

This is exactly what this patch does, it uses tasklet_instant_wakeup()
to dequeue pending requests, which has the effect of not bloating the
shared queue, hence not requiring the global queue lock, which in turn
results in the wakeup to be much faster, and the queue lock to be much
shorter. In the end, a test with 4k concurrent connections that was
being limited to 40-80k requests/s before with 16 threads, some of
which were stuck at 100% CPU now reaches 570k req/s with 4% idle.

Given that it's been found that it was possible to trigger the watchdog
on the queue lock under extreme conditions, and that such conditions
could happen when users want to protect their servers during a DoS, it
would definitely make sense to backport it to the most recent releases
(2.5 and 2.4 seem like good candidates especially because their scheduler
is modern enough to receive the change above). If a backport is performed,
the following patch is needed:

    MINOR: task: add a new task_instant_wakeup() function

diff --git a/src/queue.c b/src/queue.c
index 9c34a4719b6b08ab31a3fbc73962f9c171f076df..c2dd981d6b16e343f7317e61ddce684147d88772 100644 (file)
--- a/src/queue.c
+++ b/src/queue.c
@@ -322,7 +322,7 @@ static int pendconn_process_next_strm(struct server *srv, struct proxy *px, int
         * under us since the task will need to take the lock anyway and to wait
         * if it wakes up on a different thread.
         */
-       task_wakeup(pp->strm->task, TASK_WOKEN_RES);
+       task_instant_wakeup(pp->strm->task, TASK_WOKEN_RES);
        HA_SPIN_UNLOCK(QUEUE_LOCK, &pp->del_lock);
 
        _HA_ATOMIC_DEC(&px->queue.length);
@@ -343,7 +343,7 @@ static int pendconn_process_next_strm(struct server *srv, struct proxy *px, int
         * under us since the task will need to take the lock anyway and to wait
         * if it wakes up on a different thread.
         */
-       task_wakeup(p->strm->task, TASK_WOKEN_RES);
+       task_instant_wakeup(p->strm->task, TASK_WOKEN_RES);
        __pendconn_unlink_srv(p);
 
        _HA_ATOMIC_DEC(&srv->queue.length);
@@ -497,7 +497,7 @@ int pendconn_redistribute(struct server *s)
                __pendconn_unlink_srv(p);
                p->strm_flags &= ~(SF_DIRECT | SF_ASSIGNED | SF_ADDR_SET);
 
-               task_wakeup(p->strm->task, TASK_WOKEN_RES);
+               task_instant_wakeup(p->strm->task, TASK_WOKEN_RES);
                xferred++;
        }
        HA_SPIN_UNLOCK(QUEUE_LOCK, &s->queue.lock);
@@ -541,7 +541,7 @@ int pendconn_grab_from_px(struct server *s)
                __pendconn_unlink_prx(p);
                p->target = s;
 
-               task_wakeup(p->strm->task, TASK_WOKEN_RES);
+               task_instant_wakeup(p->strm->task, TASK_WOKEN_RES);
                xferred++;
        }
        HA_SPIN_UNLOCK(QUEUE_LOCK, &s->proxy->queue.lock);