rts/Sparks.c

   1 /* ---------------------------------------------------------------------------
   2  *
   3  * (c) The GHC Team, 2000-2008
   4  *
   5  * Sparking support for PARALLEL_HASKELL and THREADED_RTS versions of the RTS.
   6  *
   7  -------------------------------------------------------------------------*/
   8
   9 #include "PosixSource.h"
  10 #include "Rts.h"
  11
  12 #include "Schedule.h"
  13 #include "RtsUtils.h"
  14 #include "Trace.h"
  15 #include "Prelude.h"
  16 #include "Sparks.h"
  17
  18 #if defined(THREADED_RTS)
  19
  20 void
  21 initSparkPools( void )
  22 {
  23     /* walk over the capabilities, allocating a spark pool for each one */
  24     nat i;
  25     for (i = 0; i < n_capabilities; i++) {
  26       capabilities[i].sparks = newWSDeque(RtsFlags.ParFlags.maxLocalSparks);
  27     }
  28 }
  29
  30 void
  31 freeSparkPool (SparkPool *pool)
  32 {
  33     freeWSDeque(pool);
  34 }
  35
  36 /* -----------------------------------------------------------------------------
  37  *
  38  * Turn a spark into a real thread
  39  *
  40  * -------------------------------------------------------------------------- */
  41
  42 void
  43 createSparkThread (Capability *cap)
  44 {
  45     StgTSO *tso;
  46
  47     tso = createIOThread (cap, RtsFlags.GcFlags.initialStkSize,
  48                           (StgClosure *)runSparks_closure);
  49
  50     traceEventCreateSparkThread(cap, tso->id);
  51
  52     appendToRunQueue(cap,tso);
  53 }
  54
  55 /* --------------------------------------------------------------------------
  56  * newSpark: create a new spark, as a result of calling "par"
  57  * Called directly from STG.
  58  * -------------------------------------------------------------------------- */
  59
  60 StgInt
  61 newSpark (StgRegTable *reg, StgClosure *p)
  62 {
  63     Capability *cap = regTableToCapability(reg);
  64     SparkPool *pool = cap->sparks;
  65
  66     /* I am not sure whether this is the right thing to do.
  67      * Maybe it is better to exploit the tag information
  68      * instead of throwing it away?
  69      */
  70     p = UNTAG_CLOSURE(p);
  71
  72     if (closure_SHOULD_SPARK(p)) {
  73         pushWSDeque(pool,p);
  74         cap->sparks_created++;
  75     } else {
  76         cap->sparks_dud++;
  77     }
  78
  79     return 1;
  80 }
  81
  82 /* -----------------------------------------------------------------------------
  83  *
  84  * tryStealSpark: try to steal a spark from a Capability.
  85  *
  86  * Returns a valid spark, or NULL if the pool was empty, and can
  87  * occasionally return NULL if there was a race with another thread
  88  * stealing from the same pool.  In this case, try again later.
  89  *
  90  -------------------------------------------------------------------------- */
  91
  92 StgClosure *
  93 tryStealSpark (Capability *cap)
  94 {
  95   SparkPool *pool = cap->sparks;
  96   StgClosure *stolen;
  97
  98   do {
  99       stolen = stealWSDeque_(pool);
 100       // use the no-loopy version, stealWSDeque_(), since if we get a
 101       // spurious NULL here the caller may want to try stealing from
 102       // other pools before trying again.
 103   } while (stolen != NULL && !closure_SHOULD_SPARK(stolen));
 104
 105   return stolen;
 106 }
 107
 108 /* --------------------------------------------------------------------------
 109  * Remove all sparks from the spark queues which should not spark any
 110  * more.  Called after GC. We assume exclusive access to the structure
 111  * and replace  all sparks in the queue, see explanation below. At exit,
 112  * the spark pool only contains sparkable closures.
 113  * -------------------------------------------------------------------------- */
 114
 115 void
 116 pruneSparkQueue (Capability *cap)
 117 {
 118     SparkPool *pool;
 119     StgClosurePtr spark, tmp, *elements;
 120     nat n, pruned_sparks; // stats only
 121     StgWord botInd,oldBotInd,currInd; // indices in array (always < size)
 122     const StgInfoTable *info;
 123
 124     n = 0;
 125     pruned_sparks = 0;
 126
 127     pool = cap->sparks;
 128
 129     // it is possible that top > bottom, indicating an empty pool.  We
 130     // fix that here; this is only necessary because the loop below
 131     // assumes it.
 132     if (pool->top > pool->bottom)
 133         pool->top = pool->bottom;
 134
 135     // Take this opportunity to reset top/bottom modulo the size of
 136     // the array, to avoid overflow.  This is only possible because no
 137     // stealing is happening during GC.
 138     pool->bottom  -= pool->top & ~pool->moduloSize;
 139     pool->top     &= pool->moduloSize;
 140     pool->topBound = pool->top;
 141
 142     debugTrace(DEBUG_sparks,
 143                "markSparkQueue: current spark queue len=%ld; (hd=%ld; tl=%ld)",
 144                sparkPoolSize(pool), pool->bottom, pool->top);
 145
 146     ASSERT_WSDEQUE_INVARIANTS(pool);
 147
 148     elements = (StgClosurePtr *)pool->elements;
 149
 150     /* We have exclusive access to the structure here, so we can reset
 151        bottom and top counters, and prune invalid sparks. Contents are
 152        copied in-place if they are valuable, otherwise discarded. The
 153        routine uses "real" indices t and b, starts by computing them
 154        as the modulus size of top and bottom,
 155
 156        Copying:
 157
 158        At the beginning, the pool structure can look like this:
 159        ( bottom % size >= top % size , no wrap-around)
 160                   t          b
 161        ___________***********_________________
 162
 163        or like this ( bottom % size < top % size, wrap-around )
 164                   b         t
 165        ***********__________******************
 166        As we need to remove useless sparks anyway, we make one pass
 167        between t and b, moving valuable content to b and subsequent
 168        cells (wrapping around when the size is reached).
 169
 170                      b      t
 171        ***********OOO_______XX_X__X?**********
 172                      ^____move?____/
 173
 174        After this movement, botInd becomes the new bottom, and old
 175        bottom becomes the new top index, both as indices in the array
 176        size range.
 177     */
 178     // starting here
 179     currInd = (pool->top) & (pool->moduloSize); // mod
 180
 181     // copies of evacuated closures go to space from botInd on
 182     // we keep oldBotInd to know when to stop
 183     oldBotInd = botInd = (pool->bottom) & (pool->moduloSize); // mod
 184
 185     // on entry to loop, we are within the bounds
 186     ASSERT( currInd < pool->size && botInd  < pool->size );
 187
 188     while (currInd != oldBotInd ) {
 189       /* must use != here, wrap-around at size
 190          subtle: loop not entered if queue empty
 191        */
 192
 193       /* check element at currInd. if valuable, evacuate and move to
 194          botInd, otherwise move on */
 195       spark = elements[currInd];
 196
 197       // We have to be careful here: in the parallel GC, another
 198       // thread might evacuate this closure while we're looking at it,
 199       // so grab the info pointer just once.
 200       if (GET_CLOSURE_TAG(spark) != 0) {
 201           // Tagged pointer is a value, so the spark has fizzled.  It
 202           // probably never happens that we get a tagged pointer in
 203           // the spark pool, because we would have pruned the spark
 204           // during the previous GC cycle if it turned out to be
 205           // evaluated, but it doesn't hurt to have this check for
 206           // robustness.
 207           pruned_sparks++;
 208           cap->sparks_fizzled++;
 209       } else {
 210           info = spark->header.info;
 211           if (IS_FORWARDING_PTR(info)) {
 212               tmp = (StgClosure*)UN_FORWARDING_PTR(info);
 213               /* if valuable work: shift inside the pool */
 214               if (closure_SHOULD_SPARK(tmp)) {
 215                   elements[botInd] = tmp; // keep entry (new address)
 216                   botInd++;
 217                   n++;
 218               } else {
 219                   pruned_sparks++; // discard spark
 220                   cap->sparks_fizzled++;
 221               }
 222           } else if (HEAP_ALLOCED(spark)) {
 223               if ((Bdescr((P_)spark)->flags & BF_EVACUATED)) {
 224                   if (closure_SHOULD_SPARK(spark)) {
 225                       elements[botInd] = spark; // keep entry (new address)
 226                       botInd++;
 227                       n++;
 228                   } else {
 229                       pruned_sparks++; // discard spark
 230                       cap->sparks_fizzled++;
 231                   }
 232               } else {
 233                   pruned_sparks++; // discard spark
 234                   cap->sparks_gcd++;
 235               }
 236           } else {
 237               if (INFO_PTR_TO_STRUCT(info)->type == THUNK_STATIC) {
 238                   if (*THUNK_STATIC_LINK(spark) != NULL) {
 239                       elements[botInd] = spark; // keep entry (new address)
 240                       botInd++;
 241                       n++;
 242                   } else {
 243                       pruned_sparks++; // discard spark
 244                       cap->sparks_gcd++;
 245                   }
 246               } else {
 247                   pruned_sparks++; // discard spark
 248                   cap->sparks_fizzled++;
 249               }
 250           }
 251       }
 252
 253       currInd++;
 254
 255       // in the loop, we may reach the bounds, and instantly wrap around
 256       ASSERT( currInd <= pool->size && botInd <= pool->size );
 257       if ( currInd == pool->size ) { currInd = 0; }
 258       if ( botInd == pool->size )  { botInd = 0;  }
 259
 260     } // while-loop over spark pool elements
 261
 262     ASSERT(currInd == oldBotInd);
 263
 264     pool->top = oldBotInd; // where we started writing
 265     pool->topBound = pool->top;
 266
 267     pool->bottom = (oldBotInd <= botInd) ? botInd : (botInd + pool->size);
 268     // first free place we did not use (corrected by wraparound)
 269
 270     debugTrace(DEBUG_sparks, "pruned %d sparks", pruned_sparks);
 271
 272     debugTrace(DEBUG_sparks,
 273                "new spark queue len=%ld; (hd=%ld; tl=%ld)",
 274                sparkPoolSize(pool), pool->bottom, pool->top);
 275
 276     ASSERT_WSDEQUE_INVARIANTS(pool);
 277 }
 278
 279 /* GC for the spark pool, called inside Capability.c for all
 280    capabilities in turn. Blindly "evac"s complete spark pool. */
 281 void
 282 traverseSparkQueue (evac_fn evac, void *user, Capability *cap)
 283 {
 284     StgClosure **sparkp;
 285     SparkPool *pool;
 286     StgWord top,bottom, modMask;
 287
 288     pool = cap->sparks;
 289
 290     ASSERT_WSDEQUE_INVARIANTS(pool);
 291
 292     top = pool->top;
 293     bottom = pool->bottom;
 294     sparkp = (StgClosurePtr*)pool->elements;
 295     modMask = pool->moduloSize;
 296
 297     while (top < bottom) {
 298     /* call evac for all closures in range (wrap-around via modulo)
 299      * In GHC-6.10, evac takes an additional 1st argument to hold a
 300      * GC-specific register, see rts/sm/GC.c::mark_root()
 301      */
 302       evac( user , sparkp + (top & modMask) );
 303       top++;
 304     }
 305
 306     debugTrace(DEBUG_sparks,
 307                "traversed spark queue, len=%ld; (hd=%ld; tl=%ld)",
 308                sparkPoolSize(pool), pool->bottom, pool->top);
 309 }
 310
 311 /* ----------------------------------------------------------------------------
 312  * balanceSparkPoolsCaps: takes an array of capabilities (usually: all
 313  * capabilities) and its size. Accesses all spark pools and equally
 314  * distributes the sparks among them.
 315  *
 316  * Could be called after GC, before Cap. release, from scheduler.
 317  * -------------------------------------------------------------------------- */
 318 void balanceSparkPoolsCaps(nat n_caps, Capability caps[])
 319    GNUC3_ATTRIBUTE(__noreturn__);
 320
 321 void balanceSparkPoolsCaps(nat n_caps STG_UNUSED,
 322                            Capability caps[] STG_UNUSED) {
 323   barf("not implemented");
 324 }
 325
 326 #else
 327
 328 StgInt
 329 newSpark (StgRegTable *reg STG_UNUSED, StgClosure *p STG_UNUSED)
 330 {
 331     /* nothing */
 332     return 1;
 333 }
 334
 335 #endif /* THREADED_RTS */