Ticket #7606: 0001-Heap-draft-3-rebased.patch

includes/rts/storage/TSO.h

@@ -46 +46 @@
 /* Reason for thread being blocked. See comment above struct StgTso_. */
 typedef union {
   StgClosure *closure;
-  StgTSO *prev; // a back-link when the TSO is on the run queue (NotBlocked)
   struct MessageBlackHole_ *bh;
   struct MessageThrowTo_ *throwto;
   struct MessageWakeup_  *wakeup;
@@ -87 +86 @@
     struct StgTSO_*         _link;
     /*
       Currently used for linking TSOs on:
-      * cap->run_queue_{hd,tl}
       * (non-THREADED_RTS); the blocked_queue
       * and pointing to the next chunk for a ThreadOldStack
 

rts/Capability.c

@@ -230 +230 @@
     cap->idle              = 0;
     cap->disabled          = rtsFalse;
 
-    cap->run_queue_hd      = END_TSO_QUEUE;
-    cap->run_queue_tl      = END_TSO_QUEUE;
+    cap->run_pqueue        = newPQueue(4); // tune this...
+    cap->promoted_run_queue_hd = END_TSO_QUEUE;
 
 #if defined(THREADED_RTS)
     initMutex(&cap->lock);
@@ -1007 +1007 @@
     // or fewer Capabilities as GC threads, but just in case there
     // are more, we mark every Capability whose number is the GC
     // thread's index plus a multiple of the number of GC threads.
-    evac(user, (StgClosure **)(void *)&cap->run_queue_hd);
-    evac(user, (StgClosure **)(void *)&cap->run_queue_tl);
+    nat i;
+    if (cap->promoted_run_queue_hd != END_TSO_QUEUE) {
+        evac(user, (StgClosure **)&cap->promoted_run_queue_hd);
+    }
+    markPQueue(cap->run_pqueue, evac, user);
 #if defined(THREADED_RTS)
     evac(user, (StgClosure **)(void *)&cap->inbox);
 #endif

rts/Capability.h

@@ -24 +24 @@
 #include "sm/GC.h" // for evac_fn
 #include "Task.h"
 #include "Sparks.h"
+#include "PQueue.h"
 
 #include "BeginPrivate.h"
 
@@ -55 +56 @@
     // access to its run queue, so can wake up threads without
     // taking a lock, and the common path through the scheduler is
     // also lock-free.
-    StgTSO *run_queue_hd;
-    StgTSO *run_queue_tl;
+    StgTSO *promoted_run_queue_hd;
+    PQueue *run_pqueue;
 
     // [SSS] Stride scheduling extensions.  The Task with this
     // Capability has exclusive access to this variable.
@@ -162 +163 @@
 // Task is bound, its thread has just blocked, and it may have been
 // moved to another Capability.
 #define ASSERT_PARTIAL_CAPABILITY_INVARIANTS(cap,task)  \
-  ASSERT(cap->run_queue_hd == END_TSO_QUEUE ?           \
-            cap->run_queue_tl == END_TSO_QUEUE : 1);    \
   ASSERT(myTask() == task);                             \
   ASSERT_TASK_ID(task);
 

(a) /dev/null vs. (b) b/rts/PQueue.c

@@ +1 @@
+/*
+ * Priority queue data structure (not concurrent).
+ *
+ * Classic undergraduate implementation of a Heap.
+ */
+
+#include "PosixSource.h"
+#include "Rts.h"
+
+#include "RtsUtils.h"
+#include "PQueue.h"
+#include "sm/GC.h"
+
+PQueue *
+newPQueue (nat size)
+{
+    StgWord realsize;
+    PQueue *q;
+    realsize = roundUp2(size);
+    q = (PQueue*) stgMallocBytes(sizeof(PQueue), "newPQueue");
+    int sz = realsize * sizeof(StgClosurePtr);
+    q->elements = stgMallocBytes(sz, "newPQueue: elements");
+    q->capacity = realsize;
+    q->size = 0;
+    q->deferred = 0;
+    return q;
+}
+
+void
+freePQueue (PQueue *q)
+{
+    stgFree(q->elements);
+    stgFree(q);
+}
+
+void
+iteratePQueue(PQueue *q, void (*fp)(StgTSO *e)) {
+    nat i;
+    // 0 1 2 (size = 3, deferred = 0)
+    //   1 2 (size = 2, deferred = 1)
+    for (i = q->deferred; i < q->size + q->deferred; i++) {
+        fp(q->elements[i]);
+    }
+}
+
+void
+markPQueue(PQueue *q, evac_fn evac, void *user) {
+    nat i;
+    for (i = q->deferred; i < q->size + q->deferred; i++) {
+        evac(user, (StgClosure **)&q->elements[i]);
+    }
+}
+
+void
+truncatePQueue(PQueue *q) {
+    q->size = 0;
+}
+
+void
+growPQueue(PQueue *q) {
+    int esz = q->capacity * sizeof(StgClosurePtr);
+    void ** nelements = stgMallocBytes(2 * esz, "insertPQueue: elements");
+    memcpy(/*dst*/nelements, q->elements, esz);
+    stgFree(q->elements);
+    q->elements = nelements;
+    q->capacity *= 2;
+}

(a) /dev/null vs. (b) b/rts/PQueue.h

@@ +1 @@
+/*
+ * Priority queue data structure (not concurrent)
+ */
+
+#ifndef PQUEUE_H
+#define PQUEUE_H
+
+#include "sm/GC.h"
+
+#define PARENT(i) (((i)-1)/2)
+#define LEFT(i)   (2*(i)+1)
+#define RIGHT(i)  (2*(i)+2)
+
+typedef struct PQueue_ {
+    // size of the heap array
+    StgWord size;
+    // capacity of the heap array
+    StgWord capacity;
+    // the elements array
+    StgTSO **elements;
+    // deferred?
+    int deferred;
+} PQueue;
+
+PQueue * newPQueue(nat size);
+void freePQueue(PQueue *q);
+
+EXTERN_INLINE void upHeap(PQueue*, StgTSO*);
+EXTERN_INLINE void
+upHeap(PQueue *q, StgTSO *last)
+{
+    StgWord64 lastk = last->ss_pass;
+    nat c, i;
+    for (i = 0; LEFT(i) < q->size; i = c) {
+        if (LEFT(i) != q->size && q->elements[LEFT(i)]->ss_pass > q->elements[RIGHT(i)]->ss_pass) {
+            c = RIGHT(i); // right exists and wins
+        } else {
+            c = LEFT(i); // only had left node, or left wins
+        }
+        if (lastk > q->elements[c]->ss_pass) {
+            q->elements[i] = q->elements[c];
+        } else {
+            break;
+        }
+    }
+    q->elements[i] = last;
+}
+
+EXTERN_INLINE void * deleteMinPQueue(PQueue *q);
+EXTERN_INLINE void * peekMinPQueue(PQueue *q);
+EXTERN_INLINE void insertPQueue(PQueue *q, StgTSO *elem);
+void truncatePQueue(PQueue *q);
+void iteratePQueue(PQueue *q, void (*fp)(StgTSO *e));
+void markPQueue(PQueue *q, evac_fn evac, void *user);
+void growPQueue(PQueue *q);
+
+EXTERN_INLINE void
+insertPQueue(PQueue *q, StgTSO *elem)
+{
+    StgWord64 key = elem->ss_pass;
+    if (q->size == q->capacity) { growPQueue(q); }
+    nat i;
+    for (i = q->size; i > 0 && q->elements[PARENT(i)]->ss_pass > key; i = PARENT(i)) {
+        q->elements[i] = q->elements[PARENT(i)];
+    }
+    q->elements[i] = elem;
+    q->size++;
+}
+
+EXTERN_INLINE void *
+peekMinPQueue(PQueue *q) {
+    return q->elements[0];
+}
+
+EXTERN_INLINE void *
+deleteMinPQueue(PQueue *q) {
+    if (q->size == 0) {
+        return NULL;
+    }
+    StgTSO *min = q->elements[0];
+    q->size--;
+    // XXX implementing properly amortized downsizing
+    if (q->size == 0) {
+        q->elements[0] = NULL; // make peekMinPQueue work
+    } else {
+        upHeap(q, q->elements[q->size]);
+    }
+    return min;
+}
+
+#undef PARENT
+#undef LEFT
+#undef RIGHT
+
+#endif // PQUEUE_H

rts/RtsUtils.c

@@ -51 +51 @@
 #include <windows.h>
 #endif
 
+StgWord
+roundUp2(StgWord val)
+{
+    StgWord rounded = 1;
+    
+    /* StgWord is unsigned anyway, only catch 0 */
+    if (val == 0) {
+        barf("roundUp2: invalid size 0 requested");
+    }
+    /* at least 1 bit set, shift up to its place */
+    do {
+        rounded = rounded << 1;
+    } while (0 != (val = val>>1));
+    return rounded;
+}
+
 /* -----------------------------------------------------------------------------
    Result-checking malloc wrappers.
    -------------------------------------------------------------------------- */

rts/RtsUtils.h

@@ -11 +11 @@
 
 #include "BeginPrivate.h"
 
+StgWord roundUp2(StgWord);
+
 /* -----------------------------------------------------------------------------
  * (Checked) dynamic allocation
  * -------------------------------------------------------------------------- */

rts/Schedule.c

@@ -12 +12 @@
 
 #include "sm/Storage.h"
 #include "RtsUtils.h"
+#include "PQueue.h"
 #include "StgRun.h"
 #include "Schedule.h"
 #include "Interpreter.h"
@@ -560 +561 @@
  * -------------------------------------------------------------------------- */
 
 void
-removeFromRunQueue (Capability *cap, StgTSO *tso)
+promoteInRunQueue (Capability *cap STG_UNUSED, StgTSO *tso STG_UNUSED)
 {
-    if (tso->block_info.prev == END_TSO_QUEUE) {
-        ASSERT(cap->run_queue_hd == tso);
-        cap->run_queue_hd = tso->_link;
-    } else {
-        setTSOLink(cap, tso->block_info.prev, tso->_link);
-    }
-    if (tso->_link == END_TSO_QUEUE) {
-        ASSERT(cap->run_queue_tl == tso);
-        cap->run_queue_tl = tso->block_info.prev;
-    } else {
-        setTSOPrev(cap, tso->_link, tso->block_info.prev);
-    }
-    tso->_link = tso->block_info.prev = END_TSO_QUEUE;
-
-    IF_DEBUG(sanity, checkRunQueue(cap));
-}
-
-void
-promoteInRunQueue (Capability *cap, StgTSO *tso)
-{
-    removeFromRunQueue(cap, tso);
-    pushOnRunQueue(cap, tso);
+    // no-op, at the moment; if removeFromRunQueue was implemented it'd be
+    // removeFromRunQueue(cap, tso);
+    // pushOnRunQueue(cap, tso);
 }
 
 /* ----------------------------------------------------------------------------
@@ -700 +682 @@
 #if defined(THREADED_RTS)
 
     Capability *free_caps[n_capabilities], *cap0;
-    nat i, n_free_caps;
+    nat i, j, n_free_caps;
 
     // migration can be turned off with +RTS -qm
     if (!RtsFlags.ParFlags.migrate) return;
@@ -742 +724 @@
     //   - giving low priority to moving long-lived threads
 
     if (n_free_caps > 0) {
-        StgTSO *prev, *t, *next;
+        StgTSO *t;
 #ifdef SPARK_PUSHING
         rtsBool pushed_to_all;
 #endif
@@ -759 +741 @@
         pushed_to_all = rtsFalse;
 #endif
 
-        if (cap->run_queue_hd != END_TSO_QUEUE) {
-            prev = cap->run_queue_hd;
-            t = prev->_link;
-            prev->_link = END_TSO_QUEUE;
-            for (; t != END_TSO_QUEUE; t = next) {
-                next = t->_link;
-                t->_link = END_TSO_QUEUE;
+        // go through all of the TSOs in the run queue and decide where
+        // they should go
+        // XXX We can create the new heap more efficiently O(n) by just
+        // blitting them in and then re-heapifying
+        if (!emptyRunQueue(cap)) {
+            StgWord64 k;
+            PQueue *oq = cap->run_pqueue;
+            cap->run_pqueue = newPQueue(oq->size);
+            insertPQueue(cap->run_pqueue, oq->elements[0]);
+            for (j = 1; j < oq->size; j++) {
+                t = oq->elements[j];
                 if (t->bound == task->incall // don't move my bound thread
-                    || tsoLocked(t)) {  // don't move a locked thread
-                    setTSOLink(cap, prev, t);
-                    setTSOPrev(cap, t, prev);
-                    prev = t;
-                } else if (i == n_free_caps) {
+                    || tsoLocked(t)) {  // don't move a locked thread
+                    insertPQueue(cap->run_pqueue, t);
+                } else if (i == n_free_caps) {
 #ifdef SPARK_PUSHING
-                    pushed_to_all = rtsTrue;
+                    pushed_to_all = rtsTrue;
 #endif
-                    i = 0;
-                    // keep one for us
-                    setTSOLink(cap, prev, t);
-                    setTSOPrev(cap, t, prev);
-                    prev = t;
-                } else {
-                    leaveRunQueue(cap,t);
-                    joinRunQueue(free_caps[i],t);
+                    i = 0;
+                    insertPQueue(cap->run_pqueue, t); // keep one for us
+                } else {
+                    leaveRunQueue(cap, t);
+                    joinRunQueue(free_caps[i], t);
 
                     traceEventMigrateThread (cap, t, free_caps[i]->no);
 
-                    if (t->bound) { t->bound->task->cap = free_caps[i]; }
-                    t->cap = free_caps[i];
-                    i++;
-                }
-            }
-            cap->run_queue_tl = prev;
-
-            IF_DEBUG(sanity, checkRunQueue(cap));
-        }
+                    if (t->bound) { t->bound->task->cap = free_caps[i]; }
+                    t->cap = free_caps[i];
+                    i++;
+                }
+            }
+            freePQueue(oq);
+        }
 
 #ifdef SPARK_PUSHING
         /* JB I left this code in place, it would work but is not necessary */

rts/Schedule.h

@@ -128 +128 @@
 
 #if !IN_STG_CODE
 
+EXTERN_INLINE void
+annulTSO(StgTSO *tso);
+
+EXTERN_INLINE void
+annulTSO(StgTSO *tso) {
+    // hack to make some invariants with regards to block_info and _link work
+    // this is called whereever we would have stepped all over the
+    // fields in the linked list implementation
+    ASSERT(tso->_link == END_TSO_QUEUE);
+    tso->block_info.closure = (StgClosure*)END_TSO_QUEUE;
+}
+
 /* END_TSO_QUEUE and friends now defined in includes/StgMiscClosures.h */
 
 /* Add a thread to the end of the run queue.
@@ -140 +152 @@
 EXTERN_INLINE void
 appendToRunQueue (Capability *cap, StgTSO *tso)
 {
-    ASSERT(tso->_link == END_TSO_QUEUE);
-    if (cap->run_queue_hd == END_TSO_QUEUE) {
-        cap->run_queue_hd = tso;
-        tso->block_info.prev = END_TSO_QUEUE;
-    } else {
-        setTSOLink(cap, cap->run_queue_tl, tso);
-        setTSOPrev(cap, tso, cap->run_queue_tl);
-    }
-    cap->run_queue_tl = tso;
+    tso->ss_pass = cap->ss_pass++;                                          // [EMU]
+    //tso->ss_pass += STRIDE1 / tso->ss_tickets;                            // [EMU]
+    annulTSO(tso);
+    insertPQueue(cap->run_pqueue, tso);
 }
 
 EXTERN_INLINE void
@@ -157 +164 @@
 EXTERN_INLINE void
 joinRunQueue(Capability *cap, StgTSO *tso)
 {
+    tso->ss_pass = cap->ss_pass++;
     appendToRunQueue(cap, tso);
 }
 
@@ -169 +177 @@
 EXTERN_INLINE void
 pushOnRunQueue (Capability *cap, StgTSO *tso)
 {
-    setTSOLink(cap, tso, cap->run_queue_hd);
-    tso->block_info.prev = END_TSO_QUEUE;
-    if (cap->run_queue_hd != END_TSO_QUEUE) {
-        setTSOPrev(cap, cap->run_queue_hd, tso);
-    }
-    cap->run_queue_hd = tso;
-    if (cap->run_queue_tl == END_TSO_QUEUE) {
-        cap->run_queue_tl = tso;
+    annulTSO(tso);                                                          // [EMU]
+    //tso->ss_pass += STRIDE1 / tso->ss_tickets;                            // [EMU]
+    if (cap->promoted_run_queue_hd == END_TSO_QUEUE) {
+        // annulTSO invariant used here
+    } else {
+        setTSOLink(cap, tso, cap->promoted_run_queue_hd);
     }
+    cap->promoted_run_queue_hd = tso;
 }
 
 EXTERN_INLINE void
@@ -186 +193 @@
 EXTERN_INLINE void
 fastJoinRunQueue(Capability *cap, StgTSO *tso)
 {
+    tso->ss_pass = cap->ss_pass++;                                          // [EMU]
     pushOnRunQueue(cap, tso);
 }
 
@@ -193 +201 @@
  */
 INLINE_HEADER StgTSO *
 popRunQueue (Capability *cap)
-{ 
-    StgTSO *t = cap->run_queue_hd;
-    ASSERT(t != END_TSO_QUEUE);
-    cap->run_queue_hd = t->_link;
-    if (t->_link != END_TSO_QUEUE) {
-        t->_link->block_info.prev = END_TSO_QUEUE;
-    }
-    t->_link = END_TSO_QUEUE; // no write barrier req'd
-    if (cap->run_queue_hd == END_TSO_QUEUE) {
-        cap->run_queue_tl = END_TSO_QUEUE;
+{
+    if (cap->promoted_run_queue_hd == END_TSO_QUEUE) {
+        StgTSO *t = deleteMinPQueue(cap->run_pqueue);
+        StgTSO *next = peekMinPQueue(cap->run_pqueue);
+        /* if (next != NULL) {
+            if (cap->ss_pass < next->ss_pass) {
+                cap->ss_pass = next->ss_pass;
+            }
+        } */                                                                // [EMU]
+        return t;
+    } else {
+        StgTSO *t = cap->promoted_run_queue_hd;
+        cap->promoted_run_queue_hd = t->_link;
+        t->_link = END_TSO_QUEUE; // no write barrier req'd
+        return t;
     }
-    return t;
 }
 
 INLINE_HEADER StgTSO *
 peekRunQueue (Capability *cap)
 {
-    return cap->run_queue_hd;
+    if (cap->promoted_run_queue_hd == END_TSO_QUEUE) {
+        return peekMinPQueue(cap->run_pqueue);
+    } else {
+        return cap->promoted_run_queue_hd;
+    }
 }
 
 EXTERN_INLINE void
-leaveRunQueue (Capability *cap, StgTSO *tso);
+leaveRunQueue(Capability *cap, StgTSO *tso);
 
 EXTERN_INLINE void
-leaveRunQueue (Capability *cap, StgTSO *tso)
+leaveRunQueue(Capability *cap, StgTSO *tso)
 {
     // XXX implement me
 }
 
-void removeFromRunQueue (Capability *cap, StgTSO *tso);
 extern void promoteInRunQueue (Capability *cap, StgTSO *tso);
 
 /* Add a thread to the end of the blocked queue.
@@ -252 +267 @@
 INLINE_HEADER rtsBool
 emptyRunQueue(Capability *cap)
 {
-    return emptyQueue(cap->run_queue_hd);
+    return cap->promoted_run_queue_hd == END_TSO_QUEUE && peekMinPQueue(cap->run_pqueue) == NULL;
 }
 
 /* assumes that the queue is not empty; so combine this with
@@ -260 +275 @@
 INLINE_HEADER rtsBool
 singletonRunQueue(Capability *cap)
 {
-    ASSERT(!emptyRunQueue(cap));
-    return cap->run_queue_hd->_link == END_TSO_QUEUE;
+    if (cap->promoted_run_queue_hd != END_TSO_QUEUE) {
+        if (cap->run_pqueue->size != 0) return 0;
+        return cap->promoted_run_queue_hd->_link == END_TSO_QUEUE;
+    } else {
+        return cap->run_pqueue->size == 1;
+    }
 }
 
 INLINE_HEADER void
 truncateRunQueue(Capability *cap)
 {
-    cap->run_queue_hd = END_TSO_QUEUE;
-    cap->run_queue_tl = END_TSO_QUEUE;
+    cap->promoted_run_queue_hd = END_TSO_QUEUE;
+    truncatePQueue(cap->run_pqueue);
 }
 
 #if !defined(THREADED_RTS)

rts/Threads.c

@@ -113 +113 @@
     tso->trec = NO_TREC;
 
     tso->ss_tickets = DEFAULT_TICKETS;
+    tso->ss_pass = cap->ss_pass;
 
 #ifdef PROFILING
     tso->prof.cccs = CCS_MAIN;
@@ -844 +845 @@
   for (i = 0; i < n_capabilities; i++) {
       cap = &capabilities[i];
       debugBelch("threads on capability %d:\n", cap->no);
-      for (t = cap->run_queue_hd; t != END_TSO_QUEUE; t = t->_link) {
-          printThreadStatus(t);
-      }
+      // XXX do the promoted queue
+      iteratePQueue(cap->run_pqueue, printThreadStatus);
   }
 
   debugBelch("other threads:\n");

rts/WSDeque.c

@@ -50 +50 @@
  * newWSDeque
  * -------------------------------------------------------------------------- */
 
-/* internal helpers ... */
-
-static StgWord
-roundUp2(StgWord val)
-{
-    StgWord rounded = 1;
-    
-    /* StgWord is unsigned anyway, only catch 0 */
-    if (val == 0) {
-        barf("DeQue,roundUp2: invalid size 0 requested");
-    }
-    /* at least 1 bit set, shift up to its place */
-    do {
-        rounded = rounded << 1;
-    } while (0 != (val = val>>1));
-    return rounded;
-}
-
 WSDeque *
 newWSDeque (nat size)
 {

rts/sm/Sanity.c

@@ -789 +789 @@
 void
 checkRunQueue(Capability *cap)
 {
-    StgTSO *prev, *tso;
-    prev = END_TSO_QUEUE;
-    for (tso = cap->run_queue_hd; tso != END_TSO_QUEUE; 
-         prev = tso, tso = tso->_link) {
-        ASSERT(prev == END_TSO_QUEUE || prev->_link == tso);
-        ASSERT(tso->block_info.prev == prev);
-    }
-    ASSERT(cap->run_queue_tl == prev);
+    // [SSS] no-op
 }
 
 /* -----------------------------------------------------------------------------

rts/sm/Storage.c

@@ -878 +878 @@
 }
 
 void
-setTSOPrev (Capability *cap, StgTSO *tso, StgTSO *target)
-{
-    if (tso->dirty == 0) {
-        tso->dirty = 1;
-        recordClosureMutated(cap,(StgClosure*)tso);
-    }
-    tso->block_info.prev = target;
-}
-
-void
 dirty_TSO (Capability *cap, StgTSO *tso)
 {
     if (tso->dirty == 0) {