From 070e9f82d160782e116b0a1632eea7299abcd4df Mon Sep 17 00:00:00 2001 From: Sandeep Joshi Date: Fri, 23 Sep 2016 13:27:32 +0530 Subject: [PATCH 1/2] Add BoundThreadFactory which binds each new thread to a specified CPU core --- wangle/concurrent/BoundThreadFactory.h | 48 +++++++++++++++++++ .../test/ThreadPoolExecutorTest.cpp | 23 +++++++++ 2 files changed, 71 insertions(+) create mode 100644 wangle/concurrent/BoundThreadFactory.h diff --git a/wangle/concurrent/BoundThreadFactory.h b/wangle/concurrent/BoundThreadFactory.h new file mode 100644 index 000000000..464c9c5cb --- /dev/null +++ b/wangle/concurrent/BoundThreadFactory.h @@ -0,0 +1,48 @@ +/* + * Copyright (c) 2016, Facebook, Inc. + * All rights reserved. + * + * This source code is licensed under the BSD-style license found in the + * LICENSE file in the root directory of this source tree. An additional grant + * of patent rights can be found in the PATENTS file in the same directory. + * + */ + +#pragma once + +#include + +namespace wangle { + +/** + * A ThreadFactory that sets binds each thread to a specific CPU core. + * The main use case for this class is NUMA-aware computing. + */ +class BoundThreadFactory : public ThreadFactory { + public: + explicit BoundThreadFactory(std::shared_ptr factory, + int coreId) + : factory_(std::move(factory)) + , coreId_(coreId) {} + + std::thread newThread(folly::Func&& func) override { + int coreId = coreId_; + return factory_->newThread([ coreId, func = std::move(func) ]() mutable { + cpu_set_t cpuSet; + CPU_ZERO(&cpuSet); + CPU_SET(coreId, &cpuSet); + int error = pthread_setaffinity_np(pthread_self(), sizeof(cpuSet), &cpuSet); + if (error != 0) { + LOG(ERROR) << "set cpu affinity failed for core=" << coreId + << " with error " << error, strerror(error); + } + func(); + }); + } + + private: + std::shared_ptr factory_; + int coreId_; +}; + +} // wangle diff --git a/wangle/concurrent/test/ThreadPoolExecutorTest.cpp b/wangle/concurrent/test/ThreadPoolExecutorTest.cpp index 5ec368768..fef926310 100644 --- a/wangle/concurrent/test/ThreadPoolExecutorTest.cpp +++ b/wangle/concurrent/test/ThreadPoolExecutorTest.cpp @@ -15,6 +15,7 @@ #include #include #include +#include #include #include #include @@ -425,6 +426,28 @@ TEST(PriorityThreadFactoryTest, ThreadPriority) { EXPECT_EQ(1, actualPriority); } +TEST(BoundThreadFactoryTest, ThreadPriority) { + int expectedCoreId = 1; + BoundThreadFactory factory( + std::make_shared("stuff"), expectedCoreId); + factory.newThread([&]() { + cpu_set_t cpuSet; + CPU_ZERO(&cpuSet); + int error = pthread_getaffinity_np(pthread_self(), sizeof(cpuSet), &cpuSet); + ASSERT_EQ(error, 0); + + for (int c = 0; c < CPU_SETSIZE; c++) { + if (c == expectedCoreId) { + ASSERT_TRUE(CPU_ISSET(c, &cpuSet)); + } else { + ASSERT_FALSE(CPU_ISSET(c, &cpuSet)); + } + } + + }).join(); +} + + class TestData : public folly::RequestData { public: explicit TestData(int data) : data_(data) {} From e26e68f138638bc9e3d002da0669653b6e37b6f0 Mon Sep 17 00:00:00 2001 From: Sandeep Joshi Date: Fri, 23 Sep 2016 13:27:32 +0530 Subject: [PATCH 2/2] Add BoundThreadFactory which binds each new thread to a specified CPU core Added test to verify thread gets bound to core --- wangle/concurrent/BoundThreadFactory.h | 48 +++++++++++++++++++ .../test/ThreadPoolExecutorTest.cpp | 23 +++++++++ 2 files changed, 71 insertions(+) create mode 100644 wangle/concurrent/BoundThreadFactory.h diff --git a/wangle/concurrent/BoundThreadFactory.h b/wangle/concurrent/BoundThreadFactory.h new file mode 100644 index 000000000..3c62e2dc5 --- /dev/null +++ b/wangle/concurrent/BoundThreadFactory.h @@ -0,0 +1,48 @@ +/* + * Copyright (c) 2016, Facebook, Inc. + * All rights reserved. + * + * This source code is licensed under the BSD-style license found in the + * LICENSE file in the root directory of this source tree. An additional grant + * of patent rights can be found in the PATENTS file in the same directory. + * + */ + +#pragma once + +#include + +namespace wangle { + +/** + * A ThreadFactory that sets binds each thread to a specific CPU core. + * The main use case for this class is NUMA-aware computing. + */ +class BoundThreadFactory : public ThreadFactory { + public: + explicit BoundThreadFactory(std::shared_ptr factory, + int32_t coreId) + : factory_(std::move(factory)) + , coreId_(coreId) {} + + std::thread newThread(folly::Func&& func) override { + int32_t coreId = coreId_; + return factory_->newThread([ coreId, func = std::move(func) ]() mutable { + cpu_set_t cpuSet; + CPU_ZERO(&cpuSet); + CPU_SET(coreId, &cpuSet); + int32_t error = pthread_setaffinity_np(pthread_self(), sizeof(cpuSet), &cpuSet); + if (error != 0) { + LOG(ERROR) << "set cpu affinity failed for core=" << coreId + << " with error " << error, strerror(error); + } + func(); + }); + } + + private: + std::shared_ptr factory_; + int32_t coreId_; +}; + +} // wangle diff --git a/wangle/concurrent/test/ThreadPoolExecutorTest.cpp b/wangle/concurrent/test/ThreadPoolExecutorTest.cpp index 5ec368768..19f2d386a 100644 --- a/wangle/concurrent/test/ThreadPoolExecutorTest.cpp +++ b/wangle/concurrent/test/ThreadPoolExecutorTest.cpp @@ -15,6 +15,7 @@ #include #include #include +#include #include #include #include @@ -425,6 +426,28 @@ TEST(PriorityThreadFactoryTest, ThreadPriority) { EXPECT_EQ(1, actualPriority); } +TEST(BoundThreadFactoryTest, ThreadPriority) { + int32_t expectedCoreId = 1; // use sysconf(_SC_NPROCESSORS_ONLN) or hwloc lib + BoundThreadFactory factory( + std::make_shared("stuff"), expectedCoreId); + factory.newThread([&]() { + cpu_set_t cpuSet; + CPU_ZERO(&cpuSet); + int error = pthread_getaffinity_np(pthread_self(), sizeof(cpuSet), &cpuSet); + ASSERT_EQ(error, 0); + + for (int32_t c = 0; c < CPU_SETSIZE; c++) { + if (c == expectedCoreId) { + ASSERT_TRUE(CPU_ISSET(c, &cpuSet)); + } else { + ASSERT_FALSE(CPU_ISSET(c, &cpuSet)); + } + } + + }).join(); +} + + class TestData : public folly::RequestData { public: explicit TestData(int data) : data_(data) {}