Patches to improve State Threads timer performance

State Threads currently uses a sorted doubly linked list to keep track
of threads waiting for a timeout.  This works fine in applications
that have only a small number of threads waiting for timeouts at any
one time, but performance quickly degrades when there are hundreds or
even thousands of waiting threads.  This has turned out to be a
significant performance bottleneck in our applications.

The attached patches remedy this performance problem by replacing the
doubly linked list with a linked heap data structure, which reduces
the asymptotic time complexity of inserting a new waiting thread from
O(N) to O(log n).  We have successfully been using a version of State
Threads with these modifications for several months now.

Please consider integrating these patches into the release.
-- 
Andreas Gustafsson, [email protected]

diff -u st-1.4.orig/common.h st-1.4/common.h
--- st-1.4.orig/common.h	2002-02-22 12:55:46.000000000 -0800
+++ st-1.4/common.h	2003-10-15 10:26:36.000000000 -0700
@@ -151,7 +151,9 @@
 } _st_cond_t;
 
 
-typedef struct _st_thread {
+typedef struct _st_thread _st_thread_t;
+
+struct _st_thread {
   int state;                  /* Thread's state */
   int flags;                  /* Thread's flags */
 
@@ -166,14 +168,18 @@
 #ifdef DEBUG
   _st_clist_t tlink;          /* For putting on thread queue */
 #endif
-  st_utime_t sleep;           /* Sleep time when thread is sleeping */
+
+  st_utime_t due;             /* Wakeup time when thread is sleeping */
+  _st_thread_t *left;         /* For putting in timeout heap */
+  _st_thread_t *right;
+  int heap_index;
 
   void **private_data;        /* Per thread private data */
 
   _st_cond_t *term;           /* Termination condition variable for join */
 
   jmp_buf context;            /* Thread's context */
-} _st_thread_t;
+};
 
 
 typedef struct _st_mutex {
@@ -197,14 +203,15 @@
 
   _st_clist_t run_q;          /* run queue for this vp */
   _st_clist_t io_q;           /* io queue for this vp */
-  _st_clist_t sleep_q;        /* sleep queue for this vp */
   _st_clist_t zombie_q;       /* zombie queue for this vp */
 #ifdef DEBUG
   _st_clist_t thread_q;       /* all threads of this vp */
 #endif
-  st_utime_t sleep_max;
   int pagesize;
 
+  _st_thread_t *sleep_q;      /* sleep queue for this vp */
+  int sleepq_size;	      /* number of threads on sleep queue */
+
 #ifndef USE_POLL
   int maxfd;
   fd_set fd_read_set, fd_write_set, fd_exception_set;
@@ -237,7 +244,6 @@
 #define _ST_CURRENT_THREAD()            (_st_this_thread)
 #define _ST_SET_CURRENT_THREAD(_thread) (_st_this_thread = (_thread))
 
-#define _ST_SLEEPQMAX                   (_st_this_vp.sleep_max)
 #define _ST_PAGE_SIZE                   (_st_this_vp.pagesize)
 
 #define _ST_FD_READ_SET                 (_st_this_vp.fd_read_set)
@@ -258,6 +264,7 @@
 #define _ST_IOQ                         (_st_this_vp.io_q)
 #define _ST_RUNQ                        (_st_this_vp.run_q)
 #define _ST_SLEEPQ                      (_st_this_vp.sleep_q)
+#define _ST_SLEEPQ_SIZE                 (_st_this_vp.sleepq_size)
 #define _ST_ZOMBIEQ                     (_st_this_vp.zombie_q)
 
 #ifdef DEBUG
diff -u st-1.4.orig/sched.c st-1.4/sched.c
--- st-1.4.orig/sched.c	2002-01-30 19:46:11.000000000 -0800
+++ st-1.4/sched.c	2003-10-15 10:24:13.000000000 -0700
@@ -183,12 +183,14 @@
 
   ST_INIT_CLIST(&_ST_RUNQ);
   ST_INIT_CLIST(&_ST_IOQ);
-  ST_INIT_CLIST(&_ST_SLEEPQ);
   ST_INIT_CLIST(&_ST_ZOMBIEQ);
 #ifdef DEBUG
   ST_INIT_CLIST(&_ST_THREADQ);
 #endif
 
+  _ST_SLEEPQ = NULL;
+  _ST_SLEEPQ_SIZE = 0;
+
 #ifndef USE_POLL
   _st_this_vp.maxfd = -1;
 #else
@@ -286,10 +288,13 @@
   wp = &w;
   ep = &e;
 
-  if (ST_CLIST_IS_EMPTY(&_ST_SLEEPQ)) {
+  if (_ST_SLEEPQ == NULL) {
     tvp = NULL;
   } else {
-    min_timeout = (_ST_THREAD_PTR(_ST_SLEEPQ.next))->sleep;
+    if (_ST_SLEEPQ->due <= _st_this_vp.last_clock)
+      min_timeout = 0;
+    else
+      min_timeout = _ST_SLEEPQ->due - _st_this_vp.last_clock;
     timeout.tv_sec  = (int) (min_timeout / 1000000);
     timeout.tv_usec = (int) (min_timeout % 1000000);
     tvp = &timeout;
@@ -613,43 +618,127 @@
 }
 
 
-void _st_add_sleep_q(_st_thread_t *thread, st_utime_t timeout)
-{
-  st_utime_t sleep;
-  _st_clist_t *q;
-  _st_thread_t *t;
+/*
+ * Insert "thread" into the timeout heap, in the position
+ * specified by thread->heap_index.
+ */
+static _st_thread_t **heap_insert(_st_thread_t *thread) {
+  int target = thread->heap_index;
+  int s = target;
+  _st_thread_t **p = &_ST_SLEEPQ;
+  int bits = 0;
+  int bit;
+  int index = 1;
+
+  while (s) {
+    s >>= 1;
+    bits++;
+  }
+  for (bit = bits - 2; bit >= 0; bit--) {
+    if (thread->due < (*p)->due) {
+      _st_thread_t *t = *p;
+      thread->left = t->left;
+      thread->right = t->right;
+      *p = thread;
+      thread->heap_index = index;
+      thread = t;
+    }
+    index <<= 1;
+    if (target & (1 << bit)) {
+      p = &((*p)->right);
+      index |= 1;
+    } else {
+      p = &((*p)->left);
+    }
+  }
+  thread->heap_index = index;
+  *p = thread;
+  thread->left = thread->right = NULL;
+  return p;
+}
 
-  /* Check if we are longest timeout */
-  if (timeout >= _ST_SLEEPQMAX) {
-    ST_APPEND_LINK(&thread->links, &_ST_SLEEPQ);
-    thread->sleep = timeout - _ST_SLEEPQMAX;
-    _ST_SLEEPQMAX = timeout;
-  } else {
-    /* Sort thread into global sleep queue at appropriate point */
-    sleep = _ST_SLEEPQMAX;
-    q = _ST_SLEEPQ.prev;
-
-    /* Now scan the list backward for where to insert this entry */
-    while (q != &_ST_SLEEPQ) {
-      t = _ST_THREAD_PTR(q);
-      sleep -= t->sleep;
-      if (timeout >= sleep) {
-	/* Found sleeper to insert in front of */
+/*
+ * Delete "thread" from the timeout heap.
+ */
+static void heap_delete(_st_thread_t *thread) {
+  _st_thread_t *t, **p;
+  int bits = 0;
+  int s, bit;
+
+  /* First find and unlink the last heap element */
+  p = &_ST_SLEEPQ;
+  s = _ST_SLEEPQ_SIZE;
+  while (s) {
+    s >>= 1;
+    bits++;
+  }
+  for (bit = bits - 2; bit >= 0; bit--) {
+    if (_ST_SLEEPQ_SIZE & (1 << bit)) {
+      p = &((*p)->right);
+    } else {
+      p = &((*p)->left);
+    }
+  }
+  t = *p;
+  *p = NULL;
+  --_ST_SLEEPQ_SIZE;
+  if (t != thread) {
+    /*
+     * Insert the unlinked last element in place of the element we are deleting
+     */
+    t->heap_index = thread->heap_index;
+    p = heap_insert(t);
+    t = *p;
+    t->left = thread->left;
+    t->right = thread->right;
+
+    /*
+     * Reestablish the heap invariant.
+     */
+    for (;;) {
+      _st_thread_t *y; /* The younger child */
+      int index_tmp;
+      if (t->left == NULL)
+	break;
+      else if (t->right == NULL)
+	y = t->left;
+      else if (t->left->due < t->right->due)
+	y = t->left;
+      else
+	y = t->right;
+      if (t->due > y->due) {
+	_st_thread_t *tl = y->left;
+	_st_thread_t *tr = y->right;
+	*p = y;
+	if (y == t->left) {
+	  y->left = t;
+	  y->right = t->right;
+	  p = &y->left;
+	} else {
+	  y->left = t->left;
+	  y->right = t;
+	  p = &y->right;
+	}
+	t->left = tl;
+	t->right = tr;
+	index_tmp = t->heap_index;
+	t->heap_index = y->heap_index;
+	y->heap_index = index_tmp;
+      } else {
 	break;
       }
-      q = q->prev;
     }
-    thread->sleep = timeout - sleep;
-    ST_INSERT_BEFORE(&thread->links, q);
-
-    /* Subtract our sleep time from the sleeper that follows us */
-    ST_ASSERT(thread->links.next != &_ST_SLEEPQ);
-    t = _ST_THREAD_PTR(thread->links.next);
-    ST_ASSERT(_ST_THREAD_PTR(t->links.prev) == thread);
-    t->sleep -= thread->sleep;
   }
+  thread->left = thread->right = NULL;
+}
+
 
+void _st_add_sleep_q(_st_thread_t *thread, st_utime_t timeout)
+{
+  thread->due = _st_this_vp.last_clock + timeout;
   thread->flags |= _ST_FL_ON_SLEEPQ;
+  thread->heap_index = ++_ST_SLEEPQ_SIZE;
+  heap_insert(thread);
 }
 
 
@@ -658,31 +747,8 @@
  */
 void _st_del_sleep_q(_st_thread_t *thread, int expired)
 {
-  _st_clist_t *q;
-  _st_thread_t *t;
-
-  /* Remove from sleep queue */
-  ST_ASSERT(thread->flags & _ST_FL_ON_SLEEPQ);
-  q = thread->links.next;
-  if (q != &_ST_SLEEPQ) {
-    if (expired) {
-      _ST_SLEEPQMAX -= thread->sleep;
-    } else {
-      t = _ST_THREAD_PTR(q);
-      t->sleep += thread->sleep;
-    }
-  } else {
-    /*
-     * Check if prev is the beginning of the list; if so,
-     * we are the only element on the list.  
-     */
-    if (thread->links.prev != &_ST_SLEEPQ)
-      _ST_SLEEPQMAX -= thread->sleep;
-    else
-      _ST_SLEEPQMAX = 0;
-  }
+  heap_delete(thread);
   thread->flags &= ~_ST_FL_ON_SLEEPQ;
-  ST_REMOVE_LINK(&thread->links);
 }
 
 
@@ -700,18 +766,12 @@
     _st_last_tset = now;
   }
 
-  while (_ST_SLEEPQ.next != &_ST_SLEEPQ) {
-    thread = _ST_THREAD_PTR(_ST_SLEEPQ.next);
+  while (_ST_SLEEPQ != NULL) {
+    thread = _ST_SLEEPQ;
     ST_ASSERT(thread->flags & _ST_FL_ON_SLEEPQ);
-
-    if (elapsed < thread->sleep) {
-      thread->sleep -= elapsed;
-      _ST_SLEEPQMAX -= elapsed;
+    if (thread->due > now)
       break;
-    }
-
     _ST_DEL_SLEEPQ(thread, 1);
-    elapsed -= thread->sleep;
 
     /* If thread is waiting on condition variable, set the time out flag */
     if (thread->state == _ST_ST_COND_WAIT)
@@ -814,6 +874,9 @@
   thread->stack = stack;
   thread->start = start;
   thread->arg = arg;
+  thread->left = 0;
+  thread->right = 0;
+  thread->heap_index = -1;
 
 #ifndef __ia64__
   _ST_INIT_CONTEXT(thread, stack->sp, _st_thread_main);

--- /dev/null	2003-07-14 13:04:09.000000000 -0700
+++ docs/timeout_heap.txt	2003-05-15 13:28:43.000000000 -0700
@@ -0,0 +1,61 @@
+
+How the timeout heap works
+
+This version of ST represents the queue of sleeping threads using a
+heap data structure rather than a sorted linked list.  This improves
+performance when there is a large number of sleeping threads, since
+insertion into a heap takes O(log N) time while insertion into a
+sorted list takes O(N) time.  For example, in one test 1000 threads
+were created, each thread called st_usleep() with a random time
+interval, and then all the threads where immediately interrupted and
+joined before the sleeps had a chance to finish.  The whole process
+was repeated 1000 times, for a total of a million sleep queue
+insertions and removals.  With the old list-based sleep queue, this
+test took 100 seconds; now it takes only 12 seconds.
+
+Heap data structures are typically based on dynamically resized
+arrays.  However, since the existing ST code base was very nicely
+structured around linking the thread objects into pointer-based lists
+without the need for any auxiliary data structures, implementing the
+heap using a similar nodes-and-pointers based approach seemed more
+appropriate for ST than introducing a separate array.
+
+Thus, the new ST timeout heap works by organizing the existing
+_st_thread_t objects in a balanced binary tree, just as they were
+previously organized into a doubly-linked, sorted list.  The global
+_ST_SLEEPQ variable, formerly a linked list head, is now simply a
+pointer to the root of this tree, and the root node of the tree is the
+thread with the earliest timeout.  Each thread object has two child
+pointers, "left" and "right", pointing to threads with later timeouts.
+
+Each node in the tree is numbered with an integer index, corresponding
+to the array index in an array-based heap, and the tree is kept fully
+balanced and left-adjusted at all times.  In other words, the tree
+consists of any number of fully populated top levels, followed by a
+single bottom level which may be partially populated, such that any
+existing nodes form a contiguous block to the left and the spaces for
+missing nodes form a contiguous block to the right. For example, if
+there are nine threads waiting for a timeout, they are numbered and
+arranged in a tree exactly as follows:
+
+              1
+           /     \
+          2       3
+         / \     / \
+        4   5   6   7
+       / \
+      8   9
+
+Each node has either no children, only a left child, or both a left
+and a right child.  Children always time out later than their parents
+(this is called the "heap invariant"), but when a node has two
+children, their mutual order is unspecified - the left child may time
+out before or after the right child.  If a node is numbered N, its
+left child is numbered 2N, and its right child is numbered 2N+1.
+
+There is no pointer from a child to its parent; all pointers point
+downward.  Additions and deletions both work by starting at the root
+and traversing the tree towards the leaves, going left or right
+according to the binary digits forming the index of the destination
+node.  As nodes are added or deleted, existing nodes are rearranged to
+maintain the heap invariant.


-------------------------------------------------------
This SF.net email is sponsored by: SF.net Giveback Program.
SourceForge.net hosts over 70,000 Open Source Projects.
See the people who have HELPED US provide better services:
Click here: http://sourceforge.net/supporters.php