cxx11_tensor_thread_local.cpp File Reference

#include <iostream>
#include <unordered_set>
#include "main.h"
#include <Eigen/ThreadPool>

Classes
struct	Counter
struct	InitCounter

Macros
#define	EIGEN_USE_THREADS

Functions
void	test_simple_thread_local ()
void	test_zero_sized_thread_local ()
void	test_large_number_of_tasks_no_spill ()
void	test_large_number_of_tasks_with_spill ()
	EIGEN_DECLARE_TEST (cxx11_tensor_thread_local)

Macro Definition Documentation

EIGEN_USE_THREADS

#define EIGEN_USE_THREADS

Function Documentation

EIGEN_DECLARE_TEST()

EIGEN_DECLARE_TEST

(

cxx11_tensor_thread_local

)

                                              {
  CALL_SUBTEST(test_simple_thread_local());
  CALL_SUBTEST(test_zero_sized_thread_local());
  CALL_SUBTEST(test_large_number_of_tasks_no_spill());
  CALL_SUBTEST(test_large_number_of_tasks_with_spill());
}

References CALL_SUBTEST, test_large_number_of_tasks_no_spill(), test_large_number_of_tasks_with_spill(), test_simple_thread_local(), and test_zero_sized_thread_local().

test_large_number_of_tasks_no_spill()

void test_large_number_of_tasks_no_spill

(

)

                                           {
  int num_threads = internal::random<int>(4, 32);
  Eigen::ThreadPool thread_pool(num_threads);
  Eigen::ThreadLocal<Counter, InitCounter> counter(num_threads, InitCounter());
 
  int num_tasks = 10000;
  Eigen::Barrier barrier(num_tasks);
 
  for (int i = 0; i < num_tasks; ++i) {
    thread_pool.Schedule([&counter, &barrier]() {
      Counter& local = counter.local();
      local.inc();
      barrier.Notify();
    });
  }
 
  barrier.Wait();
 
  int total = 0;
  std::unordered_set<std::thread::id> unique_threads;
 
  counter.ForEach([&](std::thread::id id, Counter& cnt) {
    total += cnt.value();
    unique_threads.insert(id);
  });
 
  VERIFY_IS_EQUAL(total, num_tasks);
  // Not all threads in a pool might be woken up to execute submitted tasks.
  // Also thread_pool.Schedule() might use current thread if queue is full.
  VERIFY_IS_EQUAL(unique_threads.size() <= (static_cast<size_t>(num_threads + 1)), true);
}

References Eigen::ThreadLocal< T, Initialize, Release >::ForEach(), i, Eigen::ThreadLocal< T, Initialize, Release >::local(), Eigen::Barrier::Notify(), Eigen::ThreadPoolTempl< Environment >::Schedule(), Counter::value(), VERIFY_IS_EQUAL, and Eigen::Barrier::Wait().

Referenced by EIGEN_DECLARE_TEST().

test_large_number_of_tasks_with_spill()

void test_large_number_of_tasks_with_spill

(

)

                                             {
  int num_threads = internal::random<int>(4, 32);
  Eigen::ThreadPool thread_pool(num_threads);
  Eigen::ThreadLocal<Counter, InitCounter> counter(1, InitCounter());
 
  int num_tasks = 10000;
  Eigen::Barrier barrier(num_tasks);
 
  for (int i = 0; i < num_tasks; ++i) {
    thread_pool.Schedule([&counter, &barrier]() {
      Counter& local = counter.local();
      local.inc();
      barrier.Notify();
    });
  }
 
  barrier.Wait();
 
  int total = 0;
  std::unordered_set<std::thread::id> unique_threads;
 
  counter.ForEach([&](std::thread::id id, Counter& cnt) {
    total += cnt.value();
    unique_threads.insert(id);
  });
 
  VERIFY_IS_EQUAL(total, num_tasks);
  // Not all threads in a pool might be woken up to execute submitted tasks.
  // Also thread_pool.Schedule() might use current thread if queue is full.
  VERIFY_IS_EQUAL(unique_threads.size() <= (static_cast<size_t>(num_threads + 1)), true);
}

References Eigen::ThreadLocal< T, Initialize, Release >::ForEach(), i, Eigen::ThreadLocal< T, Initialize, Release >::local(), Eigen::Barrier::Notify(), Eigen::ThreadPoolTempl< Environment >::Schedule(), Counter::value(), VERIFY_IS_EQUAL, and Eigen::Barrier::Wait().

Referenced by EIGEN_DECLARE_TEST().

test_simple_thread_local()

void test_simple_thread_local

(

)

                                {
  int num_threads = internal::random<int>(4, 32);
  Eigen::ThreadPool thread_pool(num_threads);
  Eigen::ThreadLocal<Counter, InitCounter> counter(num_threads, InitCounter());
 
  int num_tasks = 3 * num_threads;
  Eigen::Barrier barrier(num_tasks);
 
  for (int i = 0; i < num_tasks; ++i) {
    thread_pool.Schedule([&counter, &barrier]() {
      Counter& local = counter.local();
      local.inc();
 
      std::this_thread::sleep_for(std::chrono::milliseconds(100));
      barrier.Notify();
    });
  }
 
  barrier.Wait();
 
  counter.ForEach([](std::thread::id, Counter& cnt) { VERIFY_IS_EQUAL(cnt.value(), 3); });
}

References Eigen::ThreadLocal< T, Initialize, Release >::ForEach(), i, Eigen::ThreadLocal< T, Initialize, Release >::local(), Eigen::Barrier::Notify(), Eigen::ThreadPoolTempl< Environment >::Schedule(), Counter::value(), VERIFY_IS_EQUAL, and Eigen::Barrier::Wait().

Referenced by EIGEN_DECLARE_TEST().

test_zero_sized_thread_local()

void test_zero_sized_thread_local

(

)

                                    {
  Eigen::ThreadLocal<Counter, InitCounter> counter(0, InitCounter());
 
  Counter& local = counter.local();
  local.inc();
 
  int total = 0;
  counter.ForEach([&total](std::thread::id, Counter& cnt) {
    total += cnt.value();
    VERIFY_IS_EQUAL(cnt.value(), 1);
  });
 
  VERIFY_IS_EQUAL(total, 1);
}

References Eigen::ThreadLocal< T, Initialize, Release >::ForEach(), Eigen::ThreadLocal< T, Initialize, Release >::local(), Counter::value(), and VERIFY_IS_EQUAL.

Referenced by EIGEN_DECLARE_TEST().