threads_fork_join.cpp File Reference

#include "main.h"
#include "Eigen/ThreadPool"

Classes
struct	TestData

Macros
#define	EIGEN_USE_THREADS

Functions
TestData	make_test_data (int num_threads, int num_shards)
static void	test_unary_parallel_for (int granularity)
static void	test_binary_parallel_for (int granularity)
static void	test_async_parallel_for ()
	EIGEN_DECLARE_TEST (fork_join)

Macro Definition Documentation

EIGEN_USE_THREADS

#define EIGEN_USE_THREADS

Function Documentation

EIGEN_DECLARE_TEST()

EIGEN_DECLARE_TEST

(

fork_join

)

                              {
  CALL_SUBTEST(test_unary_parallel_for(1));
  CALL_SUBTEST(test_unary_parallel_for(2));
  CALL_SUBTEST(test_binary_parallel_for(1));
  CALL_SUBTEST(test_binary_parallel_for(2));
  CALL_SUBTEST(test_async_parallel_for());
}

References CALL_SUBTEST, test_async_parallel_for(), test_binary_parallel_for(), and test_unary_parallel_for().

make_test_data()

TestData make_test_data	(	int	num_threads,
		int	num_shards
	)

                                                         {
  return {ThreadPool(num_threads), std::vector<double>(num_shards, 1.0)};
}

Referenced by test_async_parallel_for(), test_binary_parallel_for(), and test_unary_parallel_for().

test_async_parallel_for()

static void test_async_parallel_for ( )

static

                                      {
  // Test correctness.
  // NOTE: Granularity and type of `do_func` are checked in the synchronous tests.
  const int kNumThreads = 4;
  const int kNumTasks = 100;
  const int kNumAsyncCalls = kNumThreads * 4;
  TestData test_data = make_test_data(kNumThreads, kNumTasks);
  std::atomic<double> sum = 0.0;
  std::function<void(int)> unary_do_fn = [&](int i) {
    for (double new_sum = sum; !sum.compare_exchange_weak(new_sum, new_sum + test_data.data[i]);) {
    };
  };
  Barrier barrier(kNumTasks * kNumAsyncCalls);
  std::function<void()> done = [&]() { barrier.Notify(); };
  for (int k = 0; k < kNumAsyncCalls; ++k) {
    test_data.tp.Schedule([&]() {
      ForkJoinScheduler::ParallelForAsync(0, kNumTasks, /*granularity=*/1, unary_do_fn, done, &test_data.tp);
    });
  }
  barrier.Wait();
  VERIFY_IS_EQUAL(sum, kNumTasks * kNumAsyncCalls);
}

References TestData::data, i, k, make_test_data(), Eigen::Barrier::Notify(), Eigen::ThreadPoolTempl< Environment >::Schedule(), TestData::tp, VERIFY_IS_EQUAL, and Eigen::Barrier::Wait().

Referenced by EIGEN_DECLARE_TEST().

test_binary_parallel_for()

static void test_binary_parallel_for ( int  granularity )

static

                                                      {
  // Test correctness.
  const int kNumTasks = 100000;
  TestData test_data = make_test_data(/*num_threads=*/4, kNumTasks);
  std::atomic<double> sum = 0.0;
  std::function<void(int, int)> binary_do_fn = [&](int i, int j) {
    for (int k = i; k < j; ++k)
      for (double new_sum = sum; !sum.compare_exchange_weak(new_sum, new_sum + test_data.data[k]);) {
      };
  };
  ForkJoinScheduler::ParallelFor(0, kNumTasks, granularity, std::move(binary_do_fn), &test_data.tp);
  VERIFY_IS_EQUAL(sum, kNumTasks);
}

References TestData::data, i, j, k, make_test_data(), TestData::tp, and VERIFY_IS_EQUAL.

Referenced by EIGEN_DECLARE_TEST().

test_unary_parallel_for()

static void test_unary_parallel_for ( int  granularity )

static

                                                     {
  // Test correctness.
  const int kNumTasks = 100000;
  TestData test_data = make_test_data(/*num_threads=*/4, kNumTasks);
  std::atomic<double> sum = 0.0;
  std::function<void(int)> unary_do_fn = [&](int i) {
    for (double new_sum = sum; !sum.compare_exchange_weak(new_sum, new_sum + test_data.data[i]);) {
    };
  };
  ForkJoinScheduler::ParallelFor(0, kNumTasks, granularity, std::move(unary_do_fn), &test_data.tp);
  VERIFY_IS_EQUAL(sum, kNumTasks);
}

References TestData::data, i, make_test_data(), TestData::tp, and VERIFY_IS_EQUAL.

Referenced by EIGEN_DECLARE_TEST().