rts/Sparks.c

   1 /* ---------------------------------------------------------------------------
   2  *
   3  * (c) The GHC Team, 2000-2006
   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 # if defined(PARALLEL_HASKELL)
  18 # include "ParallelRts.h"
  19 # include "GranSimRts.h"   // for GR_...
  20 # elif defined(GRAN)
  21 # include "GranSimRts.h"
  22 # endif
  23 #include "Sparks.h"
  24 #include "Trace.h"
  25
  26 #if defined(THREADED_RTS) || defined(PARALLEL_HASKELL)
  27
  28 static INLINE_ME void bump_hd (StgSparkPool *p)
  29 { p->hd++; if (p->hd == p->lim) p->hd = p->base; }
  30
  31 static INLINE_ME void bump_tl (StgSparkPool *p)
  32 { p->tl++; if (p->tl == p->lim) p->tl = p->base; }
  33
  34 /* -----------------------------------------------------------------------------
  35  *
  36  * Initialising spark pools.
  37  *
  38  * -------------------------------------------------------------------------- */
  39
  40 static void
  41 initSparkPool(StgSparkPool *pool)
  42 {
  43     pool->base = stgMallocBytes(RtsFlags.ParFlags.maxLocalSparks
  44                                 * sizeof(StgClosure *),
  45                                 "initSparkPools");
  46     pool->lim = pool->base + RtsFlags.ParFlags.maxLocalSparks;
  47     pool->hd  = pool->base;
  48     pool->tl  = pool->base;
  49 }
  50
  51 void
  52 initSparkPools( void )
  53 {
  54 #ifdef THREADED_RTS
  55     /* walk over the capabilities, allocating a spark pool for each one */
  56     nat i;
  57     for (i = 0; i < n_capabilities; i++) {
  58         initSparkPool(&capabilities[i].r.rSparks);
  59     }
  60 #else
  61     /* allocate a single spark pool */
  62     initSparkPool(&MainCapability.r.rSparks);
  63 #endif
  64 }
  65
  66 void
  67 freeSparkPool(StgSparkPool *pool) {
  68     stgFree(pool->base);
  69 }
  70
  71 /* -----------------------------------------------------------------------------
  72  *
  73  * findSpark: find a spark on the current Capability that we can fork
  74  * into a thread.
  75  *
  76  * -------------------------------------------------------------------------- */
  77
  78 StgClosure *
  79 findSpark (Capability *cap)
  80 {
  81     StgSparkPool *pool;
  82     StgClosure *spark;
  83
  84     pool = &(cap->r.rSparks);
  85     ASSERT_SPARK_POOL_INVARIANTS(pool);
  86
  87     while (pool->hd != pool->tl) {
  88         spark = *pool->hd;
  89         bump_hd(pool);
  90         if (closure_SHOULD_SPARK(spark)) {
  91 #ifdef GRAN
  92             if (RtsFlags.ParFlags.ParStats.Sparks)
  93                 DumpRawGranEvent(CURRENT_PROC, CURRENT_PROC,
  94                                  GR_STEALING, ((StgTSO *)NULL), spark,
  95                                  0, 0 /* spark_queue_len(ADVISORY_POOL) */);
  96 #endif
  97             return spark;
  98         }
  99     }
 100     // spark pool is now empty
 101     return NULL;
 102 }
 103
 104 /* -----------------------------------------------------------------------------
 105  *
 106  * Turn a spark into a real thread
 107  *
 108  * -------------------------------------------------------------------------- */
 109
 110 void
 111 createSparkThread (Capability *cap, StgClosure *p)
 112 {
 113     StgTSO *tso;
 114
 115     tso = createGenThread (cap, RtsFlags.GcFlags.initialStkSize, p);
 116     appendToRunQueue(cap,tso);
 117 }
 118
 119 /* -----------------------------------------------------------------------------
 120  *
 121  * Create a new spark
 122  *
 123  * -------------------------------------------------------------------------- */
 124
 125 #define DISCARD_NEW
 126
 127 StgInt
 128 newSpark (StgRegTable *reg, StgClosure *p)
 129 {
 130     StgSparkPool *pool = &(reg->rSparks);
 131
 132     /* I am not sure whether this is the right thing to do.
 133      * Maybe it is better to exploit the tag information
 134      * instead of throwing it away?
 135      */
 136     p = UNTAG_CLOSURE(p);
 137
 138     ASSERT_SPARK_POOL_INVARIANTS(pool);
 139
 140     if (closure_SHOULD_SPARK(p)) {
 141 #ifdef DISCARD_NEW
 142         StgClosure **new_tl;
 143         new_tl = pool->tl + 1;
 144         if (new_tl == pool->lim) { new_tl = pool->base; }
 145         if (new_tl != pool->hd) {
 146             *pool->tl = p;
 147             pool->tl = new_tl;
 148         } else if (!closure_SHOULD_SPARK(*pool->hd)) {
 149             // if the old closure is not sparkable, discard it and
 150             // keep the new one.  Otherwise, keep the old one.
 151             *pool->tl = p;
 152             bump_hd(pool);
 153         }
 154 #else  /* DISCARD OLD */
 155         *pool->tl = p;
 156         bump_tl(pool);
 157         if (pool->tl == pool->hd) { bump_hd(pool); }
 158 #endif
 159     }
 160
 161     ASSERT_SPARK_POOL_INVARIANTS(pool);
 162     return 1;
 163 }
 164
 165 #else
 166
 167 StgInt
 168 newSpark (StgRegTable *reg STG_UNUSED, StgClosure *p STG_UNUSED)
 169 {
 170     /* nothing */
 171     return 1;
 172 }
 173
 174 #endif /* PARALLEL_HASKELL || THREADED_RTS */
 175
 176
 177 /* -----------------------------------------------------------------------------
 178  *
 179  * GRAN & PARALLEL_HASKELL stuff beyond here.
 180  *
 181  * -------------------------------------------------------------------------- */
 182
 183 #if defined(PARALLEL_HASKELL) || defined(GRAN)
 184
 185 static void slide_spark_pool( StgSparkPool *pool );
 186
 187 rtsBool
 188 add_to_spark_queue( StgClosure *closure, StgSparkPool *pool )
 189 {
 190   if (pool->tl == pool->lim)
 191     slide_spark_pool(pool);
 192
 193   if (closure_SHOULD_SPARK(closure) &&
 194       pool->tl < pool->lim) {
 195     *(pool->tl++) = closure;
 196
 197 #if defined(PARALLEL_HASKELL)
 198     // collect parallel global statistics (currently done together with GC stats)
 199     if (RtsFlags.ParFlags.ParStats.Global &&
 200         RtsFlags.GcFlags.giveStats > NO_GC_STATS) {
 201       // debugBelch("Creating spark for %x @ %11.2f\n", closure, usertime());
 202       globalParStats.tot_sparks_created++;
 203     }
 204 #endif
 205     return rtsTrue;
 206   } else {
 207 #if defined(PARALLEL_HASKELL)
 208     // collect parallel global statistics (currently done together with GC stats)
 209     if (RtsFlags.ParFlags.ParStats.Global &&
 210         RtsFlags.GcFlags.giveStats > NO_GC_STATS) {
 211       //debugBelch("Ignoring spark for %x @ %11.2f\n", closure, usertime());
 212       globalParStats.tot_sparks_ignored++;
 213     }
 214 #endif
 215     return rtsFalse;
 216   }
 217 }
 218
 219 static void
 220 slide_spark_pool( StgSparkPool *pool )
 221 {
 222   StgClosure **sparkp, **to_sparkp;
 223
 224   sparkp = pool->hd;
 225   to_sparkp = pool->base;
 226   while (sparkp < pool->tl) {
 227     ASSERT(to_sparkp<=sparkp);
 228     ASSERT(*sparkp!=NULL);
 229     ASSERT(LOOKS_LIKE_GHC_INFO((*sparkp)->header.info));
 230
 231     if (closure_SHOULD_SPARK(*sparkp)) {
 232       *to_sparkp++ = *sparkp++;
 233     } else {
 234       sparkp++;
 235     }
 236   }
 237   pool->hd = pool->base;
 238   pool->tl = to_sparkp;
 239 }
 240
 241 void
 242 disposeSpark(spark)
 243 StgClosure *spark;
 244 {
 245 #if !defined(THREADED_RTS)
 246   Capability *cap;
 247   StgSparkPool *pool;
 248
 249   cap = &MainRegTable;
 250   pool = &(cap->rSparks);
 251   ASSERT(pool->hd <= pool->tl && pool->tl <= pool->lim);
 252 #endif
 253   ASSERT(spark != (StgClosure *)NULL);
 254   /* Do nothing */
 255 }
 256
 257
 258 #elif defined(GRAN)
 259
 260 /*
 261    Search the spark queue of the proc in event for a spark that's worth
 262    turning into a thread
 263    (was gimme_spark in the old RTS)
 264 */
 265 void
 266 findLocalSpark (rtsEvent *event, rtsBool *found_res, rtsSparkQ *spark_res)
 267 {
 268    PEs proc = event->proc,       /* proc to search for work */
 269        creator = event->creator; /* proc that requested work */
 270    StgClosure* node;
 271    rtsBool found;
 272    rtsSparkQ spark_of_non_local_node = NULL,
 273              spark_of_non_local_node_prev = NULL,
 274              low_priority_spark = NULL,
 275              low_priority_spark_prev = NULL,
 276              spark = NULL, prev = NULL;
 277
 278    /* Choose a spark from the local spark queue */
 279    prev = (rtsSpark*)NULL;
 280    spark = pending_sparks_hds[proc];
 281    found = rtsFalse;
 282
 283    // ToDo: check this code & implement local sparking !! -- HWL
 284    while (!found && spark != (rtsSpark*)NULL)
 285      {
 286        ASSERT((prev!=(rtsSpark*)NULL || spark==pending_sparks_hds[proc]) &&
 287               (prev==(rtsSpark*)NULL || prev->next==spark) &&
 288               (spark->prev==prev));
 289        node = spark->node;
 290        if (!closure_SHOULD_SPARK(node))
 291          {
 292            IF_GRAN_DEBUG(checkSparkQ,
 293                          debugBelch("^^ pruning spark %p (node %p) in gimme_spark",
 294                                spark, node));
 295
 296            if (RtsFlags.GranFlags.GranSimStats.Sparks)
 297              DumpRawGranEvent(proc, (PEs)0, SP_PRUNED,(StgTSO*)NULL,
 298                               spark->node, spark->name, spark_queue_len(proc));
 299
 300            ASSERT(spark != (rtsSpark*)NULL);
 301            ASSERT(SparksAvail>0);
 302            --SparksAvail;
 303
 304            ASSERT(prev==(rtsSpark*)NULL || prev->next==spark);
 305            spark = delete_from_sparkq (spark, proc, rtsTrue);
 306            if (spark != (rtsSpark*)NULL)
 307              prev = spark->prev;
 308            continue;
 309          }
 310        /* -- node should eventually be sparked */
 311        else if (RtsFlags.GranFlags.PreferSparksOfLocalNodes &&
 312                !IS_LOCAL_TO(PROCS(node),CurrentProc))
 313          {
 314            barf("Local sparking not yet implemented");
 315
 316            /* Remember first low priority spark */
 317            if (spark_of_non_local_node==(rtsSpark*)NULL) {
 318              spark_of_non_local_node_prev = prev;
 319              spark_of_non_local_node = spark;
 320               }
 321
 322            if (spark->next == (rtsSpark*)NULL) {
 323              /* ASSERT(spark==SparkQueueTl);  just for testing */
 324              prev = spark_of_non_local_node_prev;
 325              spark = spark_of_non_local_node;
 326              found = rtsTrue;
 327              break;
 328            }
 329
 330 # if defined(GRAN) && defined(GRAN_CHECK)
 331            /* Should never happen; just for testing
 332            if (spark==pending_sparks_tl) {
 333              debugBelch("ReSchedule: Last spark != SparkQueueTl\n");
 334                 stg_exit(EXIT_FAILURE);
 335                 } */
 336 # endif
 337            prev = spark;
 338            spark = spark->next;
 339            ASSERT(SparksAvail>0);
 340            --SparksAvail;
 341            continue;
 342          }
 343        else if ( RtsFlags.GranFlags.DoPrioritySparking ||
 344                  (spark->gran_info >= RtsFlags.GranFlags.SparkPriority2) )
 345          {
 346            if (RtsFlags.GranFlags.DoPrioritySparking)
 347              barf("Priority sparking not yet implemented");
 348
 349            found = rtsTrue;
 350          }
 351 #if 0
 352        else /* only used if SparkPriority2 is defined */
 353          {
 354            /* ToDo: fix the code below and re-integrate it */
 355            /* Remember first low priority spark */
 356            if (low_priority_spark==(rtsSpark*)NULL) {
 357              low_priority_spark_prev = prev;
 358              low_priority_spark = spark;
 359            }
 360
 361            if (spark->next == (rtsSpark*)NULL) {
 362                 /* ASSERT(spark==spark_queue_tl);  just for testing */
 363              prev = low_priority_spark_prev;
 364              spark = low_priority_spark;
 365              found = rtsTrue;       /* take low pri spark => rc is 2  */
 366              break;
 367            }
 368
 369            /* Should never happen; just for testing
 370            if (spark==pending_sparks_tl) {
 371              debugBelch("ReSchedule: Last spark != SparkQueueTl\n");
 372                 stg_exit(EXIT_FAILURE);
 373              break;
 374            } */
 375            prev = spark;
 376            spark = spark->next;
 377
 378            IF_GRAN_DEBUG(pri,
 379                          debugBelch("++ Ignoring spark of priority %u (SparkPriority=%u); node=%p; name=%u\n",
 380                                spark->gran_info, RtsFlags.GranFlags.SparkPriority,
 381                                spark->node, spark->name);)
 382            }
 383 #endif
 384    }  /* while (spark!=NULL && !found) */
 385
 386    *spark_res = spark;
 387    *found_res = found;
 388 }
 389
 390 /*
 391   Turn the spark into a thread.
 392   In GranSim this basically means scheduling a StartThread event for the
 393   node pointed to by the spark at some point in the future.
 394   (was munch_spark in the old RTS)
 395 */
 396 rtsBool
 397 activateSpark (rtsEvent *event, rtsSparkQ spark)
 398 {
 399   PEs proc = event->proc,       /* proc to search for work */
 400       creator = event->creator; /* proc that requested work */
 401   StgTSO* tso;
 402   StgClosure* node;
 403   rtsTime spark_arrival_time;
 404
 405   /*
 406      We've found a node on PE proc requested by PE creator.
 407      If proc==creator we can turn the spark into a thread immediately;
 408      otherwise we schedule a MoveSpark event on the requesting PE
 409   */
 410
 411   /* DaH Qu' yIchen */
 412   if (proc!=creator) {
 413
 414     /* only possible if we simulate GUM style fishing */
 415     ASSERT(RtsFlags.GranFlags.Fishing);
 416
 417     /* Message packing costs for sending a Fish; qeq jabbI'ID */
 418     CurrentTime[proc] += RtsFlags.GranFlags.Costs.mpacktime;
 419
 420     if (RtsFlags.GranFlags.GranSimStats.Sparks)
 421       DumpRawGranEvent(proc, (PEs)0, SP_EXPORTED,
 422                        (StgTSO*)NULL, spark->node,
 423                        spark->name, spark_queue_len(proc));
 424
 425     /* time of the spark arrival on the remote PE */
 426     spark_arrival_time = CurrentTime[proc] + RtsFlags.GranFlags.Costs.latency;
 427
 428     new_event(creator, proc, spark_arrival_time,
 429               MoveSpark,
 430               (StgTSO*)NULL, spark->node, spark);
 431
 432     CurrentTime[proc] += RtsFlags.GranFlags.Costs.mtidytime;
 433
 434   } else { /* proc==creator i.e. turn the spark into a thread */
 435
 436     if ( RtsFlags.GranFlags.GranSimStats.Global &&
 437          spark->gran_info < RtsFlags.GranFlags.SparkPriority2 ) {
 438
 439       globalGranStats.tot_low_pri_sparks++;
 440       IF_GRAN_DEBUG(pri,
 441                     debugBelch("++ No high priority spark available; low priority (%u) spark chosen: node=%p; name=%u\n",
 442                           spark->gran_info,
 443                           spark->node, spark->name));
 444     }
 445
 446     CurrentTime[proc] += RtsFlags.GranFlags.Costs.threadcreatetime;
 447
 448     node = spark->node;
 449
 450 # if 0
 451     /* ToDo: fix the GC interface and move to StartThread handling-- HWL */
 452     if (GARBAGE COLLECTION IS NECESSARY) {
 453       /* Some kind of backoff needed here in case there's too little heap */
 454 #  if defined(GRAN_CHECK) && defined(GRAN)
 455       if (RtsFlags.GcFlags.giveStats)
 456         fprintf(RtsFlags.GcFlags.statsFile,"***** vIS Qu' chen veQ boSwI'; spark=%p, node=%p;  name=%u\n",
 457                 /* (found==2 ? "no hi pri spark" : "hi pri spark"), */
 458                 spark, node, spark->name);
 459 #  endif
 460       new_event(CurrentProc, CurrentProc, CurrentTime[CurrentProc]+1,
 461                   FindWork,
 462                   (StgTSO*)NULL, (StgClosure*)NULL, (rtsSpark*)NULL);
 463       barf("//// activateSpark: out of heap ; ToDo: call GarbageCollect()");
 464       GarbageCollect(GetRoots, rtsFalse);
 465       // HWL old: ReallyPerformThreadGC(TSO_HS+TSO_CTS_SIZE,rtsFalse);
 466       // HWL old: SAVE_Hp -= TSO_HS+TSO_CTS_SIZE;
 467       spark = NULL;
 468       return; /* was: continue; */ /* to the next event, eventually */
 469     }
 470 # endif
 471
 472     if (RtsFlags.GranFlags.GranSimStats.Sparks)
 473       DumpRawGranEvent(CurrentProc,(PEs)0,SP_USED,(StgTSO*)NULL,
 474                        spark->node, spark->name,
 475                        spark_queue_len(CurrentProc));
 476
 477     new_event(proc, proc, CurrentTime[proc],
 478               StartThread,
 479               END_TSO_QUEUE, node, spark); // (rtsSpark*)NULL);
 480
 481     procStatus[proc] = Starting;
 482   }
 483 }
 484
 485 /* -------------------------------------------------------------------------
 486    This is the main point where handling granularity information comes into
 487    play.
 488    ------------------------------------------------------------------------- */
 489
 490 #define MAX_RAND_PRI    100
 491
 492 /*
 493    Granularity info transformers.
 494    Applied to the GRAN_INFO field of a spark.
 495 */
 496 STATIC_INLINE nat  ID(nat x) { return(x); };
 497 STATIC_INLINE nat  INV(nat x) { return(-x); };
 498 STATIC_INLINE nat  IGNORE(nat x) { return (0); };
 499 STATIC_INLINE nat  RAND(nat x) { return ((random() % MAX_RAND_PRI) + 1); }
 500
 501 /* NB: size_info and par_info are currently unused (what a shame!) -- HWL */
 502 rtsSpark *
 503 newSpark(node,name,gran_info,size_info,par_info,local)
 504 StgClosure *node;
 505 nat name, gran_info, size_info, par_info, local;
 506 {
 507   nat pri;
 508   rtsSpark *newspark;
 509
 510   pri = RtsFlags.GranFlags.RandomPriorities ? RAND(gran_info) :
 511         RtsFlags.GranFlags.InversePriorities ? INV(gran_info) :
 512         RtsFlags.GranFlags.IgnorePriorities ? IGNORE(gran_info) :
 513                            ID(gran_info);
 514
 515   if ( RtsFlags.GranFlags.SparkPriority!=0 &&
 516        pri<RtsFlags.GranFlags.SparkPriority ) {
 517     IF_GRAN_DEBUG(pri,
 518       debugBelch(",, NewSpark: Ignoring spark of priority %u (SparkPriority=%u); node=%#x; name=%u\n",
 519               pri, RtsFlags.GranFlags.SparkPriority, node, name));
 520     return ((rtsSpark*)NULL);
 521   }
 522
 523   newspark = (rtsSpark*) stgMallocBytes(sizeof(rtsSpark), "NewSpark");
 524   newspark->prev = newspark->next = (rtsSpark*)NULL;
 525   newspark->node = node;
 526   newspark->name = (name==1) ? CurrentTSO->gran.sparkname : name;
 527   newspark->gran_info = pri;
 528   newspark->global = !local;      /* Check that with parAt, parAtAbs !!*/
 529
 530   if (RtsFlags.GranFlags.GranSimStats.Global) {
 531     globalGranStats.tot_sparks_created++;
 532     globalGranStats.sparks_created_on_PE[CurrentProc]++;
 533   }
 534
 535   return(newspark);
 536 }
 537
 538 void
 539 disposeSpark(spark)
 540 rtsSpark *spark;
 541 {
 542   ASSERT(spark!=NULL);
 543   stgFree(spark);
 544 }
 545
 546 void
 547 disposeSparkQ(spark)
 548 rtsSparkQ spark;
 549 {
 550   if (spark==NULL)
 551     return;
 552
 553   disposeSparkQ(spark->next);
 554
 555 # ifdef GRAN_CHECK
 556   if (SparksAvail < 0) {
 557     debugBelch("disposeSparkQ: SparksAvail<0 after disposing sparkq @ %p\n", &spark);
 558     print_spark(spark);
 559   }
 560 # endif
 561
 562   stgFree(spark);
 563 }
 564
 565 /*
 566    With PrioritySparking add_to_spark_queue performs an insert sort to keep
 567    the spark queue sorted. Otherwise the spark is just added to the end of
 568    the queue.
 569 */
 570
 571 void
 572 add_to_spark_queue(spark)
 573 rtsSpark *spark;
 574 {
 575   rtsSpark *prev = NULL, *next = NULL;
 576   nat count = 0;
 577   rtsBool found = rtsFalse;
 578
 579   if ( spark == (rtsSpark *)NULL ) {
 580     return;
 581   }
 582
 583   if (RtsFlags.GranFlags.DoPrioritySparking && (spark->gran_info != 0) ) {
 584     /* Priority sparking is enabled i.e. spark queues must be sorted */
 585
 586     for (prev = NULL, next = pending_sparks_hd, count=0;
 587          (next != NULL) &&
 588          !(found = (spark->gran_info >= next->gran_info));
 589          prev = next, next = next->next, count++)
 590      {}
 591
 592   } else {   /* 'utQo' */
 593     /* Priority sparking is disabled */
 594
 595     found = rtsFalse;   /* to add it at the end */
 596
 597   }
 598
 599   if (found) {
 600     /* next points to the first spark with a gran_info smaller than that
 601        of spark; therefore, add spark before next into the spark queue */
 602     spark->next = next;
 603     if ( next == NULL ) {
 604       pending_sparks_tl = spark;
 605     } else {
 606       next->prev = spark;
 607     }
 608     spark->prev = prev;
 609     if ( prev == NULL ) {
 610       pending_sparks_hd = spark;
 611     } else {
 612       prev->next = spark;
 613     }
 614   } else {  /* (RtsFlags.GranFlags.DoPrioritySparking && !found) || !DoPrioritySparking */
 615     /* add the spark at the end of the spark queue */
 616     spark->next = NULL;
 617     spark->prev = pending_sparks_tl;
 618     if (pending_sparks_hd == NULL)
 619       pending_sparks_hd = spark;
 620     else
 621       pending_sparks_tl->next = spark;
 622     pending_sparks_tl = spark;
 623   }
 624   ++SparksAvail;
 625
 626   /* add costs for search in priority sparking */
 627   if (RtsFlags.GranFlags.DoPrioritySparking) {
 628     CurrentTime[CurrentProc] += count * RtsFlags.GranFlags.Costs.pri_spark_overhead;
 629   }
 630
 631   IF_GRAN_DEBUG(checkSparkQ,
 632                 debugBelch("++ Spark stats after adding spark %p (node %p) to queue on PE %d",
 633                       spark, spark->node, CurrentProc);
 634                 print_sparkq_stats());
 635
 636 #  if defined(GRAN_CHECK)
 637   if (RtsFlags.GranFlags.Debug.checkSparkQ) {
 638     for (prev = NULL, next =  pending_sparks_hd;
 639          (next != NULL);
 640          prev = next, next = next->next)
 641       {}
 642     if ( (prev!=NULL) && (prev!=pending_sparks_tl) )
 643       debugBelch("SparkQ inconsistency after adding spark %p: (PE %u) pending_sparks_tl (%p) not end of queue (%p)\n",
 644               spark,CurrentProc,
 645               pending_sparks_tl, prev);
 646   }
 647 #  endif
 648
 649 #  if defined(GRAN_CHECK)
 650   /* Check if the sparkq is still sorted. Just for testing, really!  */
 651   if ( RtsFlags.GranFlags.Debug.checkSparkQ &&
 652        RtsFlags.GranFlags.Debug.pri ) {
 653     rtsBool sorted = rtsTrue;
 654     rtsSpark *prev, *next;
 655
 656     if (pending_sparks_hd == NULL ||
 657         pending_sparks_hd->next == NULL ) {
 658       /* just 1 elem => ok */
 659     } else {
 660       for (prev = pending_sparks_hd,
 661            next = pending_sparks_hd->next;
 662            (next != NULL) ;
 663            prev = next, next = next->next) {
 664         sorted = sorted &&
 665                  (prev->gran_info >= next->gran_info);
 666       }
 667     }
 668     if (!sorted) {
 669       debugBelch("ghuH: SPARKQ on PE %d is not sorted:\n",
 670               CurrentProc);
 671       print_sparkq(CurrentProc);
 672     }
 673   }
 674 #  endif
 675 }
 676
 677 nat
 678 spark_queue_len(proc)
 679 PEs proc;
 680 {
 681  rtsSpark *prev, *spark;                     /* prev only for testing !! */
 682  nat len;
 683
 684  for (len = 0, prev = NULL, spark = pending_sparks_hds[proc];
 685       spark != NULL;
 686       len++, prev = spark, spark = spark->next)
 687    {}
 688
 689 #  if defined(GRAN_CHECK)
 690   if ( RtsFlags.GranFlags.Debug.checkSparkQ )
 691     if ( (prev!=NULL) && (prev!=pending_sparks_tls[proc]) )
 692       debugBelch("ERROR in spark_queue_len: (PE %u) pending_sparks_tl (%p) not end of queue (%p)\n",
 693               proc, pending_sparks_tls[proc], prev);
 694 #  endif
 695
 696  return (len);
 697 }
 698
 699 /*
 700    Take spark out of the spark queue on PE p and nuke the spark. Adjusts
 701    hd and tl pointers of the spark queue. Returns a pointer to the next
 702    spark in the queue.
 703 */
 704 rtsSpark *
 705 delete_from_sparkq (spark, p, dispose_too)     /* unlink and dispose spark */
 706 rtsSpark *spark;
 707 PEs p;
 708 rtsBool dispose_too;
 709 {
 710   rtsSpark *new_spark;
 711
 712   if (spark==NULL)
 713     barf("delete_from_sparkq: trying to delete NULL spark\n");
 714
 715 #  if defined(GRAN_CHECK)
 716   if ( RtsFlags.GranFlags.Debug.checkSparkQ ) {
 717     debugBelch("## |%p:%p| (%p)<-spark=%p->(%p) <-(%p)\n",
 718             pending_sparks_hd, pending_sparks_tl,
 719             spark->prev, spark, spark->next,
 720             (spark->next==NULL ? 0 : spark->next->prev));
 721   }
 722 #  endif
 723
 724   if (spark->prev==NULL) {
 725     /* spark is first spark of queue => adjust hd pointer */
 726     ASSERT(pending_sparks_hds[p]==spark);
 727     pending_sparks_hds[p] = spark->next;
 728   } else {
 729     spark->prev->next = spark->next;
 730   }
 731   if (spark->next==NULL) {
 732     ASSERT(pending_sparks_tls[p]==spark);
 733     /* spark is first spark of queue => adjust tl pointer */
 734     pending_sparks_tls[p] = spark->prev;
 735   } else {
 736     spark->next->prev = spark->prev;
 737   }
 738   new_spark = spark->next;
 739
 740 #  if defined(GRAN_CHECK)
 741   if ( RtsFlags.GranFlags.Debug.checkSparkQ ) {
 742     debugBelch("## |%p:%p| (%p)<-spark=%p->(%p) <-(%p); spark=%p will be deleted NOW \n",
 743             pending_sparks_hd, pending_sparks_tl,
 744             spark->prev, spark, spark->next,
 745             (spark->next==NULL ? 0 : spark->next->prev), spark);
 746   }
 747 #  endif
 748
 749   if (dispose_too)
 750     disposeSpark(spark);
 751
 752   return new_spark;
 753 }
 754
 755 /* Mark all nodes pointed to by sparks in the spark queues (for GC) */
 756 void
 757 markSparkQueue(void)
 758 {
 759   StgClosure *MarkRoot(StgClosure *root); // prototype
 760   PEs p;
 761   rtsSpark *sp;
 762
 763   for (p=0; p<RtsFlags.GranFlags.proc; p++)
 764     for (sp=pending_sparks_hds[p]; sp!=NULL; sp=sp->next) {
 765       ASSERT(sp->node!=NULL);
 766       ASSERT(LOOKS_LIKE_GHC_INFO(sp->node->header.info));
 767       // ToDo?: statistics gathering here (also for GUM!)
 768       sp->node = (StgClosure *)MarkRoot(sp->node);
 769     }
 770
 771   IF_DEBUG(gc,
 772            debugBelch("markSparkQueue: spark statistics at start of GC:");
 773            print_sparkq_stats());
 774 }
 775
 776 void
 777 print_spark(spark)
 778 rtsSpark *spark;
 779 {
 780   char str[16];
 781
 782   if (spark==NULL) {
 783     debugBelch("Spark: NIL\n");
 784     return;
 785   } else {
 786     sprintf(str,
 787             ((spark->node==NULL) ? "______" : "%#6lx"),
 788             stgCast(StgPtr,spark->node));
 789
 790     debugBelch("Spark: Node %8s, Name %#6x, Global %5s, Creator %5x, Prev %6p, Next %6p\n",
 791             str, spark->name,
 792             ((spark->global)==rtsTrue?"True":"False"), spark->creator,
 793             spark->prev, spark->next);
 794   }
 795 }
 796
 797 void
 798 print_sparkq(proc)
 799 PEs proc;
 800 // rtsSpark *hd;
 801 {
 802   rtsSpark *x = pending_sparks_hds[proc];
 803
 804   debugBelch("Spark Queue of PE %d with root at %p:\n", proc, x);
 805   for (; x!=(rtsSpark*)NULL; x=x->next) {
 806     print_spark(x);
 807   }
 808 }
 809
 810 /*
 811    Print a statistics of all spark queues.
 812 */
 813 void
 814 print_sparkq_stats(void)
 815 {
 816   PEs p;
 817
 818   debugBelch("SparkQs: [");
 819   for (p=0; p<RtsFlags.GranFlags.proc; p++)
 820     debugBelch(", PE %d: %d", p, spark_queue_len(p));
 821   debugBelch("\n");
 822 }
 823
 824 #endif