File: FJTaskRunner.java

package info (click to toggle)
concurrent-dfsg 1.3.4-4
  • links: PTS, VCS
  • area: main
  • in suites: buster, jessie, jessie-kfreebsd, squeeze, stretch, wheezy
  • size: 976 kB
  • ctags: 2,018
  • sloc: java: 10,704; xml: 49; makefile: 12
file content (974 lines) | stat: -rw-r--r-- 29,129 bytes parent folder | download | duplicates (3)
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
562
563
564
565
566
567
568
569
570
571
572
573
574
575
576
577
578
579
580
581
582
583
584
585
586
587
588
589
590
591
592
593
594
595
596
597
598
599
600
601
602
603
604
605
606
607
608
609
610
611
612
613
614
615
616
617
618
619
620
621
622
623
624
625
626
627
628
629
630
631
632
633
634
635
636
637
638
639
640
641
642
643
644
645
646
647
648
649
650
651
652
653
654
655
656
657
658
659
660
661
662
663
664
665
666
667
668
669
670
671
672
673
674
675
676
677
678
679
680
681
682
683
684
685
686
687
688
689
690
691
692
693
694
695
696
697
698
699
700
701
702
703
704
705
706
707
708
709
710
711
712
713
714
715
716
717
718
719
720
721
722
723
724
725
726
727
728
729
730
731
732
733
734
735
736
737
738
739
740
741
742
743
744
745
746
747
748
749
750
751
752
753
754
755
756
757
758
759
760
761
762
763
764
765
766
767
768
769
770
771
772
773
774
775
776
777
778
779
780
781
782
783
784
785
786
787
788
789
790
791
792
793
794
795
796
797
798
799
800
801
802
803
804
805
806
807
808
809
810
811
812
813
814
815
816
817
818
819
820
821
822
823
824
825
826
827
828
829
830
831
832
833
834
835
836
837
838
839
840
841
842
843
844
845
846
847
848
849
850
851
852
853
854
855
856
857
858
859
860
861
862
863
864
865
866
867
868
869
870
871
872
873
874
875
876
877
878
879
880
881
882
883
884
885
886
887
888
889
890
891
892
893
894
895
896
897
898
899
900
901
902
903
904
905
906
907
908
909
910
911
912
913
914
915
916
917
918
919
920
921
922
923
924
925
926
927
928
929
930
931
932
933
934
935
936
937
938
939
940
941
942
943
944
945
946
947
948
949
950
951
952
953
954
955
956
957
958
959
960
961
962
963
964
965
966
967
968
969
970
971
972
973
974
/*
  File: FJTaskRunner.java

  Originally written by Doug Lea and released into the public domain.
  This may be used for any purposes whatsoever without acknowledgment.
  Thanks for the assistance and support of Sun Microsystems Labs,
  and everyone contributing, testing, and using this code.

  History:
  Date       Who                What
  7Jan1999   dl                 First public release
  13Jan1999  dl                 correct a stat counter update; 
                                ensure inactive status on run termination;
                                misc minor cleaup
  14Jan1999  dl                 Use random starting point in scan;
                                variable renamings.
  18Jan1999  dl                 Runloop allowed to die on task exception;
                                remove useless timed join
  22Jan1999  dl                 Rework scan to allow use of priorities.
  6Feb1999   dl                 Documentation updates.
  7Mar1999   dl                 Add array-based coInvoke
  31Mar1999  dl                 Revise scan to remove need for NullTasks
  27Apr1999  dl                 Renamed
  23oct1999  dl                 Earlier detect of interrupt in scanWhileIdling
  24nov1999  dl                 Now works on JVMs that do not properly
                                implement read-after-write of 2 volatiles.
*/

package EDU.oswego.cs.dl.util.concurrent;

import java.util.Random;

/**
 * Specialized Thread subclass for running FJTasks.
 * <p>
 * Each FJTaskRunner keeps FJTasks in a double-ended queue (DEQ).
 * Double-ended queues support stack-based operations
 * push and pop, as well as queue-based operations put and take.
 * Normally, threads run their own tasks. But they
 * may also steal tasks from each others DEQs.
 * <p>
 * The algorithms are minor variants of those used
 * in <A href="http://supertech.lcs.mit.edu/cilk/"> Cilk</A> and
 * <A href="http://www.cs.utexas.edu/users/hood/"> Hood</A>, and
 * to a lesser extent 
 * <A href="http://www.cs.uga.edu/~dkl/filaments/dist.html"> Filaments</A>,
 * but are adapted to work in Java.
 * <p>
 * The two most important capabilities are:
 * <ul>
 *  <li> Fork a FJTask: 
 *  <pre>
 *  Push task onto DEQ
 *  </pre>
 *  <li> Get a task to run (for example within taskYield)
 *  <pre>
 *  If DEQ is not empty, 
 *     Pop a task and run it.
 *  Else if any other DEQ is not empty, 
 *     Take ("steal") a task from it and run it.
 *  Else if the entry queue for our group is not empty,
 *     Take a task from it and run it.
 *  Else if current thread is otherwise idling
 *     If all threads are idling
 *        Wait for a task to be put on group entry queue
 *  Else
 *      Yield or Sleep for a while, and then retry
 *  </pre>
 * </ul>
 * The push, pop, and put are designed to only ever called by the
 * current thread, and take (steal) is only ever called by
 * other threads.
 * All other operations are composites and variants of these,
 * plus a few miscellaneous bookkeeping methods.
 * <p>
 * Implementations of the underlying representations and operations
 * are geared for use on JVMs operating on multiple CPUs (although
 * they should of course work fine on single CPUs as well).
 * <p>
 * A possible snapshot of a FJTaskRunner's DEQ is:
 * <pre>
 *     0     1     2     3     4     5     6    ...
 *  +-----+-----+-----+-----+-----+-----+-----+--
 *  |     |  t  |  t  |  t  |  t  |     |     | ...  deq array
 *  +-----+-----+-----+-----+-----+-----+-----+--
 *           ^                       ^
 *          base                    top 
 *   (incremented                     (incremented 
 *       on take,                         on push    
 *    decremented                     decremented
 *       on put)                          on pop)
 * </pre>
 * <p>
 * FJTasks are held in elements of the DEQ. 
 * They are maintained in a bounded array that
 * works similarly to a circular bounded buffer. To ensure
 * visibility of stolen FJTasks across threads, the array elements
 * must be <code>volatile</code>. 
 * Using volatile rather than synchronizing suffices here since
 * each task accessed by a thread is either one that it
 * created or one that has never seen before. Thus we cannot
 * encounter any staleness problems executing run methods,
 * although FJTask programmers must be still sure to either synch or use
 * volatile for shared data within their run methods.
 * <p>
 * However, since there is no way
 * to declare an array of volatiles in Java, the DEQ elements actually
 * hold VolatileTaskRef objects, each of which in turn holds a
 * volatile reference to a FJTask. 
 * Even with the double-indirection overhead of 
 * volatile refs, using an array for the DEQ works out
 * better than linking them since fewer shared
 * memory locations need to be
 * touched or modified by the threads while using the DEQ.
 * Further, the double indirection may alleviate cache-line
 * sharing effects (which cannot otherwise be directly dealt with in Java).
 * <p>
 * The indices for the <code>base</code> and <code>top</code> of the DEQ
 * are declared as volatile. The main contention point with
 * multiple FJTaskRunner threads occurs when one thread is trying
 * to pop its own stack while another is trying to steal from it.
 * This is handled via a specialization of Dekker's algorithm,
 * in which the popping thread pre-decrements <code>top</code>,
 * and then checks it against <code>base</code>. 
 * To be conservative in the face of JVMs that only partially
 * honor the specification for volatile, the pop proceeds
 * without synchronization only if there are apparently enough
 * items for both a simultaneous pop and take to succeed.
 * It otherwise enters a 
 * synchronized lock to check if the DEQ is actually empty,
 * if so failing. The stealing thread
 * does almost the opposite, but is set up to be less likely
 * to win in cases of contention: Steals always run under synchronized
 * locks in order to avoid conflicts with other ongoing steals.
 * They pre-increment <code>base</code>, and then check against
 * <code>top</code>. They back out (resetting the base index 
 * and failing to steal) if the
 * DEQ is empty or is about to become empty by an ongoing pop.
 * <p>
 * A push operation can normally run concurrently with a steal.
 * A push enters a synch lock only if the DEQ appears full so must
 * either be resized or have indices adjusted due to wrap-around
 * of the bounded DEQ. The put operation always requires synchronization.
 * <p>
 * When a FJTaskRunner thread has no tasks of its own to run, 
 * it tries to be a good citizen. 
 * Threads run at lower priority while scanning for work.
 * <p>
 * If the task is currently waiting
 * via yield, the thread alternates scans (starting at a randomly 
 * chosen victim) with Thread.yields. This is
 * well-behaved so long as the JVM handles Thread.yield in a
 * sensible fashion. (It need not. Thread.yield is so underspecified
 * that it is legal for a JVM to treat it as a no-op.) This also
 * keeps things well-behaved even if we are running on a uniprocessor
 * JVM using a simple cooperative threading model.
 * <p>
 * If a thread needing work is
 * is otherwise idle (which occurs only in the main runloop), and
 * there are no available tasks to steal or poll, it
 * instead enters into a sleep-based (actually timed wait(msec))
 * phase in which it progressively sleeps for longer durations
 * (up to a maximum of FJTaskRunnerGroup.MAX_SLEEP_TIME,
 * currently 100ms) between scans. 
 * If all threads in the group
 * are idling, they further progress to a hard wait phase, suspending
 * until a new task is entered into the FJTaskRunnerGroup entry queue.
 * A sleeping FJTaskRunner thread may be awakened by a new
 * task being put into the group entry queue or by another FJTaskRunner
 * becoming active, but not merely by some DEQ becoming non-empty.
 * Thus the MAX_SLEEP_TIME provides a bound for sleep durations
 * in cases where all but one worker thread start sleeping
 * even though there will eventually be work produced
 * by a thread that is taking a long time to place tasks in DEQ.
 * These sleep mechanics are handled in the FJTaskRunnerGroup class.
 * <p>
 * Composite operations such as taskJoin include heavy
 * manual inlining of the most time-critical operations
 * (mainly FJTask.invoke). 
 * This opens up a few opportunities for further hand-optimizations. 
 * Until Java compilers get a lot smarter, these tweaks
 * improve performance significantly enough for task-intensive 
 * programs to be worth the poorer maintainability and code duplication.
 * <p>
 * Because they are so fragile and performance-sensitive, nearly
 * all methods are declared as final. However, nearly all fields
 * and methods are also declared as protected, so it is possible,
 * with much care, to extend functionality in subclasses. (Normally
 * you would also need to subclass FJTaskRunnerGroup.)
 * <p>
 * None of the normal java.lang.Thread class methods should ever be called
 * on FJTaskRunners. For this reason, it might have been nicer to
 * declare FJTaskRunner as a Runnable to run within a Thread. However,
 * this would have complicated many minor logistics. And since
 * no FJTaskRunner methods should normally be called from outside the
 * FJTask and FJTaskRunnerGroup classes either, this decision doesn't impact
 * usage.
 * <p>
 * You might think that layering this kind of framework on top of
 * Java threads, which are already several levels removed from raw CPU
 * scheduling on most systems, would lead to very poor performance. 
 * But on the platforms
 * tested, the performance is quite good.
 * <p>[<a href="http://gee.cs.oswego.edu/dl/classes/EDU/oswego/cs/dl/util/concurrent/intro.html"> Introduction to this package. </a>]
 * @see FJTask
 * @see FJTaskRunnerGroup
 **/

public class FJTaskRunner extends Thread {
  
  /** The group of which this FJTaskRunner is a member **/
  protected final FJTaskRunnerGroup group;

  /**
   *  Constructor called only during FJTaskRunnerGroup initialization
   **/

  protected FJTaskRunner(FJTaskRunnerGroup g) { 
    group = g;
    victimRNG = new Random(System.identityHashCode(this));
    runPriority = getPriority();
    setDaemon(true);
  }

  /**
   * Return the FJTaskRunnerGroup of which this thread is a member
   **/
  
  protected final FJTaskRunnerGroup getGroup() { return group; }


  /* ------------ DEQ Representation ------------------- */


  /**
   * FJTasks are held in an array-based DEQ with INITIAL_CAPACITY
   * elements. The DEQ is grown if necessary, but default value is
   * normally much more than sufficient unless  there are
   * user programming errors or questionable operations generating
   * large numbers of Tasks without running them.
   * Capacities must be a power of two. 
   **/

  protected static final int INITIAL_CAPACITY = 4096; 

  /**
   * The maximum supported DEQ capacity.
   * When exceeded, FJTaskRunner operations throw Errors
   **/

  protected static final int MAX_CAPACITY = 1 << 30;

  /**
   * An object holding a single volatile reference to a FJTask.
   **/
  
  protected final static class VolatileTaskRef {
    /** The reference **/
    protected volatile FJTask ref;

    /** Set the reference **/
    protected final void put(FJTask r) { ref = r; }
    /** Return the reference **/
    protected final FJTask get()     { return ref; }
    /** Return the reference and clear it **/
    protected final FJTask take()    { FJTask r = ref; ref = null; return r;  }

    /**
     * Initialization utility for constructing arrays. 
     * Make an array of given capacity and fill it with
     * VolatileTaskRefs.
     **/
    protected static VolatileTaskRef[] newArray(int cap) {
      VolatileTaskRef[] a = new VolatileTaskRef[cap];
      for (int k = 0; k < cap; k++) a[k] = new VolatileTaskRef();
      return a;
    }

  }

  /**
   * The DEQ array.
   **/
    
  protected VolatileTaskRef[] deq = VolatileTaskRef.newArray(INITIAL_CAPACITY);

  /** Current size of the task DEQ **/
  protected int deqSize() { return deq.length; }

  /** 
   * Current top of DEQ. Generally acts just like a stack pointer in an 
   * array-based stack, except that it circularly wraps around the
   * array, as in an array-based queue. The value is NOT
   * always kept within <code>0 ... deq.length</code> though. 
   * The current top element is always at <code>top & (deq.length-1)</code>.
   * To avoid integer overflow, top is reset down 
   * within bounds whenever it is noticed to be out out bounds;
   * at worst when it is at <code>2 * deq.length</code>.
   **/
  protected volatile int top = 0;


  /** 
   * Current base of DEQ. Acts like a take-pointer in an
   * array-based bounded queue. Same bounds and usage as top.
   **/

  protected volatile int base = 0;


  /**
   * An extra object to synchronize on in order to
   * achieve a memory barrier.
   **/

  protected final Object barrier = new Object();

  /* ------------ Other BookKeeping ------------------- */

  /**
   * Record whether current thread may be processing a task
   * (i.e., has been started and is not in an idle wait).
   * Accessed, under synch, ONLY by FJTaskRunnerGroup, but the field is
   * stored here for simplicity.
   **/

  protected boolean active = false;

  /** Random starting point generator for scan() **/
  protected final Random victimRNG;


  /** Priority to use while scanning for work **/
  protected int scanPriority = FJTaskRunnerGroup.DEFAULT_SCAN_PRIORITY;

  /** Priority to use while running tasks **/
  protected int runPriority;

  /**
   * Set the priority to use while scanning.
   * We do not bother synchronizing access, since
   * by the time the value is needed, both this FJTaskRunner 
   * and its FJTaskRunnerGroup will
   * necessarily have performed enough synchronization
   * to avoid staleness problems of any consequence.
   **/
  protected void setScanPriority(int pri) { scanPriority = pri; }


  /**
   * Set the priority to use while running tasks.
   * Same usage and rationale as setScanPriority.
   **/
  protected void setRunPriority(int pri) {  runPriority = pri; }

  /**
   * Compile-time constant for statistics gathering.
   * Even when set, reported values may not be accurate
   * since all are read and written without synchronization.
   **/



  static final boolean COLLECT_STATS = true;
  // static final boolean COLLECT_STATS = false;


  // for stat collection

  /** Total number of tasks run **/
  protected int runs = 0;

  /** Total number of queues scanned for work **/
  protected int scans = 0;

  /** Total number of tasks obtained via scan **/
  protected int steals = 0;




  /* ------------ DEQ operations ------------------- */


  /**
   * Push a task onto DEQ.
   * Called ONLY by current thread.
   **/

  protected final void push(final FJTask r) {
    int t = top;

    /*
      This test catches both overflows and index wraps.  It doesn't
      really matter if base value is in the midst of changing in take. 
      As long as deq length is < 2^30, we are guaranteed to catch wrap in
      time since base can only be incremented at most length times
      between pushes (or puts). 
    */

    if (t < (base & (deq.length-1)) + deq.length) {

      deq[t & (deq.length-1)].put(r);
      top = t + 1;
    }

    else  // isolate slow case to increase chances push is inlined
      slowPush(r); // check overflow and retry
  }


  /**
   * Handle slow case for push
   **/

  protected synchronized void slowPush(final FJTask r) {
    checkOverflow();
    push(r); // just recurse -- this one is sure to succeed.
  }


  /**
   * Enqueue task at base of DEQ.
   * Called ONLY by current thread.
   * This method is currently not called from class FJTask. It could be used
   * as a faster way to do FJTask.start, but most users would
   * find the semantics too confusing and unpredictable.
   **/

  protected final synchronized void put(final FJTask r) {
    for (;;) {
      int b = base - 1;
      if (top < b + deq.length) {
        
        int newBase = b & (deq.length-1);
        deq[newBase].put(r);
        base = newBase;
        
        if (b != newBase) { // Adjust for index underflow
          int newTop = top & (deq.length-1);
          if (newTop < newBase) newTop += deq.length;
          top = newTop;
        }
        return;
      }
      else {
        checkOverflow();
        // ... and retry
      }
    }
  }

  /**
   * Return a popped task, or null if DEQ is empty.
   * Called ONLY by current thread.
   * <p>
   * This is not usually called directly but is
   * instead inlined in callers. This version differs from the
   * cilk algorithm in that pop does not fully back down and
   * retry in the case of potential conflict with take. It simply
   * rechecks under synch lock. This gives a preference
   * for threads to run their own tasks, which seems to
   * reduce flailing a bit when there are few tasks to run.
   **/

  protected final FJTask pop() {
    /* 
       Decrement top, to force a contending take to back down.
    */

    int t = --top;      

    /*
      To avoid problems with JVMs that do not properly implement
      read-after-write of a pair of volatiles, we conservatively
      grab without lock only if the DEQ appears to have at least two
      elements, thus guaranteeing that both a pop and take will succeed,
      even if the pre-increment in take is not seen by current thread.
      Otherwise we recheck under synch.
    */

    if (base + 1 < t) 
      return deq[t & (deq.length-1)].take();
    else
      return confirmPop(t);

  }


  /**
   * Check under synch lock if DEQ is really empty when doing pop. 
   * Return task if not empty, else null.
   **/

  protected final synchronized FJTask confirmPop(int provisionalTop) {
    if (base <= provisionalTop) 
      return deq[provisionalTop & (deq.length-1)].take();
    else {    // was empty
      /*
        Reset DEQ indices to zero whenever it is empty.
        This both avoids unnecessary calls to checkOverflow
        in push, and helps keep the DEQ from accumulating garbage
      */

      top = base = 0;
      return null;
    }
  }


  /** 
   * Take a task from the base of the DEQ.
   * Always called by other threads via scan()
   **/

  
  protected final synchronized FJTask take() {

    /*
      Increment base in order to suppress a contending pop
    */
    
    int b = base++;     
    
    if (b < top) 
      return confirmTake(b);
    else {
      // back out
      base = b; 
      return null;
    }
  }


  /**
   * double-check a potential take
   **/
  
  protected FJTask confirmTake(int oldBase) {

    /*
      Use a second (guaranteed uncontended) synch
      to serve as a barrier in case JVM does not
      properly process read-after-write of 2 volatiles
    */

    synchronized(barrier) {
      if (oldBase < top) {
        /*
          We cannot call deq[oldBase].take here because of possible races when
          nulling out versus concurrent push operations.  Resulting
          accumulated garbage is swept out periodically in
          checkOverflow, or more typically, just by keeping indices
          zero-based when found to be empty in pop, which keeps active
          region small and constantly overwritten. 
        */
        
        return deq[oldBase & (deq.length-1)].get();
      }
      else {
        base = oldBase;
        return null;
      }
    }
  }


  /**
   * Adjust top and base, and grow DEQ if necessary.
   * Called only while DEQ synch lock being held.
   * We don't expect this to be called very often. In most
   * programs using FJTasks, it is never called.
   **/

  protected void checkOverflow() { 
    int t = top;
    int b = base;
    
    if (t - b < deq.length-1) { // check if just need an index reset
      
      int newBase = b & (deq.length-1);
      int newTop  = top & (deq.length-1);
      if (newTop < newBase) newTop += deq.length;
      top = newTop;
      base = newBase;
      
      /* 
         Null out refs to stolen tasks. 
         This is the only time we can safely do it.
      */
      
      int i = newBase;
      while (i != newTop && deq[i].ref != null) {
        deq[i].ref = null;
        i = (i - 1) & (deq.length-1);
      }
      
    }
    else { // grow by doubling array
      
      int newTop = t - b;
      int oldcap = deq.length;
      int newcap = oldcap * 2;
      
      if (newcap >= MAX_CAPACITY)
        throw new Error("FJTask queue maximum capacity exceeded");
      
      VolatileTaskRef[] newdeq = new VolatileTaskRef[newcap];
      
      // copy in bottom half of new deq with refs from old deq
      for (int j = 0; j < oldcap; ++j) newdeq[j] = deq[b++ & (oldcap-1)];
      
      // fill top half of new deq with new refs
      for (int j = oldcap; j < newcap; ++j) newdeq[j] = new VolatileTaskRef();
      
      deq = newdeq;
      base = 0;
      top = newTop;
    }
  }


  /* ------------ Scheduling  ------------------- */


  /**
   * Do all but the pop() part of yield or join, by
   * traversing all DEQs in our group looking for a task to
   * steal. If none, it checks the entry queue. 
   * <p>
   * Since there are no good, portable alternatives,
   * we rely here on a mixture of Thread.yield and priorities
   * to reduce wasted spinning, even though these are
   * not well defined. We are hoping here that the JVM
   * does something sensible.
   * @param waitingFor if non-null, the current task being joined
   **/

  protected void scan(final FJTask waitingFor) {

    FJTask task = null;

    // to delay lowering priority until first failure to steal
    boolean lowered = false;
    
    /*
      Circularly traverse from a random start index. 
      
      This differs slightly from cilk version that uses a random index
      for each attempted steal.
      Exhaustive scanning might impede analytic tractablity of 
      the scheduling policy, but makes it much easier to deal with
      startup and shutdown.
    */
    
    FJTaskRunner[] ts = group.getArray();
    int idx = victimRNG.nextInt(ts.length);
    
    for (int i = 0; i < ts.length; ++i) {
      
      FJTaskRunner t = ts[idx];
      if (++idx >= ts.length) idx = 0; // circularly traverse
      
      if (t != null && t != this) {
        
        if (waitingFor != null && waitingFor.isDone()) {
          break;
        }
        else {
          if (COLLECT_STATS) ++scans;
          task = t.take();
          if (task != null) {
            if (COLLECT_STATS) ++steals;
            break;
          }
          else if (isInterrupted()) {
            break;
          }
          else if (!lowered) { // if this is first fail, lower priority
            lowered = true;
            setPriority(scanPriority);
          }
          else {           // otherwise we are at low priority; just yield
            yield();
          }
        }
      }
      
    } 

    if (task == null) {
      if (COLLECT_STATS) ++scans;
      task = group.pollEntryQueue();
      if (COLLECT_STATS) if (task != null) ++steals;
    }
    
    if (lowered) setPriority(runPriority);
    
    if (task != null && !task.isDone()) {
      if (COLLECT_STATS) ++runs;
      task.run(); 
      task.setDone(); 
    }

  }

  /**
   * Same as scan, but called when current thread is idling.
   * It repeatedly scans other threads for tasks,
   * sleeping while none are available. 
   * <p>
   * This differs from scan mainly in that
   * since there is no reason to return to recheck any
   * condition, we iterate until a task is found, backing
   * off via sleeps if necessary.
   **/

  protected void scanWhileIdling() {
    FJTask task = null;
    
    boolean lowered = false;
    long iters = 0;
    
    FJTaskRunner[] ts = group.getArray();
    int idx = victimRNG.nextInt(ts.length);
    
    do {
      for (int i = 0; i < ts.length; ++i) {
        
        FJTaskRunner t = ts[idx];
        if (++idx >= ts.length) idx = 0; // circularly traverse
        
        if (t != null && t != this) {
          if (COLLECT_STATS) ++scans;
          
          task = t.take();
          if (task != null) {
            if (COLLECT_STATS) ++steals;
            if (lowered) setPriority(runPriority);
            group.setActive(this);
            break;
          }
        }
      } 
      
      if (task == null) {
        if (isInterrupted()) 
          return;
        
        if (COLLECT_STATS) ++scans;
        task = group.pollEntryQueue();
        
        if (task != null) {
          if (COLLECT_STATS) ++steals;
          if (lowered) setPriority(runPriority);
          group.setActive(this);
        }
        else {
          ++iters;
          //  Check here for yield vs sleep to avoid entering group synch lock
          if (iters >= group.SCANS_PER_SLEEP) {
            group.checkActive(this, iters);
            if (isInterrupted())
              return;
          }
          else if (!lowered) {
            lowered = true;
            setPriority(scanPriority);
          }
          else {
            yield();
          }
        }
      }
    } while (task == null);


    if (!task.isDone()) {
      if (COLLECT_STATS) ++runs;
      task.run(); 
      task.setDone(); 
    }
    
  }

  /* ------------  composite operations ------------------- */

    
  /**
   * Main runloop
   **/

  public void run() {
    try{ 
      while (!interrupted()) {
        
        FJTask task = pop();
        if (task != null) {
          if (!task.isDone()) {
            // inline FJTask.invoke
            if (COLLECT_STATS) ++runs;
            task.run(); 
            task.setDone(); 
          }
        }
        else
          scanWhileIdling();
      }
    }
    finally {
      group.setInactive(this);
    }
  }

  /**
   * Execute a task in this thread. Generally called when current task
   * cannot otherwise continue.
   **/

    
  protected final void taskYield() {
    FJTask task = pop();
    if (task != null) {
      if (!task.isDone()) {
        if (COLLECT_STATS) ++runs;
        task.run(); 
        task.setDone(); 
      }
    }
    else
      scan(null);
  }


  /**
   * Process tasks until w is done.
   * Equivalent to <code>while(!w.isDone()) taskYield(); </code>
   **/

  protected final void taskJoin(final FJTask w) {

    while (!w.isDone()) { 

      FJTask task = pop();
      if (task != null) {
        if (!task.isDone()) {
          if (COLLECT_STATS) ++runs;
          task.run(); 
          task.setDone(); 
          if (task == w) return; // fast exit if we just ran w
        }
      }
      else
        scan(w);
    }
  }

  /**
   * A specialized expansion of
   * <code> w.fork(); invoke(v); w.join(); </code>
   **/


  protected final void coInvoke(final FJTask w, final FJTask v) {

    // inline  push

    int t = top;
    if (t < (base & (deq.length-1)) + deq.length) {

      deq[t & (deq.length-1)].put(w);
      top = t + 1;

      // inline  invoke

      if (!v.isDone()) { 
        if (COLLECT_STATS) ++runs; 
        v.run(); 
        v.setDone(); 
      }
      
      // inline  taskJoin
      
      while (!w.isDone()) {
        FJTask task  = pop();
        if (task != null) {
          if (!task.isDone()) {
            if (COLLECT_STATS) ++runs;
            task.run(); 
            task.setDone(); 
            if (task == w) return; // fast exit if we just ran w
          }
        }
        else
          scan(w);
      }
    }

    else      // handle non-inlinable cases
      slowCoInvoke(w, v);
  }


  /**
   * Backup to handle noninlinable cases of coInvoke
   **/

  protected void slowCoInvoke(final FJTask w, final FJTask v) {
    push(w); // let push deal with overflow
    FJTask.invoke(v);
    taskJoin(w);
  }


  /**
   * Array-based version of coInvoke
   **/

  protected final void coInvoke(FJTask[] tasks) {
    int nforks = tasks.length - 1;

    // inline bulk push of all but one task

    int t = top;

    if (nforks >= 0 && t + nforks < (base & (deq.length-1)) + deq.length) {
      for (int i = 0; i < nforks; ++i) {
        deq[t++ & (deq.length-1)].put(tasks[i]);
        top = t;
      }

      // inline invoke of one task
      FJTask v = tasks[nforks];
      if (!v.isDone()) { 
        if (COLLECT_STATS) ++runs; 
        v.run(); 
        v.setDone(); 
      }
      
      // inline  taskJoins
      
      for (int i = 0; i < nforks; ++i) { 
        FJTask w = tasks[i];
        while (!w.isDone()) {

          FJTask task = pop();
          if (task != null) {
            if (!task.isDone()) {
              if (COLLECT_STATS) ++runs;
              task.run(); 
              task.setDone(); 
            }
          }
          else
            scan(w);
        }
      }
    }

    else  // handle non-inlinable cases
      slowCoInvoke(tasks);
  }

  /**
   * Backup to handle atypical or noninlinable cases of coInvoke
   **/

  protected void slowCoInvoke(FJTask[] tasks) {
    for (int i = 0; i < tasks.length; ++i) push(tasks[i]);
    for (int i = 0; i < tasks.length; ++i) taskJoin(tasks[i]);
  }

}