484 }
485
486 template <class E, MEMFLAGS F, unsigned int N>
487 bool OverflowTaskQueue<E, F, N>::pop_overflow(E& t)
488 {
489 if (overflow_empty()) return false;
490 t = overflow_stack()->pop();
491 return true;
492 }
493
494 template <class E, MEMFLAGS F, unsigned int N>
495 bool OverflowTaskQueue<E, F, N>::try_push_to_taskqueue(E t) {
496 return taskqueue_t::push(t);
497 }
498 class TaskQueueSetSuper {
499 protected:
500 static int randomParkAndMiller(int* seed0);
501 public:
502 // Returns "true" if some TaskQueue in the set contains a task.
503 virtual bool peek() = 0;
504 };
505
506 template <MEMFLAGS F> class TaskQueueSetSuperImpl: public CHeapObj<F>, public TaskQueueSetSuper {
507 };
508
509 template<class T, MEMFLAGS F>
510 class GenericTaskQueueSet: public TaskQueueSetSuperImpl<F> {
511 private:
512 uint _n;
513 T** _queues;
514
515 public:
516 typedef typename T::element_type E;
517
518 GenericTaskQueueSet(int n) : _n(n) {
519 typedef T* GenericTaskQueuePtr;
520 _queues = NEW_C_HEAP_ARRAY(GenericTaskQueuePtr, n, F);
521 for (int i = 0; i < n; i++) {
522 _queues[i] = NULL;
523 }
524 }
525
526 bool steal_best_of_2(uint queue_num, int* seed, E& t);
527
528 void register_queue(uint i, T* q);
529
530 T* queue(uint n);
531
532 // The thread with queue number "queue_num" (and whose random number seed is
533 // at "seed") is trying to steal a task from some other queue. (It may try
534 // several queues, according to some configuration parameter.) If some steal
535 // succeeds, returns "true" and sets "t" to the stolen task, otherwise returns
536 // false.
537 bool steal(uint queue_num, int* seed, E& t);
538
539 bool peek();
540 };
541
542 template<class T, MEMFLAGS F> void
543 GenericTaskQueueSet<T, F>::register_queue(uint i, T* q) {
544 assert(i < _n, "index out of range.");
545 _queues[i] = q;
546 }
547
548 template<class T, MEMFLAGS F> T*
549 GenericTaskQueueSet<T, F>::queue(uint i) {
550 return _queues[i];
551 }
552
553 template<class T, MEMFLAGS F> bool
554 GenericTaskQueueSet<T, F>::steal(uint queue_num, int* seed, E& t) {
555 for (uint i = 0; i < 2 * _n; i++) {
556 if (steal_best_of_2(queue_num, seed, t)) {
557 TASKQUEUE_STATS_ONLY(queue(queue_num)->stats.record_steal(true));
558 return true;
559 }
577 } else if (_n == 2) {
578 // Just try the other one.
579 uint k = (queue_num + 1) % 2;
580 return _queues[k]->pop_global(t);
581 } else {
582 assert(_n == 1, "can't be zero.");
583 return false;
584 }
585 }
586
587 template<class T, MEMFLAGS F>
588 bool GenericTaskQueueSet<T, F>::peek() {
589 // Try all the queues.
590 for (uint j = 0; j < _n; j++) {
591 if (_queues[j]->peek())
592 return true;
593 }
594 return false;
595 }
596
597 // When to terminate from the termination protocol.
598 class TerminatorTerminator: public CHeapObj<mtInternal> {
599 public:
600 virtual bool should_exit_termination() = 0;
601 };
602
603 // A class to aid in the termination of a set of parallel tasks using
604 // TaskQueueSet's for work stealing.
605
606 #undef TRACESPINNING
607
608 class ParallelTaskTerminator: public StackObj {
609 private:
610 int _n_threads;
611 TaskQueueSetSuper* _queue_set;
612 char _pad_before[DEFAULT_CACHE_LINE_SIZE];
613 int _offered_termination;
614 char _pad_after[DEFAULT_CACHE_LINE_SIZE];
615
616 #ifdef TRACESPINNING
617 static uint _total_yields;
618 static uint _total_spins;
619 static uint _total_peeks;
620 #endif
621
622 bool peek_in_queue_set();
623 protected:
624 virtual void yield();
625 void sleep(uint millis);
626
627 public:
628
629 // "n_threads" is the number of threads to be terminated. "queue_set" is a
630 // queue sets of work queues of other threads.
631 ParallelTaskTerminator(int n_threads, TaskQueueSetSuper* queue_set);
632
633 // The current thread has no work, and is ready to terminate if everyone
634 // else is. If returns "true", all threads are terminated. If returns
635 // "false", available work has been observed in one of the task queues,
636 // so the global task is not complete.
637 bool offer_termination() {
638 return offer_termination(NULL);
639 }
640
641 // As above, but it also terminates if the should_exit_termination()
642 // method of the terminator parameter returns true. If terminator is
643 // NULL, then it is ignored.
644 bool offer_termination(TerminatorTerminator* terminator);
645
646 // Reset the terminator, so that it may be reused again.
647 // The caller is responsible for ensuring that this is done
648 // in an MT-safe manner, once the previous round of use of
649 // the terminator is finished.
650 void reset_for_reuse();
651 // Same as above but the number of parallel threads is set to the
652 // given number.
653 void reset_for_reuse(int n_threads);
654
655 #ifdef TRACESPINNING
656 static uint total_yields() { return _total_yields; }
657 static uint total_spins() { return _total_spins; }
|
484 }
485
486 template <class E, MEMFLAGS F, unsigned int N>
487 bool OverflowTaskQueue<E, F, N>::pop_overflow(E& t)
488 {
489 if (overflow_empty()) return false;
490 t = overflow_stack()->pop();
491 return true;
492 }
493
494 template <class E, MEMFLAGS F, unsigned int N>
495 bool OverflowTaskQueue<E, F, N>::try_push_to_taskqueue(E t) {
496 return taskqueue_t::push(t);
497 }
498 class TaskQueueSetSuper {
499 protected:
500 static int randomParkAndMiller(int* seed0);
501 public:
502 // Returns "true" if some TaskQueue in the set contains a task.
503 virtual bool peek() = 0;
504 virtual size_t tasks() = 0;
505 };
506
507 template <MEMFLAGS F> class TaskQueueSetSuperImpl: public CHeapObj<F>, public TaskQueueSetSuper {
508 };
509
510 template<class T, MEMFLAGS F>
511 class GenericTaskQueueSet: public TaskQueueSetSuperImpl<F> {
512 private:
513 uint _n;
514 T** _queues;
515
516 public:
517 typedef typename T::element_type E;
518
519 GenericTaskQueueSet(int n) : _n(n) {
520 typedef T* GenericTaskQueuePtr;
521 _queues = NEW_C_HEAP_ARRAY(GenericTaskQueuePtr, n, F);
522 for (int i = 0; i < n; i++) {
523 _queues[i] = NULL;
524 }
525 }
526
527 bool steal_best_of_2(uint queue_num, int* seed, E& t);
528
529 void register_queue(uint i, T* q);
530
531 T* queue(uint n);
532
533 // The thread with queue number "queue_num" (and whose random number seed is
534 // at "seed") is trying to steal a task from some other queue. (It may try
535 // several queues, according to some configuration parameter.) If some steal
536 // succeeds, returns "true" and sets "t" to the stolen task, otherwise returns
537 // false.
538 bool steal(uint queue_num, int* seed, E& t);
539
540 bool peek();
541 size_t tasks();
542
543 uint size() const { return _n; }
544 };
545
546 template<class T, MEMFLAGS F> void
547 GenericTaskQueueSet<T, F>::register_queue(uint i, T* q) {
548 assert(i < _n, "index out of range.");
549 _queues[i] = q;
550 }
551
552 template<class T, MEMFLAGS F> T*
553 GenericTaskQueueSet<T, F>::queue(uint i) {
554 return _queues[i];
555 }
556
557 template<class T, MEMFLAGS F> bool
558 GenericTaskQueueSet<T, F>::steal(uint queue_num, int* seed, E& t) {
559 for (uint i = 0; i < 2 * _n; i++) {
560 if (steal_best_of_2(queue_num, seed, t)) {
561 TASKQUEUE_STATS_ONLY(queue(queue_num)->stats.record_steal(true));
562 return true;
563 }
581 } else if (_n == 2) {
582 // Just try the other one.
583 uint k = (queue_num + 1) % 2;
584 return _queues[k]->pop_global(t);
585 } else {
586 assert(_n == 1, "can't be zero.");
587 return false;
588 }
589 }
590
591 template<class T, MEMFLAGS F>
592 bool GenericTaskQueueSet<T, F>::peek() {
593 // Try all the queues.
594 for (uint j = 0; j < _n; j++) {
595 if (_queues[j]->peek())
596 return true;
597 }
598 return false;
599 }
600
601 template<class T, MEMFLAGS F>
602 size_t GenericTaskQueueSet<T, F>::tasks() {
603 size_t n = 0;
604 for (uint j = 0; j < _n; j++) {
605 n += _queues[j]->size();
606 }
607 return n;
608 }
609
610 // When to terminate from the termination protocol.
611 class TerminatorTerminator: public CHeapObj<mtInternal> {
612 public:
613 virtual bool should_exit_termination() = 0;
614 };
615
616 // A class to aid in the termination of a set of parallel tasks using
617 // TaskQueueSet's for work stealing.
618
619 #undef TRACESPINNING
620
621 class ParallelTaskTerminator: public StackObj {
622 protected:
623 int _n_threads;
624 TaskQueueSetSuper* _queue_set;
625 char _pad_before[DEFAULT_CACHE_LINE_SIZE];
626 int _offered_termination;
627 char _pad_after[DEFAULT_CACHE_LINE_SIZE];
628
629 #ifdef TRACESPINNING
630 static uint _total_yields;
631 static uint _total_spins;
632 static uint _total_peeks;
633 #endif
634
635 bool peek_in_queue_set();
636 protected:
637 virtual void yield();
638 void sleep(uint millis);
639
640 public:
641
642 // "n_threads" is the number of threads to be terminated. "queue_set" is a
643 // queue sets of work queues of other threads.
644 ParallelTaskTerminator(int n_threads, TaskQueueSetSuper* queue_set);
645
646 // The current thread has no work, and is ready to terminate if everyone
647 // else is. If returns "true", all threads are terminated. If returns
648 // "false", available work has been observed in one of the task queues,
649 // so the global task is not complete.
650 virtual bool offer_termination() {
651 return offer_termination(NULL);
652 }
653
654 // As above, but it also terminates if the should_exit_termination()
655 // method of the terminator parameter returns true. If terminator is
656 // NULL, then it is ignored.
657 bool offer_termination(TerminatorTerminator* terminator);
658
659 // Reset the terminator, so that it may be reused again.
660 // The caller is responsible for ensuring that this is done
661 // in an MT-safe manner, once the previous round of use of
662 // the terminator is finished.
663 void reset_for_reuse();
664 // Same as above but the number of parallel threads is set to the
665 // given number.
666 void reset_for_reuse(int n_threads);
667
668 #ifdef TRACESPINNING
669 static uint total_yields() { return _total_yields; }
670 static uint total_spins() { return _total_spins; }
|