Commit d8b4f2ff authored by Andrew Krieger's avatar Andrew Krieger Committed by Facebook Github Bot

Use extended alignment safe allocator in CPUThreadPoolExecutor

Summary:
Until C++17 types with extended alignment need special handling
in eg. std::vector and also std::unique_ptr. CPUThreadPoolExecutor has a
default queue that requires extended alignment, but also allows the user
to provide their own blocking queue. To handle this we move the private
member to a shared_ptr which supported type erased destructors, and then
use allocate_shared for the default queue type with a custom allocator
that will satisfy alignment constraints.

```
.../unique_ptr.h:825:34: runtime error: constructor call on misaligned address 0x613000000040 for type 'folly::UnboundedBlockingQueue<folly::CPUThreadPoolExecutor::CPUTask>', which requires 128 byte alignment
0x613000000040: note: pointer points here
 02 00 00 1c  be be be be be be be be  be be be be be be be be  be be be be be be be be  be be be be
              ^
    #0 0x75b35e in std::_MakeUniq<folly::UnboundedBlockingQueue<folly::CPUThreadPoolExecutor::CPUTask> >::__single_object std::make_unique<folly::UnboundedBlockingQueue<folly::CPUThreadPoolExecutor::CPUTask> >() .../unique_ptr.h:825:30
    #1 0x75602f in folly::CPUThreadPoolExecutor::CPUThreadPoolExecutor(unsigned long, std::shared_ptr<folly::ThreadFactory>) xplat/folly/executors/CPUThreadPoolExecutor.cpp:66:18
    #2 0x756c6c in folly::CPUThreadPoolExecutor::CPUThreadPoolExecutor(unsigned long) xplat/folly/executors/CPUThreadPoolExecutor.cpp:84:7
```

Differential Revision: D19669832

fbshipit-source-id: 7fb60a06dbb6a04edddb6b619af7a8cef4995fd2
parent 5adba359
...@@ -16,6 +16,7 @@ ...@@ -16,6 +16,7 @@
#include <folly/executors/CPUThreadPoolExecutor.h> #include <folly/executors/CPUThreadPoolExecutor.h>
#include <folly/Memory.h>
#include <folly/executors/task_queue/PriorityLifoSemMPMCQueue.h> #include <folly/executors/task_queue/PriorityLifoSemMPMCQueue.h>
#include <folly/executors/task_queue/PriorityUnboundedBlockingQueue.h> #include <folly/executors/task_queue/PriorityUnboundedBlockingQueue.h>
#include <folly/executors/task_queue/UnboundedBlockingQueue.h> #include <folly/executors/task_queue/UnboundedBlockingQueue.h>
...@@ -28,6 +29,16 @@ DEFINE_bool( ...@@ -28,6 +29,16 @@ DEFINE_bool(
namespace folly { namespace folly {
namespace {
// queue_alloc custom allocator is necessary until C++17
// http://open-std.org/JTC1/SC22/WG21/docs/papers/2012/n3396.htm
// https://gcc.gnu.org/bugzilla/show_bug.cgi?id=65122
// https://bugs.llvm.org/show_bug.cgi?id=22634
using default_queue = UnboundedBlockingQueue<CPUThreadPoolExecutor::CPUTask>;
using default_queue_alloc =
AlignedSysAllocator<default_queue, FixedAlign<alignof(default_queue)>>;
} // namespace
const size_t CPUThreadPoolExecutor::kDefaultMaxQueueSize = 1 << 14; const size_t CPUThreadPoolExecutor::kDefaultMaxQueueSize = 1 << 14;
CPUThreadPoolExecutor::CPUThreadPoolExecutor( CPUThreadPoolExecutor::CPUThreadPoolExecutor(
...@@ -38,7 +49,7 @@ CPUThreadPoolExecutor::CPUThreadPoolExecutor( ...@@ -38,7 +49,7 @@ CPUThreadPoolExecutor::CPUThreadPoolExecutor(
numThreads, numThreads,
FLAGS_dynamic_cputhreadpoolexecutor ? 0 : numThreads, FLAGS_dynamic_cputhreadpoolexecutor ? 0 : numThreads,
std::move(threadFactory)), std::move(threadFactory)),
taskQueue_(std::move(taskQueue)) { taskQueue_(taskQueue.release()) {
setNumThreads(numThreads); setNumThreads(numThreads);
registerThreadPoolExecutor(this); registerThreadPoolExecutor(this);
} }
...@@ -51,7 +62,7 @@ CPUThreadPoolExecutor::CPUThreadPoolExecutor( ...@@ -51,7 +62,7 @@ CPUThreadPoolExecutor::CPUThreadPoolExecutor(
numThreads.first, numThreads.first,
numThreads.second, numThreads.second,
std::move(threadFactory)), std::move(threadFactory)),
taskQueue_(std::move(taskQueue)) { taskQueue_(taskQueue.release()) {
setNumThreads(numThreads.first); setNumThreads(numThreads.first);
registerThreadPoolExecutor(this); registerThreadPoolExecutor(this);
} }
...@@ -63,7 +74,7 @@ CPUThreadPoolExecutor::CPUThreadPoolExecutor( ...@@ -63,7 +74,7 @@ CPUThreadPoolExecutor::CPUThreadPoolExecutor(
numThreads, numThreads,
FLAGS_dynamic_cputhreadpoolexecutor ? 0 : numThreads, FLAGS_dynamic_cputhreadpoolexecutor ? 0 : numThreads,
std::move(threadFactory)), std::move(threadFactory)),
taskQueue_(std::make_unique<UnboundedBlockingQueue<CPUTask>>()) { taskQueue_(std::allocate_shared<default_queue>(default_queue_alloc{})) {
setNumThreads(numThreads); setNumThreads(numThreads);
registerThreadPoolExecutor(this); registerThreadPoolExecutor(this);
} }
...@@ -75,7 +86,7 @@ CPUThreadPoolExecutor::CPUThreadPoolExecutor( ...@@ -75,7 +86,7 @@ CPUThreadPoolExecutor::CPUThreadPoolExecutor(
numThreads.first, numThreads.first,
numThreads.second, numThreads.second,
std::move(threadFactory)), std::move(threadFactory)),
taskQueue_(std::make_unique<UnboundedBlockingQueue<CPUTask>>()) { taskQueue_(std::allocate_shared<default_queue>(default_queue_alloc{})) {
setNumThreads(numThreads.first); setNumThreads(numThreads.first);
registerThreadPoolExecutor(this); registerThreadPoolExecutor(this);
} }
......
...@@ -146,7 +146,8 @@ class CPUThreadPoolExecutor : public ThreadPoolExecutor { ...@@ -146,7 +146,8 @@ class CPUThreadPoolExecutor : public ThreadPoolExecutor {
bool tryDecrToStop(); bool tryDecrToStop();
bool taskShouldStop(folly::Optional<CPUTask>&); bool taskShouldStop(folly::Optional<CPUTask>&);
std::unique_ptr<BlockingQueue<CPUTask>> taskQueue_; // shared_ptr for type erased dtor to handle extended alignment.
std::shared_ptr<BlockingQueue<CPUTask>> taskQueue_;
std::atomic<ssize_t> threadsToStop_{0}; std::atomic<ssize_t> threadsToStop_{0};
}; };
......
Markdown is supported
0%
or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment