Eigen::ThreadPoolTempl< Environment > Class Template Reference

#include <NonBlockingThreadPool.h>

Inheritance diagram for Eigen::ThreadPoolTempl< Environment >:

Classes
struct	PerThread
struct	SpinningState
struct	ThreadData

Public Types
typedef Environment::EnvThread	Thread
typedef Environment::Task	Task
typedef RunQueue< Task, 1024 >	Queue

Public Member Functions
	ThreadPoolTempl (int num_threads, Environment env=Environment())
	ThreadPoolTempl (int num_threads, bool allow_spinning, Environment env=Environment())
	~ThreadPoolTempl ()
void	SetStealPartitions (const std::vector< std::pair< unsigned, unsigned >> &partitions)
void	Schedule (std::function< void()> fn) EIGEN_OVERRIDE
void	ScheduleWithHint (std::function< void()> fn, int start, int limit) override
void	MaybeGetTask (Task *t)
void	Cancel () EIGEN_OVERRIDE
int	NumThreads () const EIGEN_FINAL
int	CurrentThreadId () const EIGEN_FINAL
Public Member Functions inherited from Eigen::ThreadPoolInterface
virtual	~ThreadPoolInterface ()

Private Member Functions
unsigned	EncodePartition (unsigned start, unsigned limit)
void	DecodePartition (unsigned val, unsigned *start, unsigned *limit)
void	AssertBounds (int start, int end)
void	SetStealPartition (size_t i, unsigned val)
unsigned	GetStealPartition (int i)
void	ComputeCoprimes (int N, MaxSizeVector< unsigned > *coprimes)
unsigned	NumBlockedThreads () const
unsigned	NumActiveThreads () const
void	WorkerLoop (int thread_id)
Task	Steal (unsigned start, unsigned limit)
Task	LocalSteal ()
Task	GlobalSteal ()
bool	WaitForWork (EventCount::Waiter *waiter, Task *t)
int	NonEmptyQueueIndex ()
bool	StartSpinning ()
bool	StopSpinning ()
bool	IsNotifyParkedThreadRequired ()
EIGEN_STRONG_INLINE PerThread *	GetPerThread ()

Static Private Member Functions
static EIGEN_STRONG_INLINE uint64_t	GlobalThreadIdHash ()
static EIGEN_STRONG_INLINE unsigned	Rand (uint64_t *state)

Private Attributes
Environment	env_
const int	num_threads_
const bool	allow_spinning_
const int	spin_count_
MaxSizeVector< ThreadData >	thread_data_
MaxSizeVector< MaxSizeVector< unsigned > >	all_coprimes_
MaxSizeVector< EventCount::Waiter >	waiters_
unsigned	global_steal_partition_
std::atomic< uint64_t >	spinning_state_
std::atomic< unsigned >	blocked_
std::atomic< bool >	done_
std::atomic< bool >	cancelled_
EventCount	ec_
std::unique_ptr< Barrier >	init_barrier_
EIGEN_MUTEX	per_thread_map_mutex_
std::unordered_map< uint64_t, std::unique_ptr< PerThread > >	per_thread_map_

Static Private Attributes
static constexpr int	kMaxPartitionBits = 16
static constexpr int	kMaxThreads = 1 << kMaxPartitionBits
static constexpr int	kMaxSpinningThreads = 1
static constexpr int	kSpinCount = 5000
static constexpr int	kMinActiveThreadsToStartSpinning = 4

Member Typedef Documentation

Queue

template<typename Environment >

typedef RunQueue<Task, 1024> Eigen::ThreadPoolTempl< Environment >::Queue

Task

template<typename Environment >

typedef Environment::Task Eigen::ThreadPoolTempl< Environment >::Task

Thread

template<typename Environment >

typedef Environment::EnvThread Eigen::ThreadPoolTempl< Environment >::Thread

Constructor & Destructor Documentation

ThreadPoolTempl() [1/2]

template<typename Environment >

Eigen::ThreadPoolTempl< Environment >::ThreadPoolTempl ( int  num_threads, Environment  env = Environment()  )

inline

: ThreadPoolTempl(num_threads, true, env) {}

ThreadPoolTempl() [2/2]

template<typename Environment >

Eigen::ThreadPoolTempl< Environment >::ThreadPoolTempl ( int  num_threads, bool  allow_spinning, Environment  env = Environment()  )

inline

      : env_(env),
        num_threads_(num_threads),
        allow_spinning_(allow_spinning),
        spin_count_(
            // TODO(dvyukov,rmlarsen): The time spent in NonEmptyQueueIndex() is proportional to num_threads_ and
            // we assume that new work is scheduled at a constant rate, so we divide `kSpintCount` by number of
            // threads and number of spinning threads. The constant was picked based on a fair dice roll, tune it.
            allow_spinning && num_threads > 0 ? kSpinCount / kMaxSpinningThreads / num_threads : 0),
        thread_data_(num_threads),
        all_coprimes_(num_threads),
        waiters_(num_threads),
        global_steal_partition_(EncodePartition(0, num_threads_)),
        spinning_state_(0),
        blocked_(0),
        done_(false),
        cancelled_(false),
        ec_(waiters_) {
    waiters_.resize(num_threads_);
    // Calculate coprimes of all numbers [1, num_threads].
    // Coprimes are used for random walks over all threads in Steal
    // and NonEmptyQueueIndex. Iteration is based on the fact that if we take
    // a random starting thread index t and calculate num_threads - 1 subsequent
    // indices as (t + coprime) % num_threads, we will cover all threads without
    // repetitions (effectively getting a presudo-random permutation of thread
    // indices).
    eigen_plain_assert(num_threads_ < kMaxThreads);
    for (int i = 1; i <= num_threads_; ++i) {
      all_coprimes_.emplace_back(i);
      ComputeCoprimes(i, &all_coprimes_.back());
    }
#ifndef EIGEN_THREAD_LOCAL
    init_barrier_.reset(new Barrier(num_threads_));
#endif
    thread_data_.resize(num_threads_);
    for (int i = 0; i < num_threads_; i++) {
      SetStealPartition(i, EncodePartition(0, num_threads_));
      thread_data_[i].thread.reset(env_.CreateThread([this, i]() { WorkerLoop(i); }));
    }
#ifndef EIGEN_THREAD_LOCAL
    // Wait for workers to initialize per_thread_map_. Otherwise we might race
    // with them in Schedule or CurrentThreadId.
    init_barrier_->Wait();
#endif
  }

References Eigen::ThreadPoolTempl< Environment >::all_coprimes_, Eigen::ThreadPoolTempl< Environment >::ComputeCoprimes(), eigen_plain_assert, Eigen::ThreadPoolTempl< Environment >::EncodePartition(), Eigen::ThreadPoolTempl< Environment >::env_, i, Eigen::ThreadPoolTempl< Environment >::init_barrier_, Eigen::ThreadPoolTempl< Environment >::kMaxThreads, Eigen::ThreadPoolTempl< Environment >::num_threads_, Eigen::ThreadPoolTempl< Environment >::SetStealPartition(), Eigen::ThreadPoolTempl< Environment >::thread_data_, and Eigen::ThreadPoolTempl< Environment >::waiters_.

~ThreadPoolTempl()

template<typename Environment >

Eigen::ThreadPoolTempl< Environment >::~ThreadPoolTempl ( )

inline

                     {
    done_ = true;
 
    // Now if all threads block without work, they will start exiting.
    // But note that threads can continue to work arbitrary long,
    // block, submit new work, unblock and otherwise live full life.
    if (!cancelled_) {
      ec_.Notify(true);
    } else {
      // Since we were cancelled, there might be entries in the queues.
      // Empty them to prevent their destructor from asserting.
      for (size_t i = 0; i < thread_data_.size(); i++) {
        thread_data_[i].queue.Flush();
      }
    }
    // Join threads explicitly (by destroying) to avoid destruction order within
    // this class.
    for (size_t i = 0; i < thread_data_.size(); ++i) thread_data_[i].thread.reset();
  }

References Eigen::ThreadPoolTempl< Environment >::cancelled_, Eigen::ThreadPoolTempl< Environment >::done_, Eigen::ThreadPoolTempl< Environment >::ec_, i, Eigen::EventCount::Notify(), and Eigen::ThreadPoolTempl< Environment >::thread_data_.

Member Function Documentation

AssertBounds()

template<typename Environment >

void Eigen::ThreadPoolTempl< Environment >::AssertBounds ( int  start, int  end  )

inlineprivate

                                        {
    eigen_plain_assert(start >= 0);
    eigen_plain_assert(start < end);  // non-zero sized partition
    eigen_plain_assert(end <= num_threads_);
  }

References eigen_plain_assert, Eigen::placeholders::end, Eigen::ThreadPoolTempl< Environment >::num_threads_, and oomph::CumulativeTimings::start().

Referenced by Eigen::ThreadPoolTempl< Environment >::LocalSteal(), and Eigen::ThreadPoolTempl< Environment >::SetStealPartitions().

Cancel()

template<typename Environment >

void Eigen::ThreadPoolTempl< Environment >::Cancel ( )

inlinevirtual

Reimplemented from Eigen::ThreadPoolInterface.

                               {
    cancelled_ = true;
    done_ = true;
 
    // Let each thread know it's been cancelled.
#ifdef EIGEN_THREAD_ENV_SUPPORTS_CANCELLATION
    for (size_t i = 0; i < thread_data_.size(); i++) {
      thread_data_[i].thread->OnCancel();
    }
#endif
 
    // Wake up the threads without work to let them exit on their own.
    ec_.Notify(true);
  }

References Eigen::ThreadPoolTempl< Environment >::cancelled_, Eigen::ThreadPoolTempl< Environment >::done_, Eigen::ThreadPoolTempl< Environment >::ec_, i, Eigen::EventCount::Notify(), and Eigen::ThreadPoolTempl< Environment >::thread_data_.

Referenced by test_cancel().

ComputeCoprimes()

template<typename Environment >

void Eigen::ThreadPoolTempl< Environment >::ComputeCoprimes ( int  N, MaxSizeVector< unsigned > *  coprimes  )

inlineprivate

                                                                 {
    for (int i = 1; i <= N; i++) {
      unsigned a = i;
      unsigned b = N;
      // If GCD(a, b) == 1, then a and b are coprimes.
      while (b != 0) {
        unsigned tmp = a;
        a = b;
        b = tmp % b;
      }
      if (a == 1) {
        coprimes->push_back(i);
      }
    }
  }

References a, b, i, N, Eigen::MaxSizeVector< T >::push_back(), and tmp.

Referenced by Eigen::ThreadPoolTempl< Environment >::ThreadPoolTempl().

CurrentThreadId()

template<typename Environment >

int Eigen::ThreadPoolTempl< Environment >::CurrentThreadId ( ) const

inlinevirtual

Implements Eigen::ThreadPoolInterface.

                                          {
    const PerThread* pt = const_cast<ThreadPoolTempl*>(this)->GetPerThread();
    if (pt->pool == this) {
      return pt->thread_id;
    } else {
      return -1;
    }
  }

References Eigen::ThreadPoolTempl< Environment >::GetPerThread(), Eigen::ThreadPoolTempl< Environment >::PerThread::pool, and Eigen::ThreadPoolTempl< Environment >::PerThread::thread_id.

Referenced by test_parallelism(), and test_pool_partitions().

DecodePartition()

template<typename Environment >

void Eigen::ThreadPoolTempl< Environment >::DecodePartition ( unsigned  val, unsigned *  start, unsigned *  limit  )

inlineprivate

                                                                              {
    *limit = val & (kMaxThreads - 1);
    val >>= kMaxPartitionBits;
    *start = val;
  }

References Eigen::ThreadPoolTempl< Environment >::kMaxPartitionBits, Eigen::ThreadPoolTempl< Environment >::kMaxThreads, oomph::CumulativeTimings::start(), and calibrate::val.

Referenced by Eigen::ThreadPoolTempl< Environment >::LocalSteal().

EncodePartition()

template<typename Environment >

unsigned Eigen::ThreadPoolTempl< Environment >::EncodePartition ( unsigned  start, unsigned  limit  )

inlineprivate

{ return (start << kMaxPartitionBits) | limit; }

References Eigen::ThreadPoolTempl< Environment >::kMaxPartitionBits, and oomph::CumulativeTimings::start().

Referenced by Eigen::ThreadPoolTempl< Environment >::SetStealPartitions(), and Eigen::ThreadPoolTempl< Environment >::ThreadPoolTempl().

GetPerThread()

template<typename Environment >

EIGEN_STRONG_INLINE PerThread* Eigen::ThreadPoolTempl< Environment >::GetPerThread ( )

inlineprivate

                                                {
#ifndef EIGEN_THREAD_LOCAL
    static PerThread dummy;
    auto it = per_thread_map_.find(GlobalThreadIdHash());
    if (it == per_thread_map_.end()) {
      return &dummy;
    } else {
      return it->second.get();
    }
#else
    EIGEN_THREAD_LOCAL PerThread per_thread_;
    PerThread* pt = &per_thread_;
    return pt;
#endif
  }

References Eigen::ThreadPoolTempl< Environment >::GlobalThreadIdHash(), and Eigen::ThreadPoolTempl< Environment >::per_thread_map_.

Referenced by Eigen::ThreadPoolTempl< Environment >::CurrentThreadId(), Eigen::ThreadPoolTempl< Environment >::LocalSteal(), Eigen::ThreadPoolTempl< Environment >::MaybeGetTask(), Eigen::ThreadPoolTempl< Environment >::NonEmptyQueueIndex(), Eigen::ThreadPoolTempl< Environment >::ScheduleWithHint(), Eigen::ThreadPoolTempl< Environment >::Steal(), and Eigen::ThreadPoolTempl< Environment >::WorkerLoop().

GetStealPartition()

template<typename Environment >

unsigned Eigen::ThreadPoolTempl< Environment >::GetStealPartition ( int  i )

inlineprivate

{ return thread_data_[i].steal_partition.load(std::memory_order_relaxed); }

References i, and Eigen::ThreadPoolTempl< Environment >::thread_data_.

Referenced by Eigen::ThreadPoolTempl< Environment >::LocalSteal().

GlobalSteal()

template<typename Environment >

Task Eigen::ThreadPoolTempl< Environment >::GlobalSteal ( )

inlineprivate

{ return Steal(0, num_threads_); }

References Eigen::ThreadPoolTempl< Environment >::num_threads_, and Eigen::ThreadPoolTempl< Environment >::Steal().

Referenced by Eigen::ThreadPoolTempl< Environment >::MaybeGetTask().

GlobalThreadIdHash()

template<typename Environment >

static EIGEN_STRONG_INLINE uint64_t Eigen::ThreadPoolTempl< Environment >::GlobalThreadIdHash ( )

inlinestaticprivate

                                                           {
    return std::hash<std::thread::id>()(std::this_thread::get_id());
  }

Referenced by Eigen::ThreadPoolTempl< Environment >::GetPerThread(), and Eigen::ThreadPoolTempl< Environment >::WorkerLoop().

IsNotifyParkedThreadRequired()

template<typename Environment >

bool Eigen::ThreadPoolTempl< Environment >::IsNotifyParkedThreadRequired ( )

inlineprivate

                                      {
    uint64_t spinning = spinning_state_.load(std::memory_order_relaxed);
    for (;;) {
      SpinningState state = SpinningState::Decode(spinning);
 
      // If the number of tasks submitted without notifying parked threads is
      // equal to the number of spinning threads, we must wake up one of the
      // parked threads.
      if (state.num_no_notification == state.num_spinning) return true;
 
      // Increment the number of tasks submitted without notification.
      ++state.num_no_notification;
 
      if (spinning_state_.compare_exchange_weak(spinning, state.Encode(), std::memory_order_relaxed)) {
        return false;
      }
    }
  }

References Eigen::ThreadPoolTempl< Environment >::SpinningState::Decode(), Eigen::ThreadPoolTempl< Environment >::SpinningState::Encode(), Eigen::ThreadPoolTempl< Environment >::SpinningState::num_no_notification, Eigen::ThreadPoolTempl< Environment >::SpinningState::num_spinning, and Eigen::ThreadPoolTempl< Environment >::spinning_state_.

Referenced by Eigen::ThreadPoolTempl< Environment >::ScheduleWithHint().

LocalSteal()

template<typename Environment >

Task Eigen::ThreadPoolTempl< Environment >::LocalSteal ( )

inlineprivate

                    {
    PerThread* pt = GetPerThread();
    unsigned partition = GetStealPartition(pt->thread_id);
    // If thread steal partition is the same as global partition, there is no
    // need to go through the steal loop twice.
    if (global_steal_partition_ == partition) return Task();
    unsigned start, limit;
    DecodePartition(partition, &start, &limit);
    AssertBounds(start, limit);
 
    return Steal(start, limit);
  }

References Eigen::ThreadPoolTempl< Environment >::AssertBounds(), Eigen::ThreadPoolTempl< Environment >::DecodePartition(), Eigen::ThreadPoolTempl< Environment >::GetPerThread(), Eigen::ThreadPoolTempl< Environment >::GetStealPartition(), Eigen::ThreadPoolTempl< Environment >::global_steal_partition_, oomph::CumulativeTimings::start(), Eigen::ThreadPoolTempl< Environment >::Steal(), and Eigen::ThreadPoolTempl< Environment >::PerThread::thread_id.

Referenced by Eigen::ThreadPoolTempl< Environment >::MaybeGetTask().

MaybeGetTask()

template<typename Environment >

void Eigen::ThreadPoolTempl< Environment >::MaybeGetTask ( Task *  t )

inline

                             {
    PerThread* pt = GetPerThread();
    Queue& q = thread_data_[pt->thread_id].queue;
    *t = q.PopFront();
    if (t->f) return;
    if (num_threads_ == 1) {
      // For num_threads_ == 1 there is no point in going through the expensive
      // steal loop. Moreover, since NonEmptyQueueIndex() calls PopBack() on the
      // victim queues it might reverse the order in which ops are executed
      // compared to the order in which they are scheduled, which tends to be
      // counter-productive for the types of I/O workloads single thread pools
      // tend to be used for.
      for (int i = 0; i < spin_count_ && !t->f; ++i) *t = q.PopFront();
    } else {
      if (EIGEN_PREDICT_FALSE(!t->f)) *t = LocalSteal();
      if (EIGEN_PREDICT_FALSE(!t->f)) *t = GlobalSteal();
      if (EIGEN_PREDICT_FALSE(!t->f)) {
        if (allow_spinning_ && StartSpinning()) {
          for (int i = 0; i < spin_count_ && !t->f; ++i) *t = GlobalSteal();
          // Notify `spinning_state_` that we are no longer spinning.
          bool has_no_notify_task = StopSpinning();
          // If a task was submitted to the queue without a call to
          // `ec_.Notify()` (if `IsNotifyParkedThreadRequired()` returned
          // false), and we didn't steal anything above, we must try to
          // steal one more time, to make sure that this task will be
          // executed. We will not necessarily find it, because it might
          // have been already stolen by some other thread.
          if (has_no_notify_task && !t->f) *t = q.PopFront();
        }
      }
    }
  }

References Eigen::ThreadPoolTempl< Environment >::allow_spinning_, EIGEN_PREDICT_FALSE, Eigen::ThreadPoolTempl< Environment >::GetPerThread(), Eigen::ThreadPoolTempl< Environment >::GlobalSteal(), i, Eigen::ThreadPoolTempl< Environment >::LocalSteal(), Eigen::ThreadPoolTempl< Environment >::num_threads_, Eigen::numext::q, Eigen::ThreadPoolTempl< Environment >::spin_count_, Eigen::ThreadPoolTempl< Environment >::StartSpinning(), Eigen::ThreadPoolTempl< Environment >::StopSpinning(), plotPSD::t, Eigen::ThreadPoolTempl< Environment >::thread_data_, and Eigen::ThreadPoolTempl< Environment >::PerThread::thread_id.

Referenced by Eigen::ForkJoinScheduler::ForkJoin(), and Eigen::ThreadPoolTempl< Environment >::WorkerLoop().

NonEmptyQueueIndex()

template<typename Environment >

int Eigen::ThreadPoolTempl< Environment >::NonEmptyQueueIndex ( )

inlineprivate

                           {
    PerThread* pt = GetPerThread();
    // We intentionally design NonEmptyQueueIndex to steal work from
    // anywhere in the queue so threads don't block in WaitForWork() forever
    // when all threads in their partition go to sleep. Steal is still local.
    const size_t size = thread_data_.size();
    unsigned r = Rand(&pt->rand);
    unsigned inc = all_coprimes_[size - 1][r % all_coprimes_[size - 1].size()];
    unsigned victim = r % size;
    for (unsigned i = 0; i < size; i++) {
      if (!thread_data_[victim].queue.Empty()) {
        return victim;
      }
      victim += inc;
      if (victim >= size) {
        victim -= static_cast<unsigned int>(size);
      }
    }
    return -1;
  }

References Eigen::ThreadPoolTempl< Environment >::all_coprimes_, Eigen::ThreadPoolTempl< Environment >::GetPerThread(), i, UniformPSDSelfTest::r, Eigen::ThreadPoolTempl< Environment >::PerThread::rand, Eigen::ThreadPoolTempl< Environment >::Rand(), size, and Eigen::ThreadPoolTempl< Environment >::thread_data_.

Referenced by Eigen::ThreadPoolTempl< Environment >::WaitForWork().

NumActiveThreads()

template<typename Environment >

unsigned Eigen::ThreadPoolTempl< Environment >::NumActiveThreads ( ) const

inlineprivate

{ return num_threads_ - blocked_.load(); }

References Eigen::ThreadPoolTempl< Environment >::blocked_, and Eigen::ThreadPoolTempl< Environment >::num_threads_.

Referenced by Eigen::ThreadPoolTempl< Environment >::StartSpinning().

NumBlockedThreads()

template<typename Environment >

unsigned Eigen::ThreadPoolTempl< Environment >::NumBlockedThreads ( ) const

inlineprivate

{ return blocked_.load(); }

References Eigen::ThreadPoolTempl< Environment >::blocked_.

NumThreads()

template<typename Environment >

int Eigen::ThreadPoolTempl< Environment >::NumThreads ( ) const

inlinevirtual

Implements Eigen::ThreadPoolInterface.

{ return num_threads_; }

References Eigen::ThreadPoolTempl< Environment >::num_threads_.

Referenced by Eigen::CoreThreadPoolDevice::calculateLevels(), and test_parallelism().

Rand()

template<typename Environment >

static EIGEN_STRONG_INLINE unsigned Eigen::ThreadPoolTempl< Environment >::Rand ( uint64_t *  state )

inlinestaticprivate

                                                            {
    uint64_t current = *state;
    // Update the internal state
    *state = current * 6364136223846793005ULL + 0xda3e39cb94b95bdbULL;
    // Generate the random output (using the PCG-XSH-RS scheme)
    return static_cast<unsigned>((current ^ (current >> 22)) >> (22 + (current >> 61)));
  }

Referenced by Eigen::ThreadPoolTempl< Environment >::NonEmptyQueueIndex(), Eigen::ThreadPoolTempl< Environment >::ScheduleWithHint(), and Eigen::ThreadPoolTempl< Environment >::Steal().

Schedule()

template<typename Environment >

void Eigen::ThreadPoolTempl< Environment >::Schedule ( std::function< void()>  fn )

inlinevirtual

Implements Eigen::ThreadPoolInterface.

{ ScheduleWithHint(std::move(fn), 0, num_threads_); }

References Eigen::ThreadPoolTempl< Environment >::num_threads_, and Eigen::ThreadPoolTempl< Environment >::ScheduleWithHint().

Referenced by Eigen::ForkJoinScheduler::ForkJoin(), Eigen::ForkJoinScheduler::ParallelFor(), Eigen::CoreThreadPoolDevice::parallelForImpl(), test_async_parallel_for(), test_cancel(), test_large_number_of_tasks_no_spill(), test_large_number_of_tasks_with_spill(), test_notification_multiple(), test_notification_single(), test_parallelism(), test_pool_partitions(), and test_simple_thread_local().

ScheduleWithHint()

template<typename Environment >

void Eigen::ThreadPoolTempl< Environment >::ScheduleWithHint ( std::function< void()>  fn, int  start, int  limit  )

inlineoverridevirtual

Reimplemented from Eigen::ThreadPoolInterface.

                                                                               {
    Task t = env_.CreateTask(std::move(fn));
    PerThread* pt = GetPerThread();
    if (pt->pool == this) {
      // Worker thread of this pool, push onto the thread's queue.
      Queue& q = thread_data_[pt->thread_id].queue;
      t = q.PushFront(std::move(t));
    } else {
      // A free-standing thread (or worker of another pool), push onto a random
      // queue.
      eigen_plain_assert(start < limit);
      eigen_plain_assert(limit <= num_threads_);
      int num_queues = limit - start;
      int rnd = Rand(&pt->rand) % num_queues;
      eigen_plain_assert(start + rnd < limit);
      Queue& q = thread_data_[start + rnd].queue;
      t = q.PushBack(std::move(t));
    }
    // Note: below we touch this after making w available to worker threads.
    // Strictly speaking, this can lead to a racy-use-after-free. Consider that
    // Schedule is called from a thread that is neither main thread nor a worker
    // thread of this pool. Then, execution of w directly or indirectly
    // completes overall computations, which in turn leads to destruction of
    // this. We expect that such scenario is prevented by program, that is,
    // this is kept alive while any threads can potentially be in Schedule.
    if (!t.f) {
      if (IsNotifyParkedThreadRequired()) {
        ec_.Notify(false);
      }
    } else {
      env_.ExecuteTask(t);  // Push failed, execute directly.
    }
  }

References Eigen::ThreadPoolTempl< Environment >::ec_, eigen_plain_assert, Eigen::ThreadPoolTempl< Environment >::env_, Eigen::ThreadPoolTempl< Environment >::GetPerThread(), Eigen::ThreadPoolTempl< Environment >::IsNotifyParkedThreadRequired(), Eigen::EventCount::Notify(), Eigen::ThreadPoolTempl< Environment >::num_threads_, Eigen::ThreadPoolTempl< Environment >::PerThread::pool, Eigen::numext::q, Eigen::ThreadPoolTempl< Environment >::PerThread::rand, Eigen::ThreadPoolTempl< Environment >::Rand(), oomph::CumulativeTimings::start(), plotPSD::t, Eigen::ThreadPoolTempl< Environment >::thread_data_, and Eigen::ThreadPoolTempl< Environment >::PerThread::thread_id.

Referenced by Eigen::ThreadPoolTempl< Environment >::Schedule(), and test_pool_partitions().

SetStealPartition()

template<typename Environment >

void Eigen::ThreadPoolTempl< Environment >::SetStealPartition ( size_t  i, unsigned  val  )

inlineprivate

                                                        {
    thread_data_[i].steal_partition.store(val, std::memory_order_relaxed);
  }

References i, Eigen::ThreadPoolTempl< Environment >::thread_data_, and calibrate::val.

Referenced by Eigen::ThreadPoolTempl< Environment >::SetStealPartitions(), and Eigen::ThreadPoolTempl< Environment >::ThreadPoolTempl().

SetStealPartitions()

template<typename Environment >

void Eigen::ThreadPoolTempl< Environment >::SetStealPartitions ( const std::vector< std::pair< unsigned, unsigned >> &  partitions )

inline

                                                                                    {
    eigen_plain_assert(partitions.size() == static_cast<std::size_t>(num_threads_));
 
    // Pass this information to each thread queue.
    for (int i = 0; i < num_threads_; i++) {
      const auto& pair = partitions[i];
      unsigned start = pair.first, end = pair.second;
      AssertBounds(start, end);
      unsigned val = EncodePartition(start, end);
      SetStealPartition(i, val);
    }
  }

References Eigen::ThreadPoolTempl< Environment >::AssertBounds(), eigen_plain_assert, Eigen::ThreadPoolTempl< Environment >::EncodePartition(), Eigen::placeholders::end, i, Eigen::ThreadPoolTempl< Environment >::num_threads_, Eigen::ThreadPoolTempl< Environment >::SetStealPartition(), oomph::CumulativeTimings::start(), and calibrate::val.

Referenced by test_pool_partitions().

StartSpinning()

template<typename Environment >

bool Eigen::ThreadPoolTempl< Environment >::StartSpinning ( )

inlineprivate

                       {
    if (NumActiveThreads() > kMinActiveThreadsToStartSpinning) return false;
 
    uint64_t spinning = spinning_state_.load(std::memory_order_relaxed);
    for (;;) {
      SpinningState state = SpinningState::Decode(spinning);
 
      if ((state.num_spinning - state.num_no_notification) >= kMaxSpinningThreads) {
        return false;
      }
 
      // Increment the number of spinning threads.
      ++state.num_spinning;
 
      if (spinning_state_.compare_exchange_weak(spinning, state.Encode(), std::memory_order_relaxed)) {
        return true;
      }
    }
  }

References Eigen::ThreadPoolTempl< Environment >::SpinningState::Decode(), Eigen::ThreadPoolTempl< Environment >::SpinningState::Encode(), Eigen::ThreadPoolTempl< Environment >::kMaxSpinningThreads, Eigen::ThreadPoolTempl< Environment >::kMinActiveThreadsToStartSpinning, Eigen::ThreadPoolTempl< Environment >::SpinningState::num_no_notification, Eigen::ThreadPoolTempl< Environment >::SpinningState::num_spinning, Eigen::ThreadPoolTempl< Environment >::NumActiveThreads(), and Eigen::ThreadPoolTempl< Environment >::spinning_state_.

Referenced by Eigen::ThreadPoolTempl< Environment >::MaybeGetTask().

Steal()

template<typename Environment >

Task Eigen::ThreadPoolTempl< Environment >::Steal ( unsigned  start, unsigned  limit  )

inlineprivate

                                             {
    PerThread* pt = GetPerThread();
    const size_t size = limit - start;
    unsigned r = Rand(&pt->rand);
    // Reduce r into [0, size) range, this utilizes trick from
    // https://lemire.me/blog/2016/06/27/a-fast-alternative-to-the-modulo-reduction/
    eigen_plain_assert(all_coprimes_[size - 1].size() < (1 << 30));
    unsigned victim = ((uint64_t)r * (uint64_t)size) >> 32;
    unsigned index = ((uint64_t)all_coprimes_[size - 1].size() * (uint64_t)r) >> 32;
    unsigned inc = all_coprimes_[size - 1][index];
 
    for (unsigned i = 0; i < size; i++) {
      eigen_plain_assert(start + victim < limit);
      Task t = thread_data_[start + victim].queue.PopBack();
      if (t.f) {
        return t;
      }
      victim += inc;
      if (victim >= size) {
        victim -= static_cast<unsigned int>(size);
      }
    }
    return Task();
  }

References Eigen::ThreadPoolTempl< Environment >::all_coprimes_, eigen_plain_assert, Eigen::ThreadPoolTempl< Environment >::GetPerThread(), i, UniformPSDSelfTest::r, Eigen::ThreadPoolTempl< Environment >::PerThread::rand, Eigen::ThreadPoolTempl< Environment >::Rand(), size, oomph::CumulativeTimings::start(), plotPSD::t, and Eigen::ThreadPoolTempl< Environment >::thread_data_.

Referenced by Eigen::ThreadPoolTempl< Environment >::GlobalSteal(), and Eigen::ThreadPoolTempl< Environment >::LocalSteal().

StopSpinning()

template<typename Environment >

bool Eigen::ThreadPoolTempl< Environment >::StopSpinning ( )

inlineprivate

                      {
    uint64_t spinning = spinning_state_.load(std::memory_order_relaxed);
    for (;;) {
      SpinningState state = SpinningState::Decode(spinning);
 
      // Decrement the number of spinning threads.
      --state.num_spinning;
 
      // Maybe decrement the number of tasks submitted without notification.
      bool has_no_notify_task = state.num_no_notification > 0;
      if (has_no_notify_task) --state.num_no_notification;
 
      if (spinning_state_.compare_exchange_weak(spinning, state.Encode(), std::memory_order_relaxed)) {
        return has_no_notify_task;
      }
    }
  }

References Eigen::ThreadPoolTempl< Environment >::SpinningState::Decode(), Eigen::ThreadPoolTempl< Environment >::SpinningState::Encode(), Eigen::ThreadPoolTempl< Environment >::SpinningState::num_no_notification, Eigen::ThreadPoolTempl< Environment >::SpinningState::num_spinning, and Eigen::ThreadPoolTempl< Environment >::spinning_state_.

Referenced by Eigen::ThreadPoolTempl< Environment >::MaybeGetTask().

WaitForWork()

template<typename Environment >

bool Eigen::ThreadPoolTempl< Environment >::WaitForWork ( EventCount::Waiter *  waiter, Task *  t  )

inlineprivate

                                                      {
    eigen_plain_assert(!t->f);
    // We already did best-effort emptiness check in Steal, so prepare for
    // blocking.
    ec_.Prewait();
    // Now do a reliable emptiness check.
    int victim = NonEmptyQueueIndex();
    if (victim != -1) {
      ec_.CancelWait();
      if (cancelled_) {
        return false;
      } else {
        *t = thread_data_[victim].queue.PopBack();
        return true;
      }
    }
    // Number of blocked threads is used as termination condition.
    // If we are shutting down and all worker threads blocked without work,
    // that's we are done.
    blocked_++;
    // TODO is blocked_ required to be unsigned?
    if (done_ && blocked_ == static_cast<unsigned>(num_threads_)) {
      ec_.CancelWait();
      // Almost done, but need to re-check queues.
      // Consider that all queues are empty and all worker threads are preempted
      // right after incrementing blocked_ above. Now a free-standing thread
      // submits work and calls destructor (which sets done_). If we don't
      // re-check queues, we will exit leaving the work unexecuted.
      if (NonEmptyQueueIndex() != -1) {
        // Note: we must not pop from queues before we decrement blocked_,
        // otherwise the following scenario is possible. Consider that instead
        // of checking for emptiness we popped the only element from queues.
        // Now other worker threads can start exiting, which is bad if the
        // work item submits other work. So we just check emptiness here,
        // which ensures that all worker threads exit at the same time.
        blocked_--;
        return true;
      }
      // Reached stable termination state.
      ec_.Notify(true);
      return false;
    }
    ec_.CommitWait(waiter);
    blocked_--;
    return true;
  }

References Eigen::ThreadPoolTempl< Environment >::blocked_, Eigen::ThreadPoolTempl< Environment >::cancelled_, Eigen::EventCount::CancelWait(), Eigen::EventCount::CommitWait(), Eigen::ThreadPoolTempl< Environment >::done_, Eigen::ThreadPoolTempl< Environment >::ec_, eigen_plain_assert, Eigen::ThreadPoolTempl< Environment >::NonEmptyQueueIndex(), Eigen::EventCount::Notify(), Eigen::ThreadPoolTempl< Environment >::num_threads_, Eigen::EventCount::Prewait(), plotPSD::t, and Eigen::ThreadPoolTempl< Environment >::thread_data_.

Referenced by Eigen::ThreadPoolTempl< Environment >::WorkerLoop().

WorkerLoop()

template<typename Environment >

void Eigen::ThreadPoolTempl< Environment >::WorkerLoop ( int  thread_id )

inlineprivate

                                 {
#ifndef EIGEN_THREAD_LOCAL
    std::unique_ptr<PerThread> new_pt(new PerThread());
    per_thread_map_mutex_.lock();
    bool insertOK = per_thread_map_.emplace(GlobalThreadIdHash(), std::move(new_pt)).second;
    eigen_plain_assert(insertOK);
    EIGEN_UNUSED_VARIABLE(insertOK);
    per_thread_map_mutex_.unlock();
    init_barrier_->Notify();
    init_barrier_->Wait();
#endif
    PerThread* pt = GetPerThread();
    pt->pool = this;
    pt->rand = GlobalThreadIdHash();
    pt->thread_id = thread_id;
    Task t;
    while (!cancelled_.load(std::memory_order_relaxed)) {
      MaybeGetTask(&t);
      // If we still don't have a task, wait for one. Return if thread pool is
      // in cancelled state.
      if (EIGEN_PREDICT_FALSE(!t.f)) {
        EventCount::Waiter* waiter = &waiters_[pt->thread_id];
        if (!WaitForWork(waiter, &t)) return;
      }
      if (EIGEN_PREDICT_TRUE(t.f)) env_.ExecuteTask(t);
    }
  }

References Eigen::ThreadPoolTempl< Environment >::cancelled_, eigen_plain_assert, EIGEN_PREDICT_FALSE, EIGEN_PREDICT_TRUE, EIGEN_UNUSED_VARIABLE, Eigen::ThreadPoolTempl< Environment >::env_, Eigen::ThreadPoolTempl< Environment >::GetPerThread(), Eigen::ThreadPoolTempl< Environment >::GlobalThreadIdHash(), Eigen::ThreadPoolTempl< Environment >::init_barrier_, Eigen::ThreadPoolTempl< Environment >::MaybeGetTask(), Eigen::ThreadPoolTempl< Environment >::per_thread_map_, Eigen::ThreadPoolTempl< Environment >::per_thread_map_mutex_, Eigen::ThreadPoolTempl< Environment >::PerThread::pool, Eigen::ThreadPoolTempl< Environment >::PerThread::rand, plotPSD::t, Eigen::ThreadPoolTempl< Environment >::PerThread::thread_id, Eigen::ThreadPoolTempl< Environment >::waiters_, and Eigen::ThreadPoolTempl< Environment >::WaitForWork().

Member Data Documentation

all_coprimes_

template<typename Environment >

MaxSizeVector<MaxSizeVector<unsigned> > Eigen::ThreadPoolTempl< Environment >::all_coprimes_

private

Referenced by Eigen::ThreadPoolTempl< Environment >::NonEmptyQueueIndex(), Eigen::ThreadPoolTempl< Environment >::Steal(), and Eigen::ThreadPoolTempl< Environment >::ThreadPoolTempl().

allow_spinning_

template<typename Environment >

const bool Eigen::ThreadPoolTempl< Environment >::allow_spinning_

private

Referenced by Eigen::ThreadPoolTempl< Environment >::MaybeGetTask().

blocked_

template<typename Environment >

std::atomic<unsigned> Eigen::ThreadPoolTempl< Environment >::blocked_

private

Referenced by Eigen::ThreadPoolTempl< Environment >::NumActiveThreads(), Eigen::ThreadPoolTempl< Environment >::NumBlockedThreads(), and Eigen::ThreadPoolTempl< Environment >::WaitForWork().

cancelled_

template<typename Environment >

std::atomic<bool> Eigen::ThreadPoolTempl< Environment >::cancelled_

private

Referenced by Eigen::ThreadPoolTempl< Environment >::Cancel(), Eigen::ThreadPoolTempl< Environment >::WaitForWork(), Eigen::ThreadPoolTempl< Environment >::WorkerLoop(), and Eigen::ThreadPoolTempl< Environment >::~ThreadPoolTempl().

done_

template<typename Environment >

std::atomic<bool> Eigen::ThreadPoolTempl< Environment >::done_

private

Referenced by Eigen::ThreadPoolTempl< Environment >::Cancel(), Eigen::ThreadPoolTempl< Environment >::WaitForWork(), and Eigen::ThreadPoolTempl< Environment >::~ThreadPoolTempl().

ec_

template<typename Environment >

EventCount Eigen::ThreadPoolTempl< Environment >::ec_

private

Referenced by Eigen::ThreadPoolTempl< Environment >::Cancel(), Eigen::ThreadPoolTempl< Environment >::ScheduleWithHint(), Eigen::ThreadPoolTempl< Environment >::WaitForWork(), and Eigen::ThreadPoolTempl< Environment >::~ThreadPoolTempl().

env_

template<typename Environment >

Environment Eigen::ThreadPoolTempl< Environment >::env_

private

Referenced by Eigen::ThreadPoolTempl< Environment >::ScheduleWithHint(), Eigen::ThreadPoolTempl< Environment >::ThreadPoolTempl(), and Eigen::ThreadPoolTempl< Environment >::WorkerLoop().

global_steal_partition_

template<typename Environment >

unsigned Eigen::ThreadPoolTempl< Environment >::global_steal_partition_

private

Referenced by Eigen::ThreadPoolTempl< Environment >::LocalSteal().

init_barrier_

template<typename Environment >

std::unique_ptr<Barrier> Eigen::ThreadPoolTempl< Environment >::init_barrier_

private

Referenced by Eigen::ThreadPoolTempl< Environment >::ThreadPoolTempl(), and Eigen::ThreadPoolTempl< Environment >::WorkerLoop().

kMaxPartitionBits

template<typename Environment >

constexpr int Eigen::ThreadPoolTempl< Environment >::kMaxPartitionBits = 16

staticconstexprprivate

Referenced by Eigen::ThreadPoolTempl< Environment >::DecodePartition(), and Eigen::ThreadPoolTempl< Environment >::EncodePartition().

kMaxSpinningThreads

template<typename Environment >

constexpr int Eigen::ThreadPoolTempl< Environment >::kMaxSpinningThreads = 1

staticconstexprprivate

Referenced by Eigen::ThreadPoolTempl< Environment >::StartSpinning().

kMaxThreads

template<typename Environment >

constexpr int Eigen::ThreadPoolTempl< Environment >::kMaxThreads = 1 << kMaxPartitionBits

staticconstexprprivate

Referenced by Eigen::ThreadPoolTempl< Environment >::DecodePartition(), and Eigen::ThreadPoolTempl< Environment >::ThreadPoolTempl().

kMinActiveThreadsToStartSpinning

template<typename Environment >

constexpr int Eigen::ThreadPoolTempl< Environment >::kMinActiveThreadsToStartSpinning = 4

staticconstexprprivate

Referenced by Eigen::ThreadPoolTempl< Environment >::StartSpinning().

kSpinCount

template<typename Environment >

constexpr int Eigen::ThreadPoolTempl< Environment >::kSpinCount = 5000

staticconstexprprivate

num_threads_

template<typename Environment >

const int Eigen::ThreadPoolTempl< Environment >::num_threads_

private

Referenced by Eigen::ThreadPoolTempl< Environment >::AssertBounds(), Eigen::ThreadPoolTempl< Environment >::GlobalSteal(), Eigen::ThreadPoolTempl< Environment >::MaybeGetTask(), Eigen::ThreadPoolTempl< Environment >::NumActiveThreads(), Eigen::ThreadPoolTempl< Environment >::NumThreads(), Eigen::ThreadPoolTempl< Environment >::Schedule(), Eigen::ThreadPoolTempl< Environment >::ScheduleWithHint(), Eigen::ThreadPoolTempl< Environment >::SetStealPartitions(), Eigen::ThreadPoolTempl< Environment >::ThreadPoolTempl(), and Eigen::ThreadPoolTempl< Environment >::WaitForWork().

per_thread_map_

template<typename Environment >

std::unordered_map<uint64_t, std::unique_ptr<PerThread> > Eigen::ThreadPoolTempl< Environment >::per_thread_map_

private

Referenced by Eigen::ThreadPoolTempl< Environment >::GetPerThread(), and Eigen::ThreadPoolTempl< Environment >::WorkerLoop().

per_thread_map_mutex_

template<typename Environment >

EIGEN_MUTEX Eigen::ThreadPoolTempl< Environment >::per_thread_map_mutex_

private

Referenced by Eigen::ThreadPoolTempl< Environment >::WorkerLoop().

spin_count_

template<typename Environment >

const int Eigen::ThreadPoolTempl< Environment >::spin_count_

private

Referenced by Eigen::ThreadPoolTempl< Environment >::MaybeGetTask().

spinning_state_

template<typename Environment >

std::atomic<uint64_t> Eigen::ThreadPoolTempl< Environment >::spinning_state_

private

Referenced by Eigen::ThreadPoolTempl< Environment >::IsNotifyParkedThreadRequired(), Eigen::ThreadPoolTempl< Environment >::StartSpinning(), and Eigen::ThreadPoolTempl< Environment >::StopSpinning().

thread_data_

template<typename Environment >

MaxSizeVector<ThreadData> Eigen::ThreadPoolTempl< Environment >::thread_data_

private

Referenced by Eigen::ThreadPoolTempl< Environment >::Cancel(), Eigen::ThreadPoolTempl< Environment >::GetStealPartition(), Eigen::ThreadPoolTempl< Environment >::MaybeGetTask(), Eigen::ThreadPoolTempl< Environment >::NonEmptyQueueIndex(), Eigen::ThreadPoolTempl< Environment >::ScheduleWithHint(), Eigen::ThreadPoolTempl< Environment >::SetStealPartition(), Eigen::ThreadPoolTempl< Environment >::Steal(), Eigen::ThreadPoolTempl< Environment >::ThreadPoolTempl(), Eigen::ThreadPoolTempl< Environment >::WaitForWork(), and Eigen::ThreadPoolTempl< Environment >::~ThreadPoolTempl().

waiters_

template<typename Environment >

MaxSizeVector<EventCount::Waiter> Eigen::ThreadPoolTempl< Environment >::waiters_

private

Referenced by Eigen::ThreadPoolTempl< Environment >::ThreadPoolTempl(), and Eigen::ThreadPoolTempl< Environment >::WorkerLoop().

---

The documentation for this class was generated from the following file:

• NonBlockingThreadPool.h