Ticket #7606: common.patch

compiler/prelude/primops.txt.pp

@@ -1839 +1839 @@
    out_of_line = True
    has_side_effects = True
 
+primop  SetTicketsOp "setTickets#" GenPrimOp
+  ThreadId# -> Int# -> State# RealWorld -> State# RealWorld
+  with
+  out_of_line = True
+  has_side_effects = True
+
+primop  GetTicketsOp "getTickets#" GenPrimOp
+  ThreadId# -> State# RealWorld -> (# State# RealWorld, Int# #)
+  with
+  out_of_line = True
+  has_side_effects = True
+
+primop  ModifyTicketsOp "modifyTickets#" GenPrimOp
+  ThreadId# -> Int# -> Int# -> Int# -> State# RealWorld -> (# State# RealWorld, Int# #)
+  with
+  out_of_line = True
+  has_side_effects = True
+
 ------------------------------------------------------------------------
 section "Weak pointers"
 ------------------------------------------------------------------------

includes/rts/storage/TSO.h

@@ -168 +168 @@
      */
     StgWord32  tot_stack_size;
 
+    // These are bounded above by STRIDE1, which is less than a max 32-bit word.
+    StgWord32 ss_tickets;
+    // 64-bit to prevent overflows; only ever accessed by the task which owns TSO.
+    StgWord64 ss_pass;
+
 } *StgTSOPtr;
 
 typedef struct StgStack_ {

includes/stg/MiscClosures.h

@@ -402 +402 @@
 RTS_FUN_DECL(stg_unmaskAsyncExceptionszh);
 RTS_FUN_DECL(stg_myThreadIdzh);
 RTS_FUN_DECL(stg_labelThreadzh);
+RTS_FUN_DECL(stg_getTicketszh);
+RTS_FUN_DECL(stg_setTicketszh);
+RTS_FUN_DECL(stg_modifyTicketszh);
 RTS_FUN_DECL(stg_isCurrentThreadBoundzh);
 RTS_FUN_DECL(stg_threadStatuszh);
 

rts/Capability.h

@@ -58 +58 @@
     StgTSO *run_queue_hd;
     StgTSO *run_queue_tl;
 
+    // [SSS] Stride scheduling extensions.  The Task with this
+    // Capability has exclusive access to this variable.
+    StgWord64 ss_pass;
+
     // Tasks currently making safe foreign calls.  Doubly-linked.
     // When returning, a task first acquires the Capability before
     // removing itself from this list, so that the GC can find all

rts/Linker.c

@@ -1130 +1130 @@
       SymI_HasProto(stg_mkApUpd0zh)                                     \
       SymI_HasProto(stg_myThreadIdzh)                                   \
       SymI_HasProto(stg_labelThreadzh)                                  \
+      SymI_HasProto(stg_getTicketszh)                                   \
+      SymI_HasProto(stg_setTicketszh)                                   \
+      SymI_HasProto(stg_modifyTicketszh)                                \
       SymI_HasProto(stg_newArrayzh)                                     \
       SymI_HasProto(stg_newArrayArrayzh)                                \
       SymI_HasProto(stg_newBCOzh)                                       \

rts/PrimOps.cmm

@@ -628 +628 @@
   return ();
 }
 
+stg_setTicketszh ( gcptr threadid, W_ n )
+{
+  ccall setTickets(threadid "ptr", n);
+  return ();
+}
+
+stg_getTicketszh ( gcptr threadid )
+{
+  W_ r;
+  (r) = ccall getTickets(threadid "ptr");
+  return (r);
+}
+
+stg_modifyTicketszh ( gcptr threadid, W_ n, W_ d, W_ x )
+{
+  W_ r;
+  (r) = ccall modifyTickets(threadid "ptr", n, d, x);
+  return (r);
+}
+
 stg_isCurrentThreadBoundzh (/* no args */)
 {
   W_ r;

rts/RaiseAsync.c

@@ -683 +683 @@
 
  done:
   tso->why_blocked = NotBlocked;
-  appendToRunQueue(cap, tso);
+  joinRunQueue(cap, tso);
 }
 
 /* -----------------------------------------------------------------------------
@@ -1045 +1045 @@
     // wake it up
     if (tso->why_blocked != NotBlocked) {
         tso->why_blocked = NotBlocked;
-        appendToRunQueue(cap,tso);
+        joinRunQueue(cap,tso);
     }        
 
     return tso;

rts/Schedule.c

@@ -148 +148 @@
 static rtsBool scheduleHandleHeapOverflow( Capability *cap, StgTSO *t );
 static rtsBool scheduleHandleYield( Capability *cap, StgTSO *t,
                                     nat prev_what_next );
-static void scheduleHandleThreadBlocked( StgTSO *t );
+static void scheduleHandleThreadBlocked( Capability *cap, StgTSO *t );
 static rtsBool scheduleHandleThreadFinished( Capability *cap, Task *task,
                                              StgTSO *t );
 static rtsBool scheduleNeedHeapProfile(rtsBool ready_to_gc);
@@ -537 +537 @@
         break;
 
     case ThreadBlocked:
-        scheduleHandleThreadBlocked(t);
+        scheduleHandleThreadBlocked(cap, t);
         break;
 
     case ThreadFinished:
@@ -781 +781 @@
                     setTSOPrev(cap, t, prev);
                     prev = t;
                 } else {
-                    appendToRunQueue(free_caps[i],t);
+                    leaveRunQueue(cap,t);
+                    joinRunQueue(free_caps[i],t);
 
                     traceEventMigrateThread (cap, t, free_caps[i]->no);
 
@@ -1217 +1218 @@
  * -------------------------------------------------------------------------- */
 
 static void
-scheduleHandleThreadBlocked( StgTSO *t
-#if !defined(DEBUG)
-    STG_UNUSED
-#endif
-    )
+scheduleHandleThreadBlocked( Capability *cap, StgTSO *t )
 {
 
       // We don't need to do anything.  The thread is blocked, and it
@@ -1234 +1231 @@
     //      threadPaused() might have raised a blocked throwTo
     //      exception, see maybePerformBlockedException().
 
+    leaveRunQueue(cap, t);
+
 #ifdef DEBUG
     traceThreadStatus(DEBUG_sched, t);
 #endif
@@ -1257 +1256 @@
     // blocked mode (see #2910).
     awakenBlockedExceptionQueue (cap, t);
 
+    leaveRunQueue(cap, t);
+
       //
       // Check whether the thread that just completed was a bound
       // thread, and if so return with the result.  
@@ -1277 +1278 @@
               // queue also ensures that the garbage collector knows about
               // this thread and its return value (it gets dropped from the
               // step->threads list so there's no other way to find it).
-              appendToRunQueue(cap,t);
+              joinRunQueue(cap,t);
               return rtsFalse;
 #else
               // this cannot happen in the threaded RTS, because a
@@ -2296 +2297 @@
 {
     // The thread goes at the *end* of the run-queue, to avoid possible
     // starvation of any threads already on the queue.
-    appendToRunQueue(cap,tso);
+    joinRunQueue(cap,tso);
 }
 
 void
@@ -2307 +2308 @@
 #if defined(THREADED_RTS)
     cpu %= enabled_capabilities;
     if (cpu == cap->no) {
-        appendToRunQueue(cap,tso);
+        joinRunQueue(cap,tso);
     } else {
         migrateThread(cap, tso, &capabilities[cpu]);
     }
 #else
-    appendToRunQueue(cap,tso);
+    joinRunQueue(cap,tso);
 #endif
 }
 
@@ -2337 +2338 @@
     task->incall->ret = ret;
     task->incall->stat = NoStatus;
 
-    appendToRunQueue(cap,tso);
+    joinRunQueue(cap,tso);
 
     DEBUG_ONLY( id = tso->id );
     debugTrace(DEBUG_sched, "new bound thread (%lu)", (unsigned long)id);

rts/Schedule.h

@@ -116 +116 @@
 
 void resurrectThreads (StgTSO *);
 
+// STRIDE1 defines the maximum resolution we can achieve in scheduling.
+#define STRIDE1 (1 << 20)
+// Defualt tickets is set to STRIDE1, so that the IO manager gets
+// maximum priority.
+#define DEFAULT_TICKETS (1 << 20)
+
 /* -----------------------------------------------------------------------------
  * Some convenient macros/inline functions...
  */
@@ -145 +151 @@
     cap->run_queue_tl = tso;
 }
 
+EXTERN_INLINE void
+joinRunQueue(Capability *cap, StgTSO *tso);
+
+EXTERN_INLINE void
+joinRunQueue(Capability *cap, StgTSO *tso)
+{
+    appendToRunQueue(cap, tso);
+}
+
 /* Push a thread on the beginning of the run queue.
  * ASSUMES: cap->running_task is the current task.
  */
@@ -165 +180 @@
     }
 }
 
+EXTERN_INLINE void
+fastJoinRunQueue(Capability *cap, StgTSO *tso);
+
+EXTERN_INLINE void
+fastJoinRunQueue(Capability *cap, StgTSO *tso)
+{
+    pushOnRunQueue(cap, tso);
+}
+
 /* Pop the first thread off the runnable queue.
  */
 INLINE_HEADER StgTSO *
@@ -189 +213 @@
     return cap->run_queue_hd;
 }
 
+EXTERN_INLINE void
+leaveRunQueue (Capability *cap, StgTSO *tso);
+
+EXTERN_INLINE void
+leaveRunQueue (Capability *cap, StgTSO *tso)
+{
+    // XXX implement me
+}
+
 void removeFromRunQueue (Capability *cap, StgTSO *tso);
 extern void promoteInRunQueue (Capability *cap, StgTSO *tso);
 

rts/Sparks.c

@@ -45 +45 @@
 
     traceEventCreateSparkThread(cap, tso->id);
 
-    appendToRunQueue(cap,tso);
+    joinRunQueue(cap,tso);
 }
 
 /* --------------------------------------------------------------------------

rts/Threads.c

@@ -112 +112 @@
 
     tso->trec = NO_TREC;
 
+    tso->ss_tickets = DEFAULT_TICKETS;
+
 #ifdef PROFILING
     tso->prof.cccs = CCS_MAIN;
 #endif
@@ -136 +138 @@
 }
 
 /* ---------------------------------------------------------------------------
+ * Ticket allocations on threads
+ * ------------------------------------------------------------------------ */
+
+#define TICKET_ERROR (STRIDE1 + 1)
+
+void
+setTickets(StgTSO *tso, W_ tickets)
+{
+    if (tickets > STRIDE1) {
+        barf("setTickets: too many tickets");
+    } else if (tickets <= 0) {
+        barf("setTickets: too few tickets");
+    }
+    ACQUIRE_LOCK(&sched_mutex);
+    tso->ss_tickets = tickets;
+    RELEASE_LOCK(&sched_mutex);
+}
+
+W_
+modifyTickets(StgTSO *tso, W_ n, W_ d, W_ x)
+{
+    ACQUIRE_LOCK(&sched_mutex);
+    W_ tickets = (tso->ss_tickets * n) / d + x;
+    W_ delta;
+    if (tickets > STRIDE1 || tickets <= 0) {
+        delta = TICKET_ERROR;
+        goto cleanup;
+    }
+    delta = tso->ss_tickets - tickets;
+    tso->ss_tickets = tickets;
+cleanup:
+    RELEASE_LOCK(&sched_mutex);
+    return delta;
+}
+
+W_
+getTickets(StgTSO *tso)
+{
+    return tso->ss_tickets;
+}
+
+/* ---------------------------------------------------------------------------
  * Comparing Thread ids.
  *
  * This is used from STG land in the implementation of the
@@ -296 +340 @@
     // just run the thread now, if the BH is not really available,
     // we'll block again.
     tso->why_blocked = NotBlocked;
-    appendToRunQueue(cap,tso);
+    joinRunQueue(cap, tso);
 
     // We used to set the context switch flag here, which would
     // trigger a context switch a short time in the future (at the end
@@ -322 +366 @@
     traceEventMigrateThread (from, tso, to->no);
     // ThreadMigrating tells the target cap that it needs to be added to
     // the run queue when it receives the MSG_TRY_WAKEUP.
+    leaveRunQueue(from, tso);
     tso->why_blocked = ThreadMigrating;
     tso->cap = to;
     tryWakeupThread(from, tso);

rts/Threads.h

@@ -13 +13 @@
 
 #define END_BLOCKED_EXCEPTIONS_QUEUE ((MessageThrowTo*)END_TSO_QUEUE)
 
+void setTickets(StgTSO *tso, W_ n);
+W_ modifyTickets(StgTSO *tso, W_ n, W_ d, W_ x);
+W_ getTickets(StgTSO *tso);
+
 StgTSO * unblockOne (Capability *cap, StgTSO *tso);
 StgTSO * unblockOne_ (Capability *cap, StgTSO *tso, rtsBool allow_migrate);
 

rts/posix/Select.c

@@ -107 +107 @@
         tso->_link = END_TSO_QUEUE;
         IF_DEBUG(scheduler,debugBelch("Waking up sleeping thread %lu\n", (unsigned long)tso->id));
         // MainCapability: this code is !THREADED_RTS
-        pushOnRunQueue(&MainCapability,tso);
+        fastJoinRunQueue(&MainCapability,tso);
         flag = rtsTrue;
     }
     return flag;
@@ -305 +305 @@
                 IF_DEBUG(scheduler,debugBelch("Waking up blocked thread %lu\n", (unsigned long)tso->id));
                   tso->why_blocked = NotBlocked;
                   tso->_link = END_TSO_QUEUE;
-                  pushOnRunQueue(&MainCapability,tso);
+                  fastJoinRunQueue(&MainCapability,tso);
               } else {
                   if (prev == NULL)
                       blocked_queue_hd = tso;

rts/win32/AsyncIO.c

@@ -303 +303 @@
                         // save the StgAsyncIOResult in the
                         // stg_block_async_info stack frame, because
                         // the block_info field will be overwritten by
-                        // pushOnRunQueue().
+                        // fastJoinRunQueue().
                         tso->stackobj->sp[1] = (W_)tso->block_info.async_result;
-                        pushOnRunQueue(&MainCapability, tso);
+                        fastJoinRunQueue(&MainCapability, tso);
                         break;
                     }
                     break;