~TaskGroup();
// Spawn a task, but does not wait for it to finish.
- void spawn(std::function<void()> f);
-
- // Similar to spawn, but execute the task immediately when ThreadsRequested ==
- // 1. The difference is to give the following pattern a more intuitive order
- // when single threading is requested.
- //
- // for (size_t begin = 0, i = 0, taskSize = 0;;) {
- // taskSize += ...
- // bool done = ++i == end;
- // if (done || taskSize >= taskSizeLimit) {
- // tg.execute([=] { fn(begin, i); });
- // if (done)
- // break;
- // begin = i;
- // taskSize = 0;
- // }
- // }
- void execute(std::function<void()> f);
+ // Tasks marked with \p Sequential will be executed
+ // exactly in the order which they were spawned.
+ // Note: Sequential tasks may be executed on different
+ // threads, but strictly in sequential order.
+ void spawn(std::function<void()> f, bool Sequential = false);
void sync() const { L.sync(); }
};
#include "llvm/Support/Threading.h"
#include <atomic>
+#include <deque>
#include <future>
-#include <stack>
#include <thread>
#include <vector>
class Executor {
public:
virtual ~Executor() = default;
- virtual void add(std::function<void()> func) = 0;
+ virtual void add(std::function<void()> func, bool Sequential = false) = 0;
static Executor *getDefaultExecutor();
};
static void call(void *Ptr) { ((ThreadPoolExecutor *)Ptr)->stop(); }
};
- void add(std::function<void()> F) override {
+ void add(std::function<void()> F, bool Sequential = false) override {
{
+ bool UseSequentialQueue =
+ Sequential || parallel::strategy.ThreadsRequested == 1;
std::lock_guard<std::mutex> Lock(Mutex);
- WorkStack.push(std::move(F));
+ if (UseSequentialQueue)
+ WorkQueueSequential.emplace_front(std::move(F));
+ else
+ WorkQueue.emplace_back(std::move(F));
}
Cond.notify_one();
}
private:
+ bool hasSequentialTasks() const {
+ return !WorkQueueSequential.empty() && !SequentialQueueIsLocked;
+ }
+
+ bool hasGeneralTasks() const { return !WorkQueue.empty(); }
+
void work(ThreadPoolStrategy S, unsigned ThreadID) {
threadIndex = ThreadID;
S.apply_thread_strategy(ThreadID);
while (true) {
std::unique_lock<std::mutex> Lock(Mutex);
- Cond.wait(Lock, [&] { return Stop || !WorkStack.empty(); });
+ Cond.wait(Lock, [&] {
+ return Stop || hasGeneralTasks() || hasSequentialTasks();
+ });
if (Stop)
break;
- auto Task = std::move(WorkStack.top());
- WorkStack.pop();
+ bool Sequential = hasSequentialTasks();
+ if (Sequential)
+ SequentialQueueIsLocked = true;
+ else
+ assert(hasGeneralTasks());
+
+ auto &Queue = Sequential ? WorkQueueSequential : WorkQueue;
+ auto Task = std::move(Queue.back());
+ Queue.pop_back();
Lock.unlock();
Task();
+ if (Sequential)
+ SequentialQueueIsLocked = false;
}
}
std::atomic<bool> Stop{false};
- std::stack<std::function<void()>> WorkStack;
+ std::atomic<bool> SequentialQueueIsLocked{false};
+ std::deque<std::function<void()>> WorkQueue;
+ std::deque<std::function<void()>> WorkQueueSequential;
std::mutex Mutex;
std::condition_variable Cond;
std::promise<void> ThreadsCreated;
--TaskGroupInstances;
}
-void TaskGroup::spawn(std::function<void()> F) {
+void TaskGroup::spawn(std::function<void()> F, bool Sequential) {
#if LLVM_ENABLE_THREADS
if (Parallel) {
L.inc();
- detail::Executor::getDefaultExecutor()->add([&, F = std::move(F)] {
- F();
- L.dec();
- });
+ detail::Executor::getDefaultExecutor()->add(
+ [&, F = std::move(F)] {
+ F();
+ L.dec();
+ },
+ Sequential);
return;
}
#endif
F();
}
-void TaskGroup::execute(std::function<void()> F) {
- if (parallel::strategy.ThreadsRequested == 1)
- F();
- else
- spawn(F);
-}
} // namespace parallel
} // namespace llvm