syncop: improve scaling and implement more tools

The current scaling of the syncop thread pool is not working properly
and can leave some tasks in the run queue more time than necessary
when the maximum number of threads is not reached.

This patch provides a better scaling condition to react faster to
pending work.

Condition variables and sleep in the context of a synctask have also
been implemented. Their purpose is to replace regular condition
variables and sleeps that block synctask threads and prevent other
tasks to be executed.

The new features have been applied to several places in glusterd.

Change-Id: Ic50b7c73c104f9e41f08101a357d30b95efccfbf
Fixes: #1116
Signed-off-by: Xavi Hernandez <xhernandez@redhat.com>

1 files changed, 44 insertions, 9 deletions

diff --git a/libglusterfs/src/glusterfs/syncop.h b/libglusterfs/src/glusterfs/syncop.h
index bfdec491ba8..1a801a91444 100644
--- a/libglusterfs/src/glusterfs/syncop.h
+++ b/libglusterfs/src/glusterfs/syncop.h
@@ -16,6 +16,8 @@
 #include <ucontext.h>
 #include "glusterfs/dict.h"   // for dict_t
 #include "glusterfs/stack.h"  // for call_frame_t, STACK_DESTROY, STACK_...
+#include "glusterfs/timer.h"
+
 #define SYNCENV_PROC_MAX 16
 #define SYNCENV_PROC_MIN 2
 #define SYNCPROC_IDLE_TIME 600
@@ -32,6 +34,7 @@
 struct synctask;
 struct syncproc;
 struct syncenv;
+struct synccond;
 
 typedef int (*synctask_cbk_t)(int ret, call_frame_t *frame, void *opaque);
 
@@ -55,9 +58,12 @@ struct synctask {
     call_frame_t *opframe;
     synctask_cbk_t synccbk;
     synctask_fn_t syncfn;
-    synctask_state_t state;
+    struct timespec *delta;
+    gf_timer_t *timer;
+    struct synccond *synccond;
     void *opaque;
     void *stack;
+    synctask_state_t state;
     int woken;
     int slept;
     int ret;
@@ -85,19 +91,21 @@ struct syncproc {
 /* hosts the scheduler thread and framework for executing synctasks */
 struct syncenv {
     struct syncproc proc[SYNCENV_PROC_MAX];
-    int procs;
+
+    pthread_mutex_t mutex;
+    pthread_cond_t cond;
 
     struct list_head runq;
-    int runcount;
     struct list_head waitq;
-    int waitcount;
+
+    int procs;
+    int procs_idle;
+
+    int runcount;
 
     int procmin;
     int procmax;
 
-    pthread_mutex_t mutex;
-    pthread_cond_t cond;
-
     size_t stacksize;
 
     int destroy; /* FLAG to mark syncenv is in destroy mode
@@ -123,6 +131,13 @@ struct synclock {
 };
 typedef struct synclock synclock_t;
 
+struct synccond {
+    pthread_mutex_t pmutex;
+    pthread_cond_t pcond;
+    struct list_head waitq;
+};
+typedef struct synccond synccond_t;
+
 struct syncbarrier {
     gf_boolean_t initialized; /*Set on successful initialization*/
     pthread_mutex_t guard;    /* guard the remaining members, pair @cond */
@@ -219,7 +234,7 @@ struct syncopctx {
 #define __yield(args)                                                          \
     do {                                                                       \
         if (args->task) {                                                      \
-            synctask_yield(args->task);                                        \
+            synctask_yield(args->task, NULL);                                  \
         } else {                                                               \
             pthread_mutex_lock(&args->mutex);                                  \
             {                                                                  \
@@ -310,7 +325,9 @@ synctask_join(struct synctask *task);
 void
 synctask_wake(struct synctask *task);
 void
-synctask_yield(struct synctask *task);
+synctask_yield(struct synctask *task, struct timespec *delta);
+void
+synctask_sleep(int32_t secs);
 void
 synctask_waitfor(struct synctask *task, int count);
 
@@ -408,6 +425,24 @@ synclock_trylock(synclock_t *lock);
 int
 synclock_unlock(synclock_t *lock);
 
+int32_t
+synccond_init(synccond_t *cond);
+
+void
+synccond_destroy(synccond_t *cond);
+
+int
+synccond_wait(synccond_t *cond, synclock_t *lock);
+
+int
+synccond_timedwait(synccond_t *cond, synclock_t *lock, struct timespec *delta);
+
+void
+synccond_signal(synccond_t *cond);
+
+void
+synccond_broadcast(synccond_t *cond);
+
 int
 syncbarrier_init(syncbarrier_t *barrier);
 int