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 #include "Storage.h"
  12 #include "Schedule.h"
  13 #include "SchedAPI.h"
  14 #include "RtsFlags.h"
  15 #include "RtsUtils.h"
  16 #include "ParTicky.h"
  17 #include "Trace.h"
  18 #include "Prelude.h"
  19
  20 #include "SMP.h" // for cas
  21
  22 #include "Sparks.h"
  23
  24 #if defined(THREADED_RTS) || defined(PARALLEL_HASKELL)
  25
  26 void
  27 initSparkPools( void )
  28 {
  29 #ifdef THREADED_RTS
  30     /* walk over the capabilities, allocating a spark pool for each one */
  31     nat i;
  32     for (i = 0; i < n_capabilities; i++) {
  33       capabilities[i].sparks = newWSDeque(RtsFlags.ParFlags.maxLocalSparks);
  34     }
  35 #else
  36     /* allocate a single spark pool */
  37     MainCapability->sparks = newWSDeque(RtsFlags.ParFlags.maxLocalSparks);
  38 #endif
  39 }
  40
  41 void
  42 freeSparkPool (SparkPool *pool)
  43 {
  44     freeWSDeque(pool);
  45 }
  46
  47 /* -----------------------------------------------------------------------------
  48  *
  49  * Turn a spark into a real thread
  50  *
  51  * -------------------------------------------------------------------------- */
  52
  53 void
  54 createSparkThread (Capability *cap)
  55 {
  56     StgTSO *tso;
  57
  58     tso = createIOThread (cap, RtsFlags.GcFlags.initialStkSize,
  59                           &base_GHCziConc_runSparks_closure);
  60
  61     if (cap->r.rCurrentTSO != NULL)
  62       // Capability in a bound thread?
  63       postEvent(cap, EVENT_SPARK_TO_THREAD, cap->r.rCurrentTSO->id, tso->id);
  64     else
  65       // Capability in a worker thread?
  66       postEvent(cap, EVENT_SPARK_TO_THREAD, 0, tso->id);
  67
  68     appendToRunQueue(cap,tso);
  69 }
  70
  71 /* --------------------------------------------------------------------------
  72  * newSpark: create a new spark, as a result of calling "par"
  73  * Called directly from STG.
  74  * -------------------------------------------------------------------------- */
  75
  76 StgInt
  77 newSpark (StgRegTable *reg, StgClosure *p)
  78 {
  79     Capability *cap = regTableToCapability(reg);
  80     SparkPool *pool = cap->sparks;
  81
  82     /* I am not sure whether this is the right thing to do.
  83      * Maybe it is better to exploit the tag information
  84      * instead of throwing it away?
  85      */
  86     p = UNTAG_CLOSURE(p);
  87
  88     if (closure_SHOULD_SPARK(p)) {
  89         pushWSDeque(pool,p);
  90     }
  91
  92     cap->sparks_created++;
  93
  94     postEvent(cap, EVENT_CREATE_SPARK, cap->r.rCurrentTSO->id, 0);
  95
  96     return 1;
  97 }
  98
  99 /* -----------------------------------------------------------------------------
 100  *
 101  * tryStealSpark: try to steal a spark from a Capability.
 102  *
 103  * Returns a valid spark, or NULL if the pool was empty, and can
 104  * occasionally return NULL if there was a race with another thread
 105  * stealing from the same pool.  In this case, try again later.
 106  *
 107  -------------------------------------------------------------------------- */
 108
 109 StgClosure *
 110 tryStealSpark (Capability *cap)
 111 {
 112   SparkPool *pool = cap->sparks;
 113   StgClosure *stolen;
 114
 115   do {
 116       stolen = stealWSDeque_(pool);
 117       // use the no-loopy version, stealWSDeque_(), since if we get a
 118       // spurious NULL here the caller may want to try stealing from
 119       // other pools before trying again.
 120   } while (stolen != NULL && !closure_SHOULD_SPARK(stolen));
 121
 122   return stolen;
 123 }
 124
 125 /* --------------------------------------------------------------------------
 126  * Remove all sparks from the spark queues which should not spark any
 127  * more.  Called after GC. We assume exclusive access to the structure
 128  * and replace  all sparks in the queue, see explanation below. At exit,
 129  * the spark pool only contains sparkable closures.
 130  * -------------------------------------------------------------------------- */
 131
 132 void
 133 pruneSparkQueue (evac_fn evac, void *user, Capability *cap)
 134 {
 135     SparkPool *pool;
 136     StgClosurePtr spark, tmp, *elements;
 137     nat n, pruned_sparks; // stats only
 138     StgWord botInd,oldBotInd,currInd; // indices in array (always < size)
 139     const StgInfoTable *info;
 140
 141     PAR_TICKY_MARK_SPARK_QUEUE_START();
 142
 143     n = 0;
 144     pruned_sparks = 0;
 145
 146     pool = cap->sparks;
 147
 148     // it is possible that top > bottom, indicating an empty pool.  We
 149     // fix that here; this is only necessary because the loop below
 150     // assumes it.
 151     if (pool->top > pool->bottom)
 152         pool->top = pool->bottom;
 153
 154     // Take this opportunity to reset top/bottom modulo the size of
 155     // the array, to avoid overflow.  This is only possible because no
 156     // stealing is happening during GC.
 157     pool->bottom  -= pool->top & ~pool->moduloSize;
 158     pool->top     &= pool->moduloSize;
 159     pool->topBound = pool->top;
 160
 161     debugTrace(DEBUG_sched,
 162                "markSparkQueue: current spark queue len=%ld; (hd=%ld; tl=%ld)",
 163                sparkPoolSize(pool), pool->bottom, pool->top);
 164
 165     ASSERT_WSDEQUE_INVARIANTS(pool);
 166
 167     elements = (StgClosurePtr *)pool->elements;
 168
 169     /* We have exclusive access to the structure here, so we can reset
 170        bottom and top counters, and prune invalid sparks. Contents are
 171        copied in-place if they are valuable, otherwise discarded. The
 172        routine uses "real" indices t and b, starts by computing them
 173        as the modulus size of top and bottom,
 174
 175        Copying:
 176
 177        At the beginning, the pool structure can look like this:
 178        ( bottom % size >= top % size , no wrap-around)
 179                   t          b
 180        ___________***********_________________
 181
 182        or like this ( bottom % size < top % size, wrap-around )
 183                   b         t
 184        ***********__________******************
 185        As we need to remove useless sparks anyway, we make one pass
 186        between t and b, moving valuable content to b and subsequent
 187        cells (wrapping around when the size is reached).
 188
 189                      b      t
 190        ***********OOO_______XX_X__X?**********
 191                      ^____move?____/
 192
 193        After this movement, botInd becomes the new bottom, and old
 194        bottom becomes the new top index, both as indices in the array
 195        size range.
 196     */
 197     // starting here
 198     currInd = (pool->top) & (pool->moduloSize); // mod
 199
 200     // copies of evacuated closures go to space from botInd on
 201     // we keep oldBotInd to know when to stop
 202     oldBotInd = botInd = (pool->bottom) & (pool->moduloSize); // mod
 203
 204     // on entry to loop, we are within the bounds
 205     ASSERT( currInd < pool->size && botInd  < pool->size );
 206
 207     while (currInd != oldBotInd ) {
 208       /* must use != here, wrap-around at size
 209          subtle: loop not entered if queue empty
 210        */
 211
 212       /* check element at currInd. if valuable, evacuate and move to
 213          botInd, otherwise move on */
 214       spark = elements[currInd];
 215
 216       // We have to be careful here: in the parallel GC, another
 217       // thread might evacuate this closure while we're looking at it,
 218       // so grab the info pointer just once.
 219       info = spark->header.info;
 220       if (IS_FORWARDING_PTR(info)) {
 221           tmp = (StgClosure*)UN_FORWARDING_PTR(info);
 222           /* if valuable work: shift inside the pool */
 223           if (closure_SHOULD_SPARK(tmp)) {
 224               elements[botInd] = tmp; // keep entry (new address)
 225               botInd++;
 226               n++;
 227           } else {
 228               pruned_sparks++; // discard spark
 229               cap->sparks_pruned++;
 230           }
 231       } else {
 232           if (!(closure_flags[INFO_PTR_TO_STRUCT(info)->type] & _NS)) {
 233               elements[botInd] = spark; // keep entry (new address)
 234               evac (user, &elements[botInd]);
 235               botInd++;
 236               n++;
 237           } else {
 238               pruned_sparks++; // discard spark
 239               cap->sparks_pruned++;
 240           }
 241       }
 242       currInd++;
 243
 244       // in the loop, we may reach the bounds, and instantly wrap around
 245       ASSERT( currInd <= pool->size && botInd <= pool->size );
 246       if ( currInd == pool->size ) { currInd = 0; }
 247       if ( botInd == pool->size )  { botInd = 0;  }
 248
 249     } // while-loop over spark pool elements
 250
 251     ASSERT(currInd == oldBotInd);
 252
 253     pool->top = oldBotInd; // where we started writing
 254     pool->topBound = pool->top;
 255
 256     pool->bottom = (oldBotInd <= botInd) ? botInd : (botInd + pool->size);
 257     // first free place we did not use (corrected by wraparound)
 258
 259     PAR_TICKY_MARK_SPARK_QUEUE_END(n);
 260
 261     debugTrace(DEBUG_sched, "pruned %d sparks", pruned_sparks);
 262
 263     debugTrace(DEBUG_sched,
 264                "new spark queue len=%ld; (hd=%ld; tl=%ld)",
 265                sparkPoolSize(pool), pool->bottom, pool->top);
 266
 267     ASSERT_WSDEQUE_INVARIANTS(pool);
 268 }
 269
 270 /* GC for the spark pool, called inside Capability.c for all
 271    capabilities in turn. Blindly "evac"s complete spark pool. */
 272 void
 273 traverseSparkQueue (evac_fn evac, void *user, Capability *cap)
 274 {
 275     StgClosure **sparkp;
 276     SparkPool *pool;
 277     StgWord top,bottom, modMask;
 278
 279     pool = cap->sparks;
 280
 281     ASSERT_WSDEQUE_INVARIANTS(pool);
 282
 283     top = pool->top;
 284     bottom = pool->bottom;
 285     sparkp = (StgClosurePtr*)pool->elements;
 286     modMask = pool->moduloSize;
 287
 288     while (top < bottom) {
 289     /* call evac for all closures in range (wrap-around via modulo)
 290      * In GHC-6.10, evac takes an additional 1st argument to hold a
 291      * GC-specific register, see rts/sm/GC.c::mark_root()
 292      */
 293       evac( user , sparkp + (top & modMask) );
 294       top++;
 295     }
 296
 297     debugTrace(DEBUG_sched,
 298                "traversed spark queue, len=%ld; (hd=%ld; tl=%ld)",
 299                sparkPoolSize(pool), pool->bottom, pool->top);
 300 }
 301
 302 /* ----------------------------------------------------------------------------
 303  * balanceSparkPoolsCaps: takes an array of capabilities (usually: all
 304  * capabilities) and its size. Accesses all spark pools and equally
 305  * distributes the sparks among them.
 306  *
 307  * Could be called after GC, before Cap. release, from scheduler.
 308  * -------------------------------------------------------------------------- */
 309 void balanceSparkPoolsCaps(nat n_caps, Capability caps[]);
 310
 311 void balanceSparkPoolsCaps(nat n_caps STG_UNUSED,
 312                            Capability caps[] STG_UNUSED) {
 313   barf("not implemented");
 314 }
 315
 316 #else
 317
 318 StgInt
 319 newSpark (StgRegTable *reg STG_UNUSED, StgClosure *p STG_UNUSED)
 320 {
 321     /* nothing */
 322     return 1;
 323 }
 324
 325
 326 #endif /* PARALLEL_HASKELL || THREADED_RTS */
 327
 328
 329 /* -----------------------------------------------------------------------------
 330  *
 331  * GRAN & PARALLEL_HASKELL stuff beyond here.
 332  *
 333  *  TODO "nuke" this!
 334  *
 335  * -------------------------------------------------------------------------- */
 336
 337 #if defined(PARALLEL_HASKELL) || defined(GRAN)
 338
 339 static void slide_spark_pool( StgSparkPool *pool );
 340
 341 rtsBool
 342 add_to_spark_queue( StgClosure *closure, StgSparkPool *pool )
 343 {
 344   if (pool->tl == pool->lim)
 345     slide_spark_pool(pool);
 346
 347   if (closure_SHOULD_SPARK(closure) &&
 348       pool->tl < pool->lim) {
 349     *(pool->tl++) = closure;
 350
 351 #if defined(PARALLEL_HASKELL)
 352     // collect parallel global statistics (currently done together with GC stats)
 353     if (RtsFlags.ParFlags.ParStats.Global &&
 354         RtsFlags.GcFlags.giveStats > NO_GC_STATS) {
 355       // debugBelch("Creating spark for %x @ %11.2f\n", closure, usertime());
 356       globalParStats.tot_sparks_created++;
 357     }
 358 #endif
 359     return rtsTrue;
 360   } else {
 361 #if defined(PARALLEL_HASKELL)
 362     // collect parallel global statistics (currently done together with GC stats)
 363     if (RtsFlags.ParFlags.ParStats.Global &&
 364         RtsFlags.GcFlags.giveStats > NO_GC_STATS) {
 365       //debugBelch("Ignoring spark for %x @ %11.2f\n", closure, usertime());
 366       globalParStats.tot_sparks_ignored++;
 367     }
 368 #endif
 369     return rtsFalse;
 370   }
 371 }
 372
 373 static void
 374 slide_spark_pool( StgSparkPool *pool )
 375 {
 376   StgClosure **sparkp, **to_sparkp;
 377
 378   sparkp = pool->hd;
 379   to_sparkp = pool->base;
 380   while (sparkp < pool->tl) {
 381     ASSERT(to_sparkp<=sparkp);
 382     ASSERT(*sparkp!=NULL);
 383     ASSERT(LOOKS_LIKE_GHC_INFO((*sparkp)->header.info));
 384
 385     if (closure_SHOULD_SPARK(*sparkp)) {
 386       *to_sparkp++ = *sparkp++;
 387     } else {
 388       sparkp++;
 389     }
 390   }
 391   pool->hd = pool->base;
 392   pool->tl = to_sparkp;
 393 }
 394
 395 void
 396 disposeSpark(spark)
 397 StgClosure *spark;
 398 {
 399 #if !defined(THREADED_RTS)
 400   Capability *cap;
 401   StgSparkPool *pool;
 402
 403   cap = &MainRegTable;
 404   pool = &(cap->rSparks);
 405   ASSERT(pool->hd <= pool->tl && pool->tl <= pool->lim);
 406 #endif
 407   ASSERT(spark != (StgClosure *)NULL);
 408   /* Do nothing */
 409 }
 410
 411
 412 #elif defined(GRAN)
 413
 414 /*
 415    Search the spark queue of the proc in event for a spark that's worth
 416    turning into a thread
 417    (was gimme_spark in the old RTS)
 418 */
 419 void
 420 findLocalSpark (rtsEvent *event, rtsBool *found_res, rtsSparkQ *spark_res)
 421 {
 422    PEs proc = event->proc,       /* proc to search for work */
 423        creator = event->creator; /* proc that requested work */
 424    StgClosure* node;
 425    rtsBool found;
 426    rtsSparkQ spark_of_non_local_node = NULL,
 427              spark_of_non_local_node_prev = NULL,
 428              low_priority_spark = NULL,
 429              low_priority_spark_prev = NULL,
 430              spark = NULL, prev = NULL;
 431
 432    /* Choose a spark from the local spark queue */
 433    prev = (rtsSpark*)NULL;
 434    spark = pending_sparks_hds[proc];
 435    found = rtsFalse;
 436
 437    // ToDo: check this code & implement local sparking !! -- HWL
 438    while (!found && spark != (rtsSpark*)NULL)
 439      {
 440        ASSERT((prev!=(rtsSpark*)NULL || spark==pending_sparks_hds[proc]) &&
 441               (prev==(rtsSpark*)NULL || prev->next==spark) &&
 442               (spark->prev==prev));
 443        node = spark->node;
 444        if (!closure_SHOULD_SPARK(node))
 445          {
 446            IF_GRAN_DEBUG(checkSparkQ,
 447                          debugBelch("^^ pruning spark %p (node %p) in gimme_spark",
 448                                spark, node));
 449
 450            if (RtsFlags.GranFlags.GranSimStats.Sparks)
 451              DumpRawGranEvent(proc, (PEs)0, SP_PRUNED,(StgTSO*)NULL,
 452                               spark->node, spark->name, spark_queue_len(proc));
 453
 454            ASSERT(spark != (rtsSpark*)NULL);
 455            ASSERT(SparksAvail>0);
 456            --SparksAvail;
 457
 458            ASSERT(prev==(rtsSpark*)NULL || prev->next==spark);
 459            spark = delete_from_sparkq (spark, proc, rtsTrue);
 460            if (spark != (rtsSpark*)NULL)
 461              prev = spark->prev;
 462            continue;
 463          }
 464        /* -- node should eventually be sparked */
 465        else if (RtsFlags.GranFlags.PreferSparksOfLocalNodes &&
 466                !IS_LOCAL_TO(PROCS(node),CurrentProc))
 467          {
 468            barf("Local sparking not yet implemented");
 469
 470            /* Remember first low priority spark */
 471            if (spark_of_non_local_node==(rtsSpark*)NULL) {
 472              spark_of_non_local_node_prev = prev;
 473              spark_of_non_local_node = spark;
 474               }
 475
 476            if (spark->next == (rtsSpark*)NULL) {
 477              /* ASSERT(spark==SparkQueueTl);  just for testing */
 478              prev = spark_of_non_local_node_prev;
 479              spark = spark_of_non_local_node;
 480              found = rtsTrue;
 481              break;
 482            }
 483
 484 # if defined(GRAN) && defined(GRAN_CHECK)
 485            /* Should never happen; just for testing
 486            if (spark==pending_sparks_tl) {
 487              debugBelch("ReSchedule: Last spark != SparkQueueTl\n");
 488                 stg_exit(EXIT_FAILURE);
 489                 } */
 490 # endif
 491            prev = spark;
 492            spark = spark->next;
 493            ASSERT(SparksAvail>0);
 494            --SparksAvail;
 495            continue;
 496          }
 497        else if ( RtsFlags.GranFlags.DoPrioritySparking ||
 498                  (spark->gran_info >= RtsFlags.GranFlags.SparkPriority2) )
 499          {
 500            if (RtsFlags.GranFlags.DoPrioritySparking)
 501              barf("Priority sparking not yet implemented");
 502
 503            found = rtsTrue;
 504          }
 505 #if 0
 506        else /* only used if SparkPriority2 is defined */
 507          {
 508            /* ToDo: fix the code below and re-integrate it */
 509            /* Remember first low priority spark */
 510            if (low_priority_spark==(rtsSpark*)NULL) {
 511              low_priority_spark_prev = prev;
 512              low_priority_spark = spark;
 513            }
 514
 515            if (spark->next == (rtsSpark*)NULL) {
 516                 /* ASSERT(spark==spark_queue_tl);  just for testing */
 517              prev = low_priority_spark_prev;
 518              spark = low_priority_spark;
 519              found = rtsTrue;       /* take low pri spark => rc is 2  */
 520              break;
 521            }
 522
 523            /* Should never happen; just for testing
 524            if (spark==pending_sparks_tl) {
 525              debugBelch("ReSchedule: Last spark != SparkQueueTl\n");
 526                 stg_exit(EXIT_FAILURE);
 527              break;
 528            } */
 529            prev = spark;
 530            spark = spark->next;
 531
 532            IF_GRAN_DEBUG(pri,
 533                          debugBelch("++ Ignoring spark of priority %u (SparkPriority=%u); node=%p; name=%u\n",
 534                                spark->gran_info, RtsFlags.GranFlags.SparkPriority,
 535                                spark->node, spark->name);)
 536            }
 537 #endif
 538    }  /* while (spark!=NULL && !found) */
 539
 540    *spark_res = spark;
 541    *found_res = found;
 542 }
 543
 544 /*
 545   Turn the spark into a thread.
 546   In GranSim this basically means scheduling a StartThread event for the
 547   node pointed to by the spark at some point in the future.
 548   (was munch_spark in the old RTS)
 549 */
 550 rtsBool
 551 activateSpark (rtsEvent *event, rtsSparkQ spark)
 552 {
 553   PEs proc = event->proc,       /* proc to search for work */
 554       creator = event->creator; /* proc that requested work */
 555   StgTSO* tso;
 556   StgClosure* node;
 557   rtsTime spark_arrival_time;
 558
 559   /*
 560      We've found a node on PE proc requested by PE creator.
 561      If proc==creator we can turn the spark into a thread immediately;
 562      otherwise we schedule a MoveSpark event on the requesting PE
 563   */
 564
 565   /* DaH Qu' yIchen */
 566   if (proc!=creator) {
 567
 568     /* only possible if we simulate GUM style fishing */
 569     ASSERT(RtsFlags.GranFlags.Fishing);
 570
 571     /* Message packing costs for sending a Fish; qeq jabbI'ID */
 572     CurrentTime[proc] += RtsFlags.GranFlags.Costs.mpacktime;
 573
 574     if (RtsFlags.GranFlags.GranSimStats.Sparks)
 575       DumpRawGranEvent(proc, (PEs)0, SP_EXPORTED,
 576                        (StgTSO*)NULL, spark->node,
 577                        spark->name, spark_queue_len(proc));
 578
 579     /* time of the spark arrival on the remote PE */
 580     spark_arrival_time = CurrentTime[proc] + RtsFlags.GranFlags.Costs.latency;
 581
 582     new_event(creator, proc, spark_arrival_time,
 583               MoveSpark,
 584               (StgTSO*)NULL, spark->node, spark);
 585
 586     CurrentTime[proc] += RtsFlags.GranFlags.Costs.mtidytime;
 587
 588   } else { /* proc==creator i.e. turn the spark into a thread */
 589
 590     if ( RtsFlags.GranFlags.GranSimStats.Global &&
 591          spark->gran_info < RtsFlags.GranFlags.SparkPriority2 ) {
 592
 593       globalGranStats.tot_low_pri_sparks++;
 594       IF_GRAN_DEBUG(pri,
 595                     debugBelch("++ No high priority spark available; low priority (%u) spark chosen: node=%p; name=%u\n",
 596                           spark->gran_info,
 597                           spark->node, spark->name));
 598     }
 599
 600     CurrentTime[proc] += RtsFlags.GranFlags.Costs.threadcreatetime;
 601
 602     node = spark->node;
 603
 604 # if 0
 605     /* ToDo: fix the GC interface and move to StartThread handling-- HWL */
 606     if (GARBAGE COLLECTION IS NECESSARY) {
 607       /* Some kind of backoff needed here in case there's too little heap */
 608 #  if defined(GRAN_CHECK) && defined(GRAN)
 609       if (RtsFlags.GcFlags.giveStats)
 610         fprintf(RtsFlags.GcFlags.statsFile,"***** vIS Qu' chen veQ boSwI'; spark=%p, node=%p;  name=%u\n",
 611                 /* (found==2 ? "no hi pri spark" : "hi pri spark"), */
 612                 spark, node, spark->name);
 613 #  endif
 614       new_event(CurrentProc, CurrentProc, CurrentTime[CurrentProc]+1,
 615                   FindWork,
 616                   (StgTSO*)NULL, (StgClosure*)NULL, (rtsSpark*)NULL);
 617       barf("//// activateSpark: out of heap ; ToDo: call GarbageCollect()");
 618       GarbageCollect(GetRoots, rtsFalse);
 619       // HWL old: ReallyPerformThreadGC(TSO_HS+TSO_CTS_SIZE,rtsFalse);
 620       // HWL old: SAVE_Hp -= TSO_HS+TSO_CTS_SIZE;
 621       spark = NULL;
 622       return; /* was: continue; */ /* to the next event, eventually */
 623     }
 624 # endif
 625
 626     if (RtsFlags.GranFlags.GranSimStats.Sparks)
 627       DumpRawGranEvent(CurrentProc,(PEs)0,SP_USED,(StgTSO*)NULL,
 628                        spark->node, spark->name,
 629                        spark_queue_len(CurrentProc));
 630
 631     new_event(proc, proc, CurrentTime[proc],
 632               StartThread,
 633               END_TSO_QUEUE, node, spark); // (rtsSpark*)NULL);
 634
 635     procStatus[proc] = Starting;
 636   }
 637 }
 638
 639 /* -------------------------------------------------------------------------
 640    This is the main point where handling granularity information comes into
 641    play.
 642    ------------------------------------------------------------------------- */
 643
 644 #define MAX_RAND_PRI    100
 645
 646 /*
 647    Granularity info transformers.
 648    Applied to the GRAN_INFO field of a spark.
 649 */
 650 STATIC_INLINE nat  ID(nat x) { return(x); };
 651 STATIC_INLINE nat  INV(nat x) { return(-x); };
 652 STATIC_INLINE nat  IGNORE(nat x) { return (0); };
 653 STATIC_INLINE nat  RAND(nat x) { return ((random() % MAX_RAND_PRI) + 1); }
 654
 655 /* NB: size_info and par_info are currently unused (what a shame!) -- HWL */
 656 rtsSpark *
 657 newSpark(node,name,gran_info,size_info,par_info,local)
 658 StgClosure *node;
 659 nat name, gran_info, size_info, par_info, local;
 660 {
 661   nat pri;
 662   rtsSpark *newspark;
 663
 664   pri = RtsFlags.GranFlags.RandomPriorities ? RAND(gran_info) :
 665         RtsFlags.GranFlags.InversePriorities ? INV(gran_info) :
 666         RtsFlags.GranFlags.IgnorePriorities ? IGNORE(gran_info) :
 667                            ID(gran_info);
 668
 669   if ( RtsFlags.GranFlags.SparkPriority!=0 &&
 670        pri<RtsFlags.GranFlags.SparkPriority ) {
 671     IF_GRAN_DEBUG(pri,
 672       debugBelch(",, NewSpark: Ignoring spark of priority %u (SparkPriority=%u); node=%#x; name=%u\n",
 673               pri, RtsFlags.GranFlags.SparkPriority, node, name));
 674     return ((rtsSpark*)NULL);
 675   }
 676
 677   newspark = (rtsSpark*) stgMallocBytes(sizeof(rtsSpark), "NewSpark");
 678   newspark->prev = newspark->next = (rtsSpark*)NULL;
 679   newspark->node = node;
 680   newspark->name = (name==1) ? CurrentTSO->gran.sparkname : name;
 681   newspark->gran_info = pri;
 682   newspark->global = !local;      /* Check that with parAt, parAtAbs !!*/
 683
 684   if (RtsFlags.GranFlags.GranSimStats.Global) {
 685     globalGranStats.tot_sparks_created++;
 686     globalGranStats.sparks_created_on_PE[CurrentProc]++;
 687   }
 688
 689   return(newspark);
 690 }
 691
 692 void
 693 disposeSpark(spark)
 694 rtsSpark *spark;
 695 {
 696   ASSERT(spark!=NULL);
 697   stgFree(spark);
 698 }
 699
 700 void
 701 disposeSparkQ(spark)
 702 rtsSparkQ spark;
 703 {
 704   if (spark==NULL)
 705     return;
 706
 707   disposeSparkQ(spark->next);
 708
 709 # ifdef GRAN_CHECK
 710   if (SparksAvail < 0) {
 711     debugBelch("disposeSparkQ: SparksAvail<0 after disposing sparkq @ %p\n", &spark);
 712     print_spark(spark);
 713   }
 714 # endif
 715
 716   stgFree(spark);
 717 }
 718
 719 /*
 720    With PrioritySparking add_to_spark_queue performs an insert sort to keep
 721    the spark queue sorted. Otherwise the spark is just added to the end of
 722    the queue.
 723 */
 724
 725 void
 726 add_to_spark_queue(spark)
 727 rtsSpark *spark;
 728 {
 729   rtsSpark *prev = NULL, *next = NULL;
 730   nat count = 0;
 731   rtsBool found = rtsFalse;
 732
 733   if ( spark == (rtsSpark *)NULL ) {
 734     return;
 735   }
 736
 737   if (RtsFlags.GranFlags.DoPrioritySparking && (spark->gran_info != 0) ) {
 738     /* Priority sparking is enabled i.e. spark queues must be sorted */
 739
 740     for (prev = NULL, next = pending_sparks_hd, count=0;
 741          (next != NULL) &&
 742          !(found = (spark->gran_info >= next->gran_info));
 743          prev = next, next = next->next, count++)
 744      {}
 745
 746   } else {   /* 'utQo' */
 747     /* Priority sparking is disabled */
 748
 749     found = rtsFalse;   /* to add it at the end */
 750
 751   }
 752
 753   if (found) {
 754     /* next points to the first spark with a gran_info smaller than that
 755        of spark; therefore, add spark before next into the spark queue */
 756     spark->next = next;
 757     if ( next == NULL ) {
 758       pending_sparks_tl = spark;
 759     } else {
 760       next->prev = spark;
 761     }
 762     spark->prev = prev;
 763     if ( prev == NULL ) {
 764       pending_sparks_hd = spark;
 765     } else {
 766       prev->next = spark;
 767     }
 768   } else {  /* (RtsFlags.GranFlags.DoPrioritySparking && !found) || !DoPrioritySparking */
 769     /* add the spark at the end of the spark queue */
 770     spark->next = NULL;
 771     spark->prev = pending_sparks_tl;
 772     if (pending_sparks_hd == NULL)
 773       pending_sparks_hd = spark;
 774     else
 775       pending_sparks_tl->next = spark;
 776     pending_sparks_tl = spark;
 777   }
 778   ++SparksAvail;
 779
 780   /* add costs for search in priority sparking */
 781   if (RtsFlags.GranFlags.DoPrioritySparking) {
 782     CurrentTime[CurrentProc] += count * RtsFlags.GranFlags.Costs.pri_spark_overhead;
 783   }
 784
 785   IF_GRAN_DEBUG(checkSparkQ,
 786                 debugBelch("++ Spark stats after adding spark %p (node %p) to queue on PE %d",
 787                       spark, spark->node, CurrentProc);
 788                 print_sparkq_stats());
 789
 790 #  if defined(GRAN_CHECK)
 791   if (RtsFlags.GranFlags.Debug.checkSparkQ) {
 792     for (prev = NULL, next =  pending_sparks_hd;
 793          (next != NULL);
 794          prev = next, next = next->next)
 795       {}
 796     if ( (prev!=NULL) && (prev!=pending_sparks_tl) )
 797       debugBelch("SparkQ inconsistency after adding spark %p: (PE %u) pending_sparks_tl (%p) not end of queue (%p)\n",
 798               spark,CurrentProc,
 799               pending_sparks_tl, prev);
 800   }
 801 #  endif
 802
 803 #  if defined(GRAN_CHECK)
 804   /* Check if the sparkq is still sorted. Just for testing, really!  */
 805   if ( RtsFlags.GranFlags.Debug.checkSparkQ &&
 806        RtsFlags.GranFlags.Debug.pri ) {
 807     rtsBool sorted = rtsTrue;
 808     rtsSpark *prev, *next;
 809
 810     if (pending_sparks_hd == NULL ||
 811         pending_sparks_hd->next == NULL ) {
 812       /* just 1 elem => ok */
 813     } else {
 814       for (prev = pending_sparks_hd,
 815            next = pending_sparks_hd->next;
 816            (next != NULL) ;
 817            prev = next, next = next->next) {
 818         sorted = sorted &&
 819                  (prev->gran_info >= next->gran_info);
 820       }
 821     }
 822     if (!sorted) {
 823       debugBelch("ghuH: SPARKQ on PE %d is not sorted:\n",
 824               CurrentProc);
 825       print_sparkq(CurrentProc);
 826     }
 827   }
 828 #  endif
 829 }
 830
 831 nat
 832 spark_queue_len(proc)
 833 PEs proc;
 834 {
 835  rtsSpark *prev, *spark;                     /* prev only for testing !! */
 836  nat len;
 837
 838  for (len = 0, prev = NULL, spark = pending_sparks_hds[proc];
 839       spark != NULL;
 840       len++, prev = spark, spark = spark->next)
 841    {}
 842
 843 #  if defined(GRAN_CHECK)
 844   if ( RtsFlags.GranFlags.Debug.checkSparkQ )
 845     if ( (prev!=NULL) && (prev!=pending_sparks_tls[proc]) )
 846       debugBelch("ERROR in spark_queue_len: (PE %u) pending_sparks_tl (%p) not end of queue (%p)\n",
 847               proc, pending_sparks_tls[proc], prev);
 848 #  endif
 849
 850  return (len);
 851 }
 852
 853 /*
 854    Take spark out of the spark queue on PE p and nuke the spark. Adjusts
 855    hd and tl pointers of the spark queue. Returns a pointer to the next
 856    spark in the queue.
 857 */
 858 rtsSpark *
 859 delete_from_sparkq (spark, p, dispose_too)     /* unlink and dispose spark */
 860 rtsSpark *spark;
 861 PEs p;
 862 rtsBool dispose_too;
 863 {
 864   rtsSpark *new_spark;
 865
 866   if (spark==NULL)
 867     barf("delete_from_sparkq: trying to delete NULL spark\n");
 868
 869 #  if defined(GRAN_CHECK)
 870   if ( RtsFlags.GranFlags.Debug.checkSparkQ ) {
 871     debugBelch("## |%p:%p| (%p)<-spark=%p->(%p) <-(%p)\n",
 872             pending_sparks_hd, pending_sparks_tl,
 873             spark->prev, spark, spark->next,
 874             (spark->next==NULL ? 0 : spark->next->prev));
 875   }
 876 #  endif
 877
 878   if (spark->prev==NULL) {
 879     /* spark is first spark of queue => adjust hd pointer */
 880     ASSERT(pending_sparks_hds[p]==spark);
 881     pending_sparks_hds[p] = spark->next;
 882   } else {
 883     spark->prev->next = spark->next;
 884   }
 885   if (spark->next==NULL) {
 886     ASSERT(pending_sparks_tls[p]==spark);
 887     /* spark is first spark of queue => adjust tl pointer */
 888     pending_sparks_tls[p] = spark->prev;
 889   } else {
 890     spark->next->prev = spark->prev;
 891   }
 892   new_spark = spark->next;
 893
 894 #  if defined(GRAN_CHECK)
 895   if ( RtsFlags.GranFlags.Debug.checkSparkQ ) {
 896     debugBelch("## |%p:%p| (%p)<-spark=%p->(%p) <-(%p); spark=%p will be deleted NOW \n",
 897             pending_sparks_hd, pending_sparks_tl,
 898             spark->prev, spark, spark->next,
 899             (spark->next==NULL ? 0 : spark->next->prev), spark);
 900   }
 901 #  endif
 902
 903   if (dispose_too)
 904     disposeSpark(spark);
 905
 906   return new_spark;
 907 }
 908
 909 /* Mark all nodes pointed to by sparks in the spark queues (for GC) */
 910 void
 911 markSparkQueue(void)
 912 {
 913   StgClosure *MarkRoot(StgClosure *root); // prototype
 914   PEs p;
 915   rtsSpark *sp;
 916
 917   for (p=0; p<RtsFlags.GranFlags.proc; p++)
 918     for (sp=pending_sparks_hds[p]; sp!=NULL; sp=sp->next) {
 919       ASSERT(sp->node!=NULL);
 920       ASSERT(LOOKS_LIKE_GHC_INFO(sp->node->header.info));
 921       // ToDo?: statistics gathering here (also for GUM!)
 922       sp->node = (StgClosure *)MarkRoot(sp->node);
 923     }
 924
 925   IF_DEBUG(gc,
 926            debugBelch("markSparkQueue: spark statistics at start of GC:");
 927            print_sparkq_stats());
 928 }
 929
 930 void
 931 print_spark(spark)
 932 rtsSpark *spark;
 933 {
 934   char str[16];
 935
 936   if (spark==NULL) {
 937     debugBelch("Spark: NIL\n");
 938     return;
 939   } else {
 940     sprintf(str,
 941             ((spark->node==NULL) ? "______" : "%#6lx"),
 942             stgCast(StgPtr,spark->node));
 943
 944     debugBelch("Spark: Node %8s, Name %#6x, Global %5s, Creator %5x, Prev %6p, Next %6p\n",
 945             str, spark->name,
 946             ((spark->global)==rtsTrue?"True":"False"), spark->creator,
 947             spark->prev, spark->next);
 948   }
 949 }
 950
 951 void
 952 print_sparkq(proc)
 953 PEs proc;
 954 // rtsSpark *hd;
 955 {
 956   rtsSpark *x = pending_sparks_hds[proc];
 957
 958   debugBelch("Spark Queue of PE %d with root at %p:\n", proc, x);
 959   for (; x!=(rtsSpark*)NULL; x=x->next) {
 960     print_spark(x);
 961   }
 962 }
 963
 964 /*
 965    Print a statistics of all spark queues.
 966 */
 967 void
 968 print_sparkq_stats(void)
 969 {
 970   PEs p;
 971
 972   debugBelch("SparkQs: [");
 973   for (p=0; p<RtsFlags.GranFlags.proc; p++)
 974     debugBelch(", PE %d: %d", p, spark_queue_len(p));
 975   debugBelch("\n");
 976 }
 977
 978 #endif