Eigen::ThreadLocal< T, Initialize, Release > Class Template Reference

#include <ThreadLocal.h>

Classes
struct	ThreadIdAndValue

Public Member Functions
	ThreadLocal (int capacity)
	ThreadLocal (int capacity, Initialize initialize)
	ThreadLocal (int capacity, Initialize initialize, Release release)
T &	local ()
void	ForEach (std::function< void(std::thread::id, T &)> f)
	~ThreadLocal ()

Private Member Functions
T &	SpilledLocal (std::thread::id this_thread)

Private Attributes
Initialize	initialize_
Release	release_
const int	capacity_
MaxSizeVector< ThreadIdAndValue >	data_
MaxSizeVector< std::atomic< ThreadIdAndValue * > >	ptr_
std::atomic< int >	filled_records_
EIGEN_MUTEX	mu_
std::unordered_map< std::thread::id, T >	per_thread_map_

Constructor & Destructor Documentation

ThreadLocal() [1/3]

template<typename T , typename Initialize = internal::ThreadLocalNoOpInitialize<T>, typename Release = internal::ThreadLocalNoOpRelease<T>>

Eigen::ThreadLocal< T, Initialize, Release >::ThreadLocal ( int  capacity )

inlineexplicit

      : ThreadLocal(capacity, internal::ThreadLocalNoOpInitialize<T>(), internal::ThreadLocalNoOpRelease<T>()) {}

ThreadLocal() [2/3]

template<typename T , typename Initialize = internal::ThreadLocalNoOpInitialize<T>, typename Release = internal::ThreadLocalNoOpRelease<T>>

Eigen::ThreadLocal< T, Initialize, Release >::ThreadLocal ( int  capacity, Initialize  initialize  )

inline

      : ThreadLocal(capacity, std::move(initialize), internal::ThreadLocalNoOpRelease<T>()) {}

ThreadLocal() [3/3]

template<typename T , typename Initialize = internal::ThreadLocalNoOpInitialize<T>, typename Release = internal::ThreadLocalNoOpRelease<T>>

Eigen::ThreadLocal< T, Initialize, Release >::ThreadLocal ( int  capacity, Initialize  initialize, Release  release  )

inline

      : initialize_(std::move(initialize)),
        release_(std::move(release)),
        capacity_(capacity),
        data_(capacity_),
        ptr_(capacity_),
        filled_records_(0) {
    eigen_assert(capacity_ >= 0);
    data_.resize(capacity_);
    for (int i = 0; i < capacity_; ++i) {
      ptr_.emplace_back(nullptr);
    }
  }

References Eigen::ThreadLocal< T, Initialize, Release >::capacity_, Eigen::ThreadLocal< T, Initialize, Release >::data_, eigen_assert, i, and Eigen::ThreadLocal< T, Initialize, Release >::ptr_.

~ThreadLocal()

template<typename T , typename Initialize = internal::ThreadLocalNoOpInitialize<T>, typename Release = internal::ThreadLocalNoOpRelease<T>>

Eigen::ThreadLocal< T, Initialize, Release >::~ThreadLocal ( )

inline

                 {
    // Reading directly from `data_` is unsafe, because only CAS to the record
    // in `ptr_` makes all changes visible to other threads.
    for (auto& ptr : ptr_) {
      ThreadIdAndValue* record = ptr.load();
      if (record == nullptr) continue;
      release_(record->value);
    }
 
    // We did not spill into the map based storage.
    if (filled_records_.load(std::memory_order_relaxed) < capacity_) return;
 
    // Adds a happens before edge from the last call to SpilledLocal().
    EIGEN_MUTEX_LOCK lock(mu_);
    for (auto& kv : per_thread_map_) {
      release_(kv.second);
    }
  }

References Eigen::ThreadLocal< T, Initialize, Release >::capacity_, Eigen::ThreadLocal< T, Initialize, Release >::filled_records_, Eigen::ThreadLocal< T, Initialize, Release >::mu_, Eigen::ThreadLocal< T, Initialize, Release >::per_thread_map_, Eigen::ThreadLocal< T, Initialize, Release >::ptr_, Eigen::ThreadLocal< T, Initialize, Release >::release_, and Eigen::ThreadLocal< T, Initialize, Release >::ThreadIdAndValue::value.

Member Function Documentation

ForEach()

template<typename T , typename Initialize = internal::ThreadLocalNoOpInitialize<T>, typename Release = internal::ThreadLocalNoOpRelease<T>>

void Eigen::ThreadLocal< T, Initialize, Release >::ForEach ( std::function< void(std::thread::id, T &)>  f )

inline

                                                     {
    // Reading directly from `data_` is unsafe, because only CAS to the
    // record in `ptr_` makes all changes visible to other threads.
    for (auto& ptr : ptr_) {
      ThreadIdAndValue* record = ptr.load();
      if (record == nullptr) continue;
      f(record->thread_id, record->value);
    }
 
    // We did not spill into the map based storage.
    if (filled_records_.load(std::memory_order_relaxed) < capacity_) return;
 
    // Adds a happens before edge from the last call to SpilledLocal().
    EIGEN_MUTEX_LOCK lock(mu_);
    for (auto& kv : per_thread_map_) {
      f(kv.first, kv.second);
    }
  }

References Eigen::ThreadLocal< T, Initialize, Release >::capacity_, f(), Eigen::ThreadLocal< T, Initialize, Release >::filled_records_, Eigen::ThreadLocal< T, Initialize, Release >::mu_, Eigen::ThreadLocal< T, Initialize, Release >::per_thread_map_, Eigen::ThreadLocal< T, Initialize, Release >::ptr_, Eigen::ThreadLocal< T, Initialize, Release >::ThreadIdAndValue::thread_id, and Eigen::ThreadLocal< T, Initialize, Release >::ThreadIdAndValue::value.

Referenced by test_large_number_of_tasks_no_spill(), test_large_number_of_tasks_with_spill(), test_simple_thread_local(), and test_zero_sized_thread_local().

local()

template<typename T , typename Initialize = internal::ThreadLocalNoOpInitialize<T>, typename Release = internal::ThreadLocalNoOpRelease<T>>

T& Eigen::ThreadLocal< T, Initialize, Release >::local ( )

inline

             {
    std::thread::id this_thread = std::this_thread::get_id();
    if (capacity_ == 0) return SpilledLocal(this_thread);
 
    std::size_t h = std::hash<std::thread::id>()(this_thread);
    const int start_idx = h % capacity_;
 
    // NOTE: From the definition of `std::this_thread::get_id()` it is
    // guaranteed that we never can have concurrent insertions with the same key
    // to our hash-map like data structure. If we didn't find an element during
    // the initial traversal, it's guaranteed that no one else could have
    // inserted it while we are in this function. This allows to massively
    // simplify out lock-free insert-only hash map.
 
    // Check if we already have an element for `this_thread`.
    int idx = start_idx;
    while (ptr_[idx].load() != nullptr) {
      ThreadIdAndValue& record = *(ptr_[idx].load());
      if (record.thread_id == this_thread) return record.value;
 
      idx += 1;
      if (idx >= capacity_) idx -= capacity_;
      if (idx == start_idx) break;
    }
 
    // If we are here, it means that we found an insertion point in lookup
    // table at `idx`, or we did a full traversal and table is full.
 
    // If lock-free storage is full, fallback on mutex.
    if (filled_records_.load() >= capacity_) return SpilledLocal(this_thread);
 
    // We double check that we still have space to insert an element into a lock
    // free storage. If old value in `filled_records_` is larger than the
    // records capacity, it means that some other thread added an element while
    // we were traversing lookup table.
    int insertion_index = filled_records_.fetch_add(1, std::memory_order_relaxed);
    if (insertion_index >= capacity_) return SpilledLocal(this_thread);
 
    // At this point it's guaranteed that we can access to
    // data_[insertion_index_] without a data race.
    data_[insertion_index].thread_id = this_thread;
    initialize_(data_[insertion_index].value);
 
    // That's the pointer we'll put into the lookup table.
    ThreadIdAndValue* inserted = &data_[insertion_index];
 
    // We'll use nullptr pointer to ThreadIdAndValue in a compare-and-swap loop.
    ThreadIdAndValue* empty = nullptr;
 
    // Now we have to find an insertion point into the lookup table. We start
    // from the `idx` that was identified as an insertion point above, it's
    // guaranteed that we will have an empty record somewhere in a lookup table
    // (because we created a record in the `data_`).
    const int insertion_idx = idx;
 
    do {
      // Always start search from the original insertion candidate.
      idx = insertion_idx;
      while (ptr_[idx].load() != nullptr) {
        idx += 1;
        if (idx >= capacity_) idx -= capacity_;
        // If we did a full loop, it means that we don't have any free entries
        // in the lookup table, and this means that something is terribly wrong.
        eigen_assert(idx != insertion_idx);
      }
      // Atomic CAS of the pointer guarantees that any other thread, that will
      // follow this pointer will see all the mutations in the `data_`.
    } while (!ptr_[idx].compare_exchange_weak(empty, inserted));
 
    return inserted->value;
  }

References Eigen::ThreadLocal< T, Initialize, Release >::capacity_, Eigen::ThreadLocal< T, Initialize, Release >::data_, eigen_assert, Eigen::ThreadLocal< T, Initialize, Release >::filled_records_, Eigen::ThreadLocal< T, Initialize, Release >::initialize_, load(), Eigen::ThreadLocal< T, Initialize, Release >::ptr_, Eigen::ThreadLocal< T, Initialize, Release >::SpilledLocal(), Eigen::ThreadLocal< T, Initialize, Release >::ThreadIdAndValue::thread_id, Eigen::value, and Eigen::ThreadLocal< T, Initialize, Release >::ThreadIdAndValue::value.

Referenced by test_large_number_of_tasks_no_spill(), test_large_number_of_tasks_with_spill(), test_simple_thread_local(), and test_zero_sized_thread_local().

SpilledLocal()

template<typename T , typename Initialize = internal::ThreadLocalNoOpInitialize<T>, typename Release = internal::ThreadLocalNoOpRelease<T>>

T& Eigen::ThreadLocal< T, Initialize, Release >::SpilledLocal ( std::thread::id  this_thread )

inlineprivate

                                           {
    EIGEN_MUTEX_LOCK lock(mu_);
 
    auto it = per_thread_map_.find(this_thread);
    if (it == per_thread_map_.end()) {
      auto result = per_thread_map_.emplace(this_thread, T());
      eigen_assert(result.second);
      initialize_((*result.first).second);
      return (*result.first).second;
    } else {
      return it->second;
    }
  }

References eigen_assert, Eigen::ThreadLocal< T, Initialize, Release >::initialize_, Eigen::ThreadLocal< T, Initialize, Release >::mu_, and Eigen::ThreadLocal< T, Initialize, Release >::per_thread_map_.

Referenced by Eigen::ThreadLocal< T, Initialize, Release >::local().

Member Data Documentation

capacity_

template<typename T , typename Initialize = internal::ThreadLocalNoOpInitialize<T>, typename Release = internal::ThreadLocalNoOpRelease<T>>

const int Eigen::ThreadLocal< T, Initialize, Release >::capacity_

private

Referenced by Eigen::ThreadLocal< T, Initialize, Release >::ForEach(), Eigen::ThreadLocal< T, Initialize, Release >::local(), Eigen::ThreadLocal< T, Initialize, Release >::ThreadLocal(), and Eigen::ThreadLocal< T, Initialize, Release >::~ThreadLocal().

data_

template<typename T , typename Initialize = internal::ThreadLocalNoOpInitialize<T>, typename Release = internal::ThreadLocalNoOpRelease<T>>

MaxSizeVector<ThreadIdAndValue> Eigen::ThreadLocal< T, Initialize, Release >::data_

private

Referenced by Eigen::ThreadLocal< T, Initialize, Release >::local(), and Eigen::ThreadLocal< T, Initialize, Release >::ThreadLocal().

filled_records_

template<typename T , typename Initialize = internal::ThreadLocalNoOpInitialize<T>, typename Release = internal::ThreadLocalNoOpRelease<T>>

std::atomic<int> Eigen::ThreadLocal< T, Initialize, Release >::filled_records_

private

Referenced by Eigen::ThreadLocal< T, Initialize, Release >::ForEach(), Eigen::ThreadLocal< T, Initialize, Release >::local(), and Eigen::ThreadLocal< T, Initialize, Release >::~ThreadLocal().

initialize_

template<typename T , typename Initialize = internal::ThreadLocalNoOpInitialize<T>, typename Release = internal::ThreadLocalNoOpRelease<T>>

Initialize Eigen::ThreadLocal< T, Initialize, Release >::initialize_

private

Referenced by Eigen::ThreadLocal< T, Initialize, Release >::local(), and Eigen::ThreadLocal< T, Initialize, Release >::SpilledLocal().

mu_

template<typename T , typename Initialize = internal::ThreadLocalNoOpInitialize<T>, typename Release = internal::ThreadLocalNoOpRelease<T>>

EIGEN_MUTEX Eigen::ThreadLocal< T, Initialize, Release >::mu_

private

Referenced by Eigen::ThreadLocal< T, Initialize, Release >::ForEach(), Eigen::ThreadLocal< T, Initialize, Release >::SpilledLocal(), and Eigen::ThreadLocal< T, Initialize, Release >::~ThreadLocal().

per_thread_map_

template<typename T , typename Initialize = internal::ThreadLocalNoOpInitialize<T>, typename Release = internal::ThreadLocalNoOpRelease<T>>

std::unordered_map<std::thread::id, T> Eigen::ThreadLocal< T, Initialize, Release >::per_thread_map_

private

Referenced by Eigen::ThreadLocal< T, Initialize, Release >::ForEach(), Eigen::ThreadLocal< T, Initialize, Release >::SpilledLocal(), and Eigen::ThreadLocal< T, Initialize, Release >::~ThreadLocal().

ptr_

template<typename T , typename Initialize = internal::ThreadLocalNoOpInitialize<T>, typename Release = internal::ThreadLocalNoOpRelease<T>>

MaxSizeVector<std::atomic<ThreadIdAndValue*> > Eigen::ThreadLocal< T, Initialize, Release >::ptr_

private

Referenced by Eigen::ThreadLocal< T, Initialize, Release >::ForEach(), Eigen::ThreadLocal< T, Initialize, Release >::local(), Eigen::ThreadLocal< T, Initialize, Release >::ThreadLocal(), and Eigen::ThreadLocal< T, Initialize, Release >::~ThreadLocal().

release_

template<typename T , typename Initialize = internal::ThreadLocalNoOpInitialize<T>, typename Release = internal::ThreadLocalNoOpRelease<T>>

Release Eigen::ThreadLocal< T, Initialize, Release >::release_

private

Referenced by Eigen::ThreadLocal< T, Initialize, Release >::~ThreadLocal().

---

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

• ThreadLocal.h