libs: support timeout for workqueue retries

Support an optional timeout/delay for use when a workqueue
determines that it is blocked, instead of retrying immediately.
Also, schedule as an 'event' instead of a 'timer' when using
a zero timeout value.

Signed-off-by: Mark Stapp <mjs@voltanet.io>

diff --git a/lib/workqueue.c b/lib/workqueue.c
index c927d5d714118c4745a99cb6358e518f8ae6ba88..24ef24c7744930fb8978ad40d41112c4a0fad91c 100644 (file)
--- a/lib/workqueue.c
+++ b/lib/workqueue.c
@@ -91,9 +91,10 @@ struct work_queue *work_queue_new(struct thread_master *m,
 
        new->cycles.granularity = WORK_QUEUE_MIN_GRANULARITY;
 
-       /* Default values, can be overriden by caller */
+       /* Default values, can be overridden by caller */
        new->spec.hold = WORK_QUEUE_DEFAULT_HOLD;
        new->spec.yield = THREAD_YIELD_TIME_SLOT;
+       new->spec.retry = WORK_QUEUE_DEFAULT_RETRY;
 
        return new;
 }
@@ -133,8 +134,17 @@ static int work_queue_schedule(struct work_queue *wq, unsigned int delay)
        if (CHECK_FLAG(wq->flags, WQ_UNPLUGGED) && (wq->thread == NULL)
            && !work_queue_empty(wq)) {
                wq->thread = NULL;
-               thread_add_timer_msec(wq->master, work_queue_run, wq, delay,
-                                     &wq->thread);
+
+               /* Schedule timer if there's a delay, otherwise just schedule
+                * as an 'event'
+                */
+               if (delay > 0)
+                       thread_add_timer_msec(wq->master, work_queue_run, wq,
+                                             delay, &wq->thread);
+               else
+                       thread_add_event(wq->master, work_queue_run, wq, 0,
+                                        &wq->thread);
+
                /* set thread yield time, if needed */
                if (wq->thread && wq->spec.yield != THREAD_YIELD_TIME_SLOT)
                        thread_set_yield_time(wq->thread, wq->spec.yield);
@@ -234,7 +244,7 @@ int work_queue_run(struct thread *thread)
 {
        struct work_queue *wq;
        struct work_queue_item *item, *titem;
-       wq_item_status ret;
+       wq_item_status ret = WQ_SUCCESS;
        unsigned int cycles = 0;
        char yielded = 0;
 
@@ -376,9 +386,14 @@ stats:
 #endif
 
        /* Is the queue done yet? If it is, call the completion callback. */
-       if (!work_queue_empty(wq))
-               work_queue_schedule(wq, 0);
-       else if (wq->spec.completion_func)
+       if (!work_queue_empty(wq)) {
+               if (ret == WQ_RETRY_LATER ||
+                   ret == WQ_QUEUE_BLOCKED)
+                       work_queue_schedule(wq, wq->spec.retry);
+               else
+                       work_queue_schedule(wq, 0);
+
+       } else if (wq->spec.completion_func)
                wq->spec.completion_func(wq);
 
        return 0;

diff --git a/lib/workqueue.h b/lib/workqueue.h
index fe1700f8ded5aaf79eff7b3a232d9e51c1cdac29..7c846550631df99d904f79aca57eb094e3007589 100644 (file)
--- a/lib/workqueue.h
+++ b/lib/workqueue.h
@@ -30,6 +30,9 @@ DECLARE_MTYPE(WORK_QUEUE)
 /* Hold time for the initial schedule of a queue run, in  millisec */
 #define WORK_QUEUE_DEFAULT_HOLD 50
 
+/* Retry for queue that is 'blocked' or 'retry later' */
+#define WORK_QUEUE_DEFAULT_RETRY 0
+
 /* action value, for use by item processor and item error handlers */
 typedef enum {
        WQ_SUCCESS = 0,
@@ -90,6 +93,8 @@ struct work_queue {
 
                unsigned long
                        yield; /* yield time in us for associated thread */
+
+               uint32_t retry; /* Optional retry timeout if queue is blocked */
        } spec;
 
        /* remaining fields should be opaque to users */