diff --git a/compiler/prelude/primops.txt.pp b/compiler/prelude/primops.txt.pp index 6d551d9..0ff7e2c 100644 --- a/compiler/prelude/primops.txt.pp +++ b/compiler/prelude/primops.txt.pp @@ -1839,6 +1839,24 @@ primop ThreadStatusOp "threadStatus#" GenPrimOp 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" ------------------------------------------------------------------------ diff --git a/includes/rts/storage/TSO.h b/includes/rts/storage/TSO.h index 82f5a75..20a67a2 100644 --- a/includes/rts/storage/TSO.h +++ b/includes/rts/storage/TSO.h @@ -168,6 +168,11 @@ typedef struct StgTSO_ { */ 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_ { diff --git a/includes/stg/MiscClosures.h b/includes/stg/MiscClosures.h index 61e6b09..e6d5fcd 100644 --- a/includes/stg/MiscClosures.h +++ b/includes/stg/MiscClosures.h @@ -402,6 +402,9 @@ RTS_FUN_DECL(stg_maskUninterruptiblezh); 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); diff --git a/rts/Capability.h b/rts/Capability.h index 3348f88..81322c8 100644 --- a/rts/Capability.h +++ b/rts/Capability.h @@ -58,6 +58,10 @@ struct Capability_ { 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 diff --git a/rts/Linker.c b/rts/Linker.c index fa1de89..c673bac 100644 --- a/rts/Linker.c +++ b/rts/Linker.c @@ -1130,6 +1130,9 @@ typedef struct _RtsSymbolVal { 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) \ diff --git a/rts/PrimOps.cmm b/rts/PrimOps.cmm index ebcee6a..a737508 100644 --- a/rts/PrimOps.cmm +++ b/rts/PrimOps.cmm @@ -628,6 +628,26 @@ stg_labelThreadzh ( gcptr threadid, W_ addr ) 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; diff --git a/rts/RaiseAsync.c b/rts/RaiseAsync.c index f5669cb..a32749c 100644 --- a/rts/RaiseAsync.c +++ b/rts/RaiseAsync.c @@ -683,7 +683,7 @@ removeFromQueues(Capability *cap, StgTSO *tso) done: tso->why_blocked = NotBlocked; - appendToRunQueue(cap, tso); + joinRunQueue(cap, tso); } /* ----------------------------------------------------------------------------- @@ -1045,7 +1045,7 @@ done: // wake it up if (tso->why_blocked != NotBlocked) { tso->why_blocked = NotBlocked; - appendToRunQueue(cap,tso); + joinRunQueue(cap,tso); } return tso; diff --git a/rts/Schedule.c b/rts/Schedule.c index a21b312..3de00c5 100644 --- a/rts/Schedule.c +++ b/rts/Schedule.c @@ -148,7 +148,7 @@ static void schedulePostRunThread(Capability *cap, StgTSO *t); 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,7 +537,7 @@ run_thread: break; case ThreadBlocked: - scheduleHandleThreadBlocked(t); + scheduleHandleThreadBlocked(cap, t); break; case ThreadFinished: @@ -781,7 +781,8 @@ schedulePushWork(Capability *cap USED_IF_THREADS, 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,11 +1218,7 @@ scheduleHandleYield( Capability *cap, StgTSO *t, nat prev_what_next ) * -------------------------------------------------------------------------- */ 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,6 +1231,8 @@ scheduleHandleThreadBlocked( StgTSO *t // threadPaused() might have raised a blocked throwTo // exception, see maybePerformBlockedException(). + leaveRunQueue(cap, t); + #ifdef DEBUG traceThreadStatus(DEBUG_sched, t); #endif @@ -1257,6 +1256,8 @@ scheduleHandleThreadFinished (Capability *cap STG_UNUSED, Task *task, StgTSO *t) // 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,7 +1278,7 @@ scheduleHandleThreadFinished (Capability *cap STG_UNUSED, Task *task, StgTSO *t) // 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,7 +2297,7 @@ scheduleThread(Capability *cap, StgTSO *tso) { // 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,12 +2308,12 @@ scheduleThreadOn(Capability *cap, StgWord cpu USED_IF_THREADS, StgTSO *tso) #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,7 +2338,7 @@ scheduleWaitThread (StgTSO* tso, /*[out]*/HaskellObj* ret, Capability **pcap) 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); diff --git a/rts/Schedule.h b/rts/Schedule.h index 8b7caea..e4425af 100644 --- a/rts/Schedule.h +++ b/rts/Schedule.h @@ -116,6 +116,12 @@ void interruptStgRts (void); 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,6 +151,15 @@ appendToRunQueue (Capability *cap, StgTSO *tso) 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,6 +180,15 @@ pushOnRunQueue (Capability *cap, StgTSO *tso) } } +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,6 +213,15 @@ peekRunQueue (Capability *cap) 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); diff --git a/rts/Sparks.c b/rts/Sparks.c index 4241656..4e9b5a5 100644 --- a/rts/Sparks.c +++ b/rts/Sparks.c @@ -45,7 +45,7 @@ createSparkThread (Capability *cap) traceEventCreateSparkThread(cap, tso->id); - appendToRunQueue(cap,tso); + joinRunQueue(cap,tso); } /* -------------------------------------------------------------------------- diff --git a/rts/Threads.c b/rts/Threads.c index b617616..758d368 100644 --- a/rts/Threads.c +++ b/rts/Threads.c @@ -112,6 +112,8 @@ createThread(Capability *cap, W_ size) tso->trec = NO_TREC; + tso->ss_tickets = DEFAULT_TICKETS; + #ifdef PROFILING tso->prof.cccs = CCS_MAIN; #endif @@ -136,6 +138,48 @@ createThread(Capability *cap, W_ size) } /* --------------------------------------------------------------------------- + * 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,7 +340,7 @@ unblock: // 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,6 +366,7 @@ migrateThread (Capability *from, StgTSO *tso, Capability *to) 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); diff --git a/rts/Threads.h b/rts/Threads.h index 6d26610..39794f3 100644 --- a/rts/Threads.h +++ b/rts/Threads.h @@ -13,6 +13,10 @@ #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); diff --git a/rts/posix/Select.c b/rts/posix/Select.c index 3d92a46..b39338b 100644 --- a/rts/posix/Select.c +++ b/rts/posix/Select.c @@ -107,7 +107,7 @@ static rtsBool wakeUpSleepingThreads (LowResTime now) 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,7 +305,7 @@ awaitEvent(rtsBool wait) 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; diff --git a/rts/win32/AsyncIO.c b/rts/win32/AsyncIO.c index 979df0c..cfd2c0f 100644 --- a/rts/win32/AsyncIO.c +++ b/rts/win32/AsyncIO.c @@ -303,9 +303,9 @@ start: // 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;