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