Try again to implement a FIFO task queue

[zturner]

Summary

In D48115 I attempted to modify ThreadPool so that it could support tasks that returned values. I got this working, but in doing so I added a test that relied on sequential execution of tasks that run on the same thread, and this uncovered some problems in my approach.

What I *really* wanted was to create a ThreadPool with a single thread, queue up asynchronous work to it, and then assume that the work would be completed sequentially. After having some difficulty making these assumptions fit be consistent with those of the ThreadPool class, we decided that ultimately ThreadPool isn't the right place for that, and we should have a separate class explicitly for this purpose. So that patch was reverted.

Enter TaskQueue. TaskQueue consists of exactly 1 thread, that always executes its work first in first out. When you queue asynchronous work to it, it returns a future that can be waited on. When the wait returns, you are guaranteed that the work you waited on, as well as all previous work, has been completed.

Getting the LLVM_ENABLE_THREADS=0 case working is quite a tricky job. There may be a better solution than what I've come up with here, but this is the only way I could get it to work. When LLVM_ENABLE_THREADS=1, the problem is very easy. Just add the work to a queue, and process the queue on a single blessed background thread. When LLVM_ENABLE_THREADS=0 however, you have to consider the case where a user queues up multiple asynchronous tasks and then waits on them out of order. So if you queue F1, F2, and F3, and wait on F3, F3 has to manually invoke F2's work, which has to manually invoke F1's work. So the implementations of LLVM_ENABLE_THREADS and !LLVM_ENABLE_THREADS diverge somewhat, but I couldn't find any other way to make this work.

Note that my final goal (supporting tasks that return a value) is still unimplemented in this patch. I plan to do that in a followup, as there are some additional complexities that need to be taken into consideration, mostly in the LLVM_ENABLE_THREADS=0 branch.

[zturner]

Added 2 more tests, the first of which exposed a bug in the wait() function. All the wait() function was doing was joining the work thread, but this is incorrect. What it's supposed to do is simply drain the queue. i.e. keep processing work until the queue is empty. It's not supposed to join the thread. Joining the thread doesn't work anyway since we weren't setting the EnableFlag to be false, so this method was deadlocking.

To fix this I had to borrow some more code from ThreadPool, but all tests now pass both with LLVM_ENABLE_THREADS=0 and LLVM_ENABLE_THREADS=1.

[zturner]

Made the test cases a little stronger. Before I was simply assigning to the local variables in the test case. Now I increment them. This way the test will fail if a task were to ever get executed more than once, for example, whereas before it would pass since it would just be re-assigning the same value.

[chandlerc]

Before I dig in too much to the code, I want to better understand the underlying goal...

The core primitive here is a *serialized* task queue from what I understand...

Is it important that it starts a separate thread and eagerly (as opposed to lazily) processes the enqueued tasks?

If not, what about making this a lazy system and a more literal work queue?

If so, would it be a problem for the thread(s) used to be shared with other work at times (provided the synchronized aspect is preserved)?

[zturner]

In D48240#1134655, @chandlerc wrote:

Before I dig in too much to the code, I want to better understand the underlying goal...

The core primitive here is a *serialized* task queue from what I understand...

Is it important that it starts a separate thread and eagerly (as opposed to lazily) processes the enqueued tasks?

Yes

If so, would it be a problem for the thread(s) used to be shared with other work at times (provided the synchronized aspect is preserved)?

Depends what you mean by shared. Certainly you can queue up any work you want here, so if your jobs are A, C, and E, and someone else's jobs are B, D, and F, then queueing up A, B, C, D, E, F should work just fine.

The motivation for this is imagine you're writing an application that segments its work into different threads. For example, you might have a UI thread, a file i/o thread, a network thread. Work that changes the state of a UI element has to happen on the UI thread. Work that physically waits on a socket has to happen on the network thread, etc. This is a pretty common architecture for certain types of applications. Another example might be if you were writing a debugger which was debugging N processes simultaneously, and due to various OS restrictions, each of these N processes was being debugged by a separate thread. The application controlling all this would like to not have to deal with synchronization, but if each of these threads just said "wait for something to happen, then invoke a callback" the callback would happen on an arbitrary thread context and the user would be responsible for synchronizing all of this. It's nice to be able to guarantee something like "the callback always gets invoked on the same thread" to simplify the user's life.

LMK if this makes sense.

[labath]

Another example might be if you were writing a debugger which was debugging N processes simultaneously, and due to various OS restrictions, each of these N processes was being debugged by a separate thread.

We've had something similar to this in lldb (it was called ThreadStateCoordinator), but it was removed when we made the whole lldb-server process single-threaded.

llvm/include/llvm/Support/TaskQueue.h

41–43

Are all of these really needed? The only use of ParentFuture I see is in the lambda in ThisFuture, and that one already has it captured. ThisTask is not captured by that lambda, but it could easily be, at which point the whole Task struct could just become std::shared_future<void>.

llvm/lib/Support/TaskQueue.cpp

79

The assert message could be updated to reflect the new home.

102–107

This could be implemented as Tasks.back().wait(), avoiding the direct access into the ThisTask function and making sure the tasks are always invoked the same way.

llvm/unittests/Support/TaskQueueTest.cpp

64–70

I'm not sure if it's necessary, but if you wanted to have the test be uniform for both kinds of builds, you could replace MainThreadReady with an integer variable and then have the tasks wait until that variable is >= some_value.

[zturner]

llvm/include/llvm/Support/TaskQueue.h

41–43

I was able to remove them as you suggested in this patch, but things get more tricky in the follow-up patch where I extend this to allow the queued up tasks to return values. Regardless, since these are independent patches, I'm uploading a new version that has the suggested changes and I can try to make the other one use a similar model separately.

[chandlerc]

Some suggestions about the way you package things and manage std::future stuff below.

Some higher level comments here...

The term I think you're looking for is "synchronized" or "serial". So I would call this a SynchronizedTaskQueue or SerializedTaskQueue. Given the other use of the term "serialized", I'd lean towards Synchronized or Sync if that's too long to type. Ultimately, the name needs to make it clear that it enforces that the tasks scheduled here execute without overlap.

I also wonder whether you want to decouple this serialized execution from the thread creation and management. Could you accept a thread pool instead, and enqueue work on the thread pool? You can do this by having each task on the thread pool, on completion, check the queue for more work, and if there is some, schedule a new task on the thread pool. The reason why this is interesting is because if you have, say, a system thread pool that maybe gets a lot of other work, you can intersperse these tasks on it without overloading the system. It also allows the client of this class to dictate which threads are allowable threads for these tasks to run by providing a correct thread pool. My hope is that worst case, creating a one-thread thread pool and handing it to this class would be the equivalent of what you have here, but with a more clean separation of roles. I understand it would add some complexity, for example you'd have to handle the case where no work is currently in-flight when asyncImpl finishes adding the task to the queue and start a new batch. But this seems reasonably tolerable.

llvm/include/llvm/Support/TaskQueue.h

36–42

Rather than these types, I would suggest you build your task more like the following:

template <typename Callable>
std::future<void> async(Callable &&C) {
  // Because we don't have init capture to use move-only local variables that
  // are captured into a lambda, we create the promise inside an explicit callable
  // struct. We want to do as much of the wrapping in the type-specialized domain
  // (before type erasure) and then erase this into a std::function.
  struct Task {
    Callable C;
    std::promise<void> P;

    void operator()() noexcept {
      // Note, we don't catch exceptions and store them in the promise here because
      // we don't support exception handling in LLVM.
      C();
      P.set_value(C());
    }
  };

  Task T = {std::forward<Callable>(C)};
  std::future<void> F = T.P.get_future();
  asyncImpl(std::move(T));
  return F;
}

This is somewhat similar to what std::packaged_task does, but due to how std::packaged_task works, this will probably end up a bit lighter weight (specifically, std::packaged_task type erases the callable separately from the promise, while this packages the callable and the promise together ahead of time and then type erases it afterward.

The end result is that your task queue operates in a super simple closure type: std::function<void()>. This will be true even when you produce a value from the callable and store it into the promise. And this should also work even when you have no threading support at all. std::shared_future is a lot more expensive ultimately than std::future.

Does all of this make sense?

And FWIW, this is where C++14 is actually helpful, as there the above goop looks a lot cleaner:

template <typename Callable>
std::future<void> async(Callable &&C) {
  std::promise<void> P;
  std::future<void> F = P.get_future();
  asyncImpl([C = std::forward<Callable>(C), P = std::move(P)] {
      // Note, we don't catch exceptions and store them in the promise here because
      // we don't support exception handling in LLVM.
      C();
      P.set_value(C());
  });

  return F;
}

50–58

I regularly find std::bind to be problematic and hard to understand. Rather than expose this at all, I would force clients to handle it by passing you a lambda (or a std::bind if they so choose) that handles all of the argument binding. Then you just have a callable (the overload below).

63–64

I would term this a Callable as it may not be a function at all. The most common case I suspect will be a lambda.

[zturner]

In D48240#1136052, @chandlerc wrote:

I also wonder whether you want to decouple this serialized execution from the thread creation and management. Could you accept a thread pool instead, and enqueue work on the thread pool? You can do this by having each task on the thread pool, on completion, check the queue for more work, and if there is some, schedule a new task on the thread pool. The reason why this is interesting is because if you have, say, a system thread pool that maybe gets a lot of other work, you can intersperse these tasks on it without overloading the system. It also allows the client of this class to dictate which threads are allowable threads for these tasks to run by providing a correct thread pool. My hope is that worst case, creating a one-thread thread pool and handing it to this class would be the equivalent of what you have here, but with a more clean separation of roles. I understand it would add some complexity, for example you'd have to handle the case where no work is currently in-flight when asyncImpl finishes adding the task to the queue and start a new batch. But this seems reasonably tolerable.

Isn't using multiple threads strictly incompatible with the idea of it being serialized or synchronized? With multiple threads, you will have work overlapping, which confuses the purpose of what this is for. If someone comes up with a use case for it, then maybe, but otherwise, YAGNI?

I like your other suggestions (which I snipped from the quote), I'll try to make it work and report back if I can't.

One thing that's annoying is that you can't call promise.set_value(V) when V is void. Bleh.

[chandlerc]

In D48240#1136061, @zturner wrote:

In D48240#1136052, @chandlerc wrote:

I also wonder whether you want to decouple this serialized execution from the thread creation and management. Could you accept a thread pool instead, and enqueue work on the thread pool? You can do this by having each task on the thread pool, on completion, check the queue for more work, and if there is some, schedule a new task on the thread pool. The reason why this is interesting is because if you have, say, a system thread pool that maybe gets a lot of other work, you can intersperse these tasks on it without overloading the system. It also allows the client of this class to dictate which threads are allowable threads for these tasks to run by providing a correct thread pool. My hope is that worst case, creating a one-thread thread pool and handing it to this class would be the equivalent of what you have here, but with a more clean separation of roles. I understand it would add some complexity, for example you'd have to handle the case where no work is currently in-flight when asyncImpl finishes adding the task to the queue and start a new batch. But this seems reasonably tolerable.

Isn't using multiple threads strictly incompatible with the idea of it being serialized or synchronized? With multiple threads, you will have work overlapping, which confuses the purpose of what this is for. If someone comes up with a use case for it, then maybe, but otherwise, YAGNI?

It isn't about *using* multiple threads, its about selecting which threads run things.

Whether tasks are serialized in their execution is independent of how many threads they run on. Its just that there doesn't *need* to be more than one if they are serialized.

But a system thread pool might expose the equivalent of <1 thread.

Think of this example. You have three different systems which each want to use this infrastructure to run a series of asynchronous tasks, but for each system the series of tasks must run in-order (FIFO) and without overlap. You can re-use a single threadpool with potentially fewer than three threads for all three of these systems. This allows the client to control the rate of work -- perhaps only allowing at most two operations to happen concurrently.

While with three systems and two threads this seems somewhat odd, if you have *many* systems (maybe 100s) all using this to coordinate a series of tasks, you might not want hundereds of threads, all spinning on their own queues. Instead, it seems nicer to have a single threadpool, sized appropriately for the amount of resources on the system, and have all of the users of this class enqueue their serial sequence of tasks into this shared threadpool.

I like your other suggestions (which I snipped from the quote), I'll try to make it work and report back if I can't.

Yeah, the intent was that they're totally orgthogonal.

One thing that's annoying is that you can't call promise.set_value(V) when V is void. Bleh.

OMG YES! =[ I wish I knew a better way than horrible specialization on void that has different code.

[zturner]

In D48240#1136073, @chandlerc wrote:

In D48240#1136061, @zturner wrote:

In D48240#1136052, @chandlerc wrote:

I also wonder whether you want to decouple this serialized execution from the thread creation and management. Could you accept a thread pool instead, and enqueue work on the thread pool? You can do this by having each task on the thread pool, on completion, check the queue for more work, and if there is some, schedule a new task on the thread pool. The reason why this is interesting is because if you have, say, a system thread pool that maybe gets a lot of other work, you can intersperse these tasks on it without overloading the system. It also allows the client of this class to dictate which threads are allowable threads for these tasks to run by providing a correct thread pool. My hope is that worst case, creating a one-thread thread pool and handing it to this class would be the equivalent of what you have here, but with a more clean separation of roles. I understand it would add some complexity, for example you'd have to handle the case where no work is currently in-flight when asyncImpl finishes adding the task to the queue and start a new batch. But this seems reasonably tolerable.

Isn't using multiple threads strictly incompatible with the idea of it being serialized or synchronized? With multiple threads, you will have work overlapping, which confuses the purpose of what this is for. If someone comes up with a use case for it, then maybe, but otherwise, YAGNI?

It isn't about *using* multiple threads, its about selecting which threads run things.

Whether tasks are serialized in their execution is independent of how many threads they run on. Its just that there doesn't *need* to be more than one if they are serialized.

But a system thread pool might expose the equivalent of <1 thread.

Think of this example. You have three different systems which each want to use this infrastructure to run a series of asynchronous tasks, but for each system the series of tasks must run in-order (FIFO) and without overlap. You can re-use a single threadpool with potentially fewer than three threads for all three of these systems. This allows the client to control the rate of work -- perhaps only allowing at most two operations to happen concurrently.

While with three systems and two threads this seems somewhat odd, if you have *many* systems (maybe 100s) all using this to coordinate a series of tasks, you might not want hundereds of threads, all spinning on their own queues. Instead, it seems nicer to have a single threadpool, sized appropriately for the amount of resources on the system, and have all of the users of this class enqueue their serial sequence of tasks into this shared threadpool.

Would you be willing to allow that as a followup patch? I want to get the single threaded case working with this patch and the extension to allow tasks that return values, and then maybe add that on in a followup? (I'm not punting on it either, I actually will do it, but as you said it seems orthogonal so maybe makes sense to do in a followup)

[zturner]

I can't get your suggestion to work, it's very bizarre. It's claiming that I can't even *move* construct a std::function because Task has a deleted copy constructor.

See for yourself here:

https://godbolt.org/g/x1eCnX

Any ideas?

[zturner]

In D48240#1136107, @zturner wrote:

I can't get your suggestion to work, it's very bizarre. It's claiming that I can't even *move* construct a std::function because Task has a deleted copy constructor.

See for yourself here:

https://godbolt.org/g/x1eCnX

Any ideas?

Hmm... http://en.cppreference.com/w/cpp/utility/functional/function/function

Type requirements
-F must meet the requirements of Callable and CopyConstructible.

Yuck. Am I out of luck with this method? I could make it a shared_ptr<promise<void>>, but...

[zturner]

Yea I'm not really able to get anything working using the suggested method of the local class. The Task class here doesn't have any logic to wait on the parent task to complete, and furthermore we just return a future from a promise without using async with deferred launch, so waiting on the future doesn't run the function. All of this is fixable, but ultimately you just end up getting pretty close to what I already have I think.

[zturner]

Updated based on suggestions from Chandler. I don't have the LLVM_ENABLE_THREADS=0 case working yet, but wanted to make sure this is generally what you had in mind.

I also removed the wait() function entirely, as it's a little more difficult to implement now. I can't just wait on the ThreadPool because there could in theory be other work on the thread pool, so waiting on the task queue is not the same as waiting on the thread pool. I don't have an immediate need for this functionality, so unless someone else does, yagni.

[zturner]

This gets the LLVM_ENABLE_THREADS=0 case working. It does so by taking advantage of pointer invalidation properties of std::deque, so I can save off a pointer to the newly inserted element and just run tasks until the pointers match.

[chandlerc]

Really nice approach for THREADS=0 w/ the deferred future that checks that "enough" of the queue has been processed. I like it.

If you want to avoid the address stability stuff, you could always keep an integer next to the things in the queue and track the integers...

I want to think a bit harder to understand if using the address creates an A-B-A problem. I feel like it does, but my brain is exhausted and not able to pin it down yet.

Unrelated comments inline....

llvm/include/llvm/Support/TaskQueue.h

33–34

Documentation needs updating based on the new design.

48

Assert that the queue is empty at this point?

Also, rather than waiting on the entire thread-pool, maybe enqueue a no-op future into the queue and wait on that? We don't need the underlying executor to finish, just the queue to empty.

66–69

This doesn't seem right...

Only if there are *no* tasks in flight do you want to mov this into the scheduler's async. And when you do, you want to set tasks in flight to be true.

90

Add a FIXME about the std::function issue? D48349 has unique_function to eventually fix this.

97

FWIW, the copy of std::function, and the eventual move of llvm::unique_function won't be free. It may be worthwhile to keep this in a non-type-erased state by making CreateTask here a lambda within the async function above. =/ Just something to think about.

[zturner]

In D48240#1137381, @chandlerc wrote:

Really nice approach for THREADS=0 w/ the deferred future that checks that "enough" of the queue has been processed. I like it.

If you want to avoid the address stability stuff, you could always keep an integer next to the things in the queue and track the integers...

I want to think a bit harder to understand if using the address creates an A-B-A problem. I feel like it does, but my brain is exhausted and not able to pin it down yet.

I don't think it does, because in the LLVM_ENABLE_THREADS=0 case, there's literally only one thread in the entire application.

[zturner]

Updated with suggestions from chandler. I'm not sure why all my tests passed even despite having that bug that he noticed. Would be nice if there were a way to write a test for that case, but I'm not sure how. I also made the cleanup suggested to eliminate the extra cost of copying the task. This was a little bit annoying, because I had to go back to using the callable object instead of using the lambda so that I could get move semantics of the callable. But this means I can't easily capture this, so I had to move the callable up to the class level so that I could make it a friend to give it access to TaskQueue's internal state. Anyway, it all works now.

[labath]

In D48240#1138646, @zturner wrote:

Updated with suggestions from chandler. I'm not sure why all my tests passed even despite having that bug that he noticed. Would be nice if there were a way to write a test for that case, but I'm not sure how.

I think what happened was that all tasks were inserted into the scheduler unconditionally. But, since that was a one-thread scheduler which happens to process tasks in sequence, everything worked as it should.

It would be interesting to have a test with scheduler with at least two threads which verifies that the tasks are still processed serially. You could have the first task block and then validate that the second task hasn't been started.

llvm/include/llvm/Support/TaskQueue.h

38

Task uses different tags here (class) and in the definition (struct). Also, I think this friend declaration isn't really needed as Task should have access to TaskQueue internals anyway.

53–67

Maybe this part could be moved to a function in the TaskQueue class? It means less code would need to be duplicated between different Task instantiations, and the management of TaskQueue internal data structures would stay within that class.

132–133

Here you're using NextPtr after you have popped the element it points to off the queue. I think this is technically UB. Instead of that you could just have bool Done variable which you set at the beginning of the loop.

After that ABA should not be a problem since you're only ever using the ElementPtr pointer while it is live.

[zturner]

Updated based on more suggestions. I switched from using deque pointer stability to using explicit ids. I think this makes the code easier to reason about, especially in the followup patch which allows the tasks to return values. I also split out the LLVM_ENABLE_THREADS=1 case from the alternative. They were diverging enough that all the pre-processor branches made it difficult to follow. I think this is cleaner and more maintainable this way.

[labath]

I think I noticed one more bug in the !threads implementation. Other than that, this looks fine to me, but I'll leave the final ok to Chandler.

llvm/include/llvm/Support/TaskQueue.h

145–151

I just realized this could be wrong when the futures are waited on out of order. E.g. if you have F1 and F2, F2.wait() will increment NextRunId to 2. Then if you do F1.wait(), it will loop until NextRunId wraps back to 1 and crash in the process.

Fortunately, with RunIds it should be easy to fix this by changing != to <. If we're worried about uint64_t wrapping we can do something like Id-NextRunId < UINT64_MAX/2

(Also, this code could be moved into a separate function to decrease the number of instantiations of the lambda.)

[zturner]

After some discussion with Chandler, we actually decided to just delete the nothreads codepath and generate a compile error if anyone tries to use this class with nothreads. I'll upload a fix that does that later today.

[zturner]

This patch deletes the LLVM_ENABLE_THREADS=0 codepath and just errors out if you try to use this class in those circumstances. Since the regular codepath is actually pretty straightforward and doesn't have much complexity, I went ahead and added support for tasks that return values directly to this patch. It was only a couple of line change and saves me from having to make another patch for it.

[thakis]

In D48240#1141268, @zturner wrote:

This patch deletes the LLVM_ENABLE_THREADS=0 codepath and just errors out if you try to use this class in those circumstances. Since the regular codepath is actually pretty straightforward and doesn't have much complexity, I went ahead and added support for tasks that return values directly to this patch. It was only a couple of line change and saves me from having to make another patch for it.

Unsurprisingly this breaks the LLVM_ENABLE_THREADS=0 build:

FAILED: unittests/Support/CMakeFiles/SupportTests.dir/TaskQueueTest.cpp.o 
/b/build/slave/linux_upload_clang/build/src/third_party/llvm-bootstrap-install/bin/clang++   -DGTEST_HAS_PTHREAD=0 -DGTEST_HAS_RTTI=0 -DGTEST_HAS_TR1_TUPLE=0 -DGTEST_LANG_CXX11=1 -D_DEBUG -D_GNU_SOURCE -D__STDC_CONSTANT_MACROS -D__STDC_FORMAT_MACROS -D__STDC_LIMIT_MACROS -Iunittests/Support -I/b/build/slave/linux_upload_clang/build/src/third_party/llvm/unittests/Support -I/usr/include/libxml2 -Iinclude -I/b/build/slave/linux_upload_clang/build/src/third_party/llvm/include -I/b/build/slave/linux_upload_clang/build/src/third_party/llvm/utils/unittest/googletest/include -I/b/build/slave/linux_upload_clang/build/src/third_party/llvm/utils/unittest/googlemock/include --gcc-toolchain=/b/build/slave/linux_upload_clang/build/src/third_party/llvm-build-tools/gcc485precise -fPIC -fvisibility-inlines-hidden -Werror=date-time -Werror=unguarded-availability-new -std=c++11 -Wall -Wextra -Wno-unused-parameter -Wwrite-strings -Wcast-qual -Wmissing-field-initializers -pedantic -Wno-long-long -Wcovered-switch-default -Wnon-virtual-dtor -Wdelete-non-virtual-dtor -Wstring-conversion -fdiagnostics-color -ffunction-sections -fdata-sections -O3    -UNDEBUG  -Wno-variadic-macros -Wno-gnu-zero-variadic-macro-arguments -fno-exceptions -fno-rtti -MMD -MT unittests/Support/CMakeFiles/SupportTests.dir/TaskQueueTest.cpp.o -MF unittests/Support/CMakeFiles/SupportTests.dir/TaskQueueTest.cpp.o.d -o unittests/Support/CMakeFiles/SupportTests.dir/TaskQueueTest.cpp.o -c /b/build/slave/linux_upload_clang/build/src/third_party/llvm/unittests/Support/TaskQueueTest.cpp
In file included from /b/build/slave/linux_upload_clang/build/src/third_party/llvm/unittests/Support/TaskQueueTest.cpp:10:
/b/build/slave/linux_upload_clang/build/src/third_party/llvm/include/llvm/Support/TaskQueue.h:84:5: error: static_assert failed "TaskQueue requires building with LLVM_ENABLE_THREADS!"
    static_assert(false,
    ^             ~~~~~

What's the thinking here? Maybe TaskQueueTest.cpp shouldn't be part of the build if LLVM_ENABLE_THREADS=0? As-is, this just gives you a compile error if you use that option and build all targets (like our bots do).

[zturner]

This whole file is supposed to be ifdefed out in this case. Is it not? I
definitely tested this case

[thakis]

In D48240#1143531, @zturner wrote:

This whole file is supposed to be ifdefed out in this case. Is it not? I
definitely tested this case

Looking at it, it seems that the code is tested out, but the #include isn't, and that's what errors. Maybe you tested it by reconfiguring an existing cmake build dir instead of using a fresh one and were led astray by cmake's overaggressive caching of build settings?

[zturner]

The static_assert is in a template function though, which shouldn’t get
instantiated unless you create an instance of the class. And creating that
instance is ifdef’ed out. If it fixes it just move the ifdef up, but I’m a
bit confused.

Also i tested with msvc I guess that could be it

[thakis]

https://stackoverflow.com/questions/5246049/c11-static-assert-and-template-instantiation

[thakis]

Tried moving it up in r335608.

llvm/include/llvm/Support/TaskQueue.h

+//===-- llvm/Support/TaskQueue.h - A TaskQueue implementation ---*- C++ -*-===//
+//
+//                     The LLVM Compiler Infrastructure
+//
+// This file is distributed under the University of Illinois Open Source
+// License. See LICENSE.TXT for details.
+//
+//===----------------------------------------------------------------------===//
+//
+// This file defines a crude C++11 based task queue.
+//
+//===----------------------------------------------------------------------===//
+
+#ifndef LLVM_SUPPORT_TASK_QUEUE_H
+#define LLVM_SUPPORT_TASK_QUEUE_H
+
+#include "llvm/Config/llvm-config.h"
+#include "llvm/Support/thread.h"
+
+#include <future>
+
+#include <atomic>
+#include <condition_variable>
+#include <functional>
+#include <memory>
+#include <mutex>
+#include <queue>
+#include <utility>
+
+namespace llvm {
+/// TaskQueue executes work on exactly 1 thread in a first in first out
+/// fashion.  All work is guaranteed to execute in the order it is received.
+class TaskQueue {
+public:
+  using TaskFunctionTy = std::function<void()>;
+
+#if LLVM_ENABLE_THREADS
+  using TaskTy = std::packaged_task<void()>;
+#else
+  struct Task {
+    std::shared_future<void> ParentFuture;
+    std::shared_future<void> ThisFuture;
+    std::function<void()> ThisTask;
+  };
+  using TaskTy = Task;
+
+#endif
+
+  /// Construct a task queue with no work.
+  TaskQueue();
+
+  /// Blocking destructor: the queue will wait for all work to complete.
+  ~TaskQueue();
+
+  /// Asynchronous submission of a task to the queue. The returned future can be
+  /// used to wait for the task (and all previous tasks that have not yet
+  /// completed) to finish.
+  template <typename Function, typename... Args>
+  inline std::shared_future<void> async(Function &&F, Args &&... ArgList) {
+    auto Task =
+        std::bind(std::forward<Function>(F), std::forward<Args>(ArgList)...);
+    return asyncImpl(std::move(Task));
+  }
+
+  /// Asynchronous submission of a task to the queue. The returned future can be
+  /// used to wait for the task (and all previous tasks that have not yet
+  /// completed) to finish.
+  template <typename Function>
+  inline std::shared_future<void> async(Function &&F) {
+    return asyncImpl(std::forward<Function>(F));
+  }
+
+  /// Blocking wait for all work to complete.
+  void wait();
+
+private:
+  void WorkThreadFunc();
+
+  /// Asynchronous submission of a task to the queue. The returned future can be
+  /// used to wait for the task (and all previous tasks that have not yet
+  /// completed) to finish.
+  std::shared_future<void> asyncImpl(TaskFunctionTy F);
+
+  /// Tasks waiting for execution in the queue.
+  std::queue<TaskTy> Tasks;
+
+#if LLVM_ENABLE_THREADS
+  /// Signal for the destruction of the pool, asking thread to exit.
+  /// This needs to come before WorkThread so that it is initialized first in
+  /// the constructor.
+  bool EnableFlag = true;
+
+  /// Flag indicating whether the queue is in the middle of processing a job.
+  bool Active = false;
+
+  /// Locking and signaling for accessing the Tasks queue.
+  std::mutex QueueLock;
+  std::condition_variable QueueCondition;
+
+  /// Locking and signaling for job completion
+  std::mutex CompletionLock;
+  std::condition_variable CompletionCondition;
+
+  /// The main work thread.
+  llvm::thread WorkThread;
+#endif
+};
+} // namespace llvm
+
+#endif // LLVM_SUPPORT_TASK_QUEUE_H

llvm/lib/Support/CMakeLists.txt

add_llvm_library(LLVMSupport
   StringExtras.cpp
   StringMap.cpp
   StringPool.cpp
   StringSaver.cpp
   StringRef.cpp
   SystemUtils.cpp
   TarWriter.cpp
   TargetParser.cpp
+  TaskQueue.cpp
   ThreadPool.cpp
   Timer.cpp
   ToolOutputFile.cpp
   TrigramIndex.cpp
   Triple.cpp
   Twine.cpp
   Unicode.cpp
   UnicodeCaseFold.cpp

llvm/lib/Support/TaskQueue.cpp

+//==-- llvm/Support/TaskQueue.cpp - A TaskQueue implementation ---*- C++ -*-==//
+//
+//                     The LLVM Compiler Infrastructure
+//
+// This file is distributed under the University of Illinois Open Source
+// License. See LICENSE.TXT for details.
+//
+//===----------------------------------------------------------------------===//
+//
+// This file implements a crude C++11 based task queue.
+//
+//===----------------------------------------------------------------------===//
+
+#include "llvm/Support/TaskQueue.h"
+
+#include "llvm/Config/llvm-config.h"
+#include "llvm/Support/Threading.h"
+#include "llvm/Support/raw_ostream.h"
+
+using namespace llvm;
+
+#if LLVM_ENABLE_THREADS
+
+TaskQueue::TaskQueue() : WorkThread([this] { WorkThreadFunc(); }) {}
+
+void TaskQueue::WorkThreadFunc() {
+  while (true) {
+    TaskTy TheTask;
+    {
+      std::unique_lock<std::mutex> LockGuard(QueueLock);
+      // Wait for tasks to be pushed in the queue
+      QueueCondition.wait(LockGuard,
+                          [&] { return !EnableFlag || !Tasks.empty(); });
+      // Exit condition
+      if (!EnableFlag && Tasks.empty())
+        return;
+
+      {
+        std::unique_lock<std::mutex> LockGuard(CompletionLock);
+        Active = true;
+      }
+      // Yeah, we have a task, grab it and release the lock on the queue
+      TheTask = std::move(Tasks.front());
+      Tasks.pop();
+    }
+
+    // Run the task we just grabbed and update the ids.
+    TheTask();
+
+    {
+      std::unique_lock<std::mutex> LockGuard(CompletionLock);
+      Active = false;
+    }
+
+    // Notify task completion, in case someone waits on ThreadPool::wait()
+    CompletionCondition.notify_all();
+  }
+}
+
+void TaskQueue::wait() {
+  // Wait for the queue to be empty
+  std::unique_lock<std::mutex> LockGuard(CompletionLock);
+
+  // The order of the checks for ActiveThreads and Tasks.empty() matters because
+  // any active threads might be modifying the Tasks queue, and this would be a
+  // race.
+  CompletionCondition.wait(LockGuard, [&] { return !Active && Tasks.empty(); });
+}
+
+std::shared_future<void> TaskQueue::asyncImpl(TaskFunctionTy TheTask) {
+  /// Wrap the Task in a packaged_task to return a future object.
+  TaskTy PackagedTask(std::move(TheTask));
+  auto Future = PackagedTask.get_future();
+  {
+    // Lock the queue and push the new task
+    std::unique_lock<std::mutex> LockGuard(QueueLock);
+
+    // Don't allow enqueueing after disabling the pool
+    assert(EnableFlag && "Queuing a thread during ThreadPool destruction");
+
+    Tasks.push(std::move(PackagedTask));
+  }
+  QueueCondition.notify_one();
+  return Future.share();
+}
+
+// The destructor joins the work thread, waiting for completion.
+TaskQueue::~TaskQueue() {
+  {
+    std::unique_lock<std::mutex> LockGuard(QueueLock);
+    EnableFlag = false;
+  }
+  QueueCondition.notify_one();
+  WorkThread.join();
+}
+
+#else // LLVM_ENABLE_THREADS Disabled
+
+TaskQueue::TaskQueue() {}
+
+void TaskQueue::wait() {
+  // Sequential implementation running the tasks
+  while (!Tasks.empty()) {
+    auto Task = std::move(Tasks.front());
+    Tasks.pop();
+    Task.ThisTask();
+  }
+}
+
+std::shared_future<void> TaskQueue::asyncImpl(TaskFunctionTy TheTask) {
+  std::shared_future<void> ThisFuture;
+  std::shared_future<void> ParentFuture;
+  if (!Tasks.empty())
+    ParentFuture = Tasks.back().ThisFuture;
+
+  ThisFuture = std::async(std::launch::deferred, [this, ParentFuture] {
+    if (ParentFuture.valid())
+      ParentFuture.wait();
+    auto Task = std::move(Tasks.front());
+    Tasks.pop();
+    Task.ThisTask();
+  });
+  Tasks.push({ParentFuture, ThisFuture, std::move(TheTask)});
+  return ThisFuture;
+}
+
+TaskQueue::~TaskQueue() { wait(); }
+
+#endif

llvm/unittests/Support/CMakeLists.txt

add_llvm_unittest(SupportTests
   ReplaceFileTest.cpp
   ScaledNumberTest.cpp
   SourceMgrTest.cpp
   SpecialCaseListTest.cpp
   StringPool.cpp
   SwapByteOrderTest.cpp
   TarWriterTest.cpp
   TargetParserTest.cpp
+  TaskQueueTest.cpp
   ThreadLocalTest.cpp
   ThreadPool.cpp
   Threading.cpp
   TimerTest.cpp
   TypeNameTest.cpp
   TrailingObjectsTest.cpp
   TrigramIndexTest.cpp
   UnicodeTest.cpp

llvm/unittests/Support/TaskQueueTest.cpp

+//========- unittests/Support/TaskQueue.cpp - TaskQueue.h tests ------========//
+//
+//                     The LLVM Compiler Infrastructure
+//
+// This file is distributed under the University of Illinois Open Source
+// License. See LICENSE.TXT for details.
+//
+//===----------------------------------------------------------------------===//
+
+#include "llvm/Support/TaskQueue.h"
+
+#include "gtest/gtest.h"
+
+using namespace llvm;
+
+class TaskQueueTest : public testing::Test {
+protected:
+  TaskQueueTest() {}
+
+  /// Make sure this thread not progress faster than the main thread.
+  void waitForMainThread() {
+    std::unique_lock<std::mutex> LockGuard(WaitMainThreadMutex);
+    WaitMainThread.wait(LockGuard, [&] { return MainThreadReady; });
+  }
+
+  /// Set the readiness of the main thread.
+  void setMainThreadReady() {
+    {
+      std::unique_lock<std::mutex> LockGuard(WaitMainThreadMutex);
+      MainThreadReady = true;
+    }
+    WaitMainThread.notify_all();
+  }
+
+  void SetUp() override { MainThreadReady = false; }
+
+  std::condition_variable WaitMainThread;
+  std::mutex WaitMainThreadMutex;
+  bool MainThreadReady;
+};
+
+TEST_F(TaskQueueTest, OrderedFutures) {
+  TaskQueue TQ;
+  std::atomic<int> X = 0;
+  std::atomic<int> Y = 0;
+  std::atomic<int> Z = 0;
+
+  std::shared_future<void> F1 = TQ.async([&] {
+    waitForMainThread();
+    X = 1;
+  });
+  std::shared_future<void> F2 = TQ.async([&] {
+    waitForMainThread();
+    Y = 1;
+  });
+  std::shared_future<void> F3 = TQ.async([&] {
+    waitForMainThread();
+    Z = 1;
+  });
+
+  setMainThreadReady();
+  F1.wait();
+  ASSERT_EQ(1, X);
+#if !LLVM_ENABLE_THREADS
+  // We can't check this in the LLVM_ENABLE_THREADS case since the work thread
+  // proceeds independently of the main thread, so this may or may not still
+  // be true.
+  ASSERT_EQ(0, Y);
+  ASSERT_EQ(0, Z);
+#endif
+
+  F2.wait();
+  ASSERT_EQ(1, X);
+  ASSERT_EQ(1, Y);
+#if !LLVM_ENABLE_THREADS
+  ASSERT_EQ(0, Z);
+#endif
+
+  F3.wait();
+  ASSERT_EQ(1, X);
+  ASSERT_EQ(1, Y);
+  ASSERT_EQ(1, Z);
+}
+
+TEST_F(TaskQueueTest, UnOrderedFutures) {
+  TaskQueue TQ;
+  std::atomic<int> X = 0;
+  std::atomic<int> Y = 0;
+  std::atomic<int> Z = 0;
+  std::shared_future<void> F1 = TQ.async([&] { X = 1; });
+  std::shared_future<void> F2 = TQ.async([&] { Y = 1; });
+  std::shared_future<void> F3 = TQ.async([&] { Z = 1; });
+
+  F2.wait();
+  ASSERT_EQ(1, X);
+  ASSERT_EQ(1, Y);
+#if !LLVM_ENABLE_THREADS
+  // We can't check this in the LLVM_ENABLE_THREADS case since the work thread
+  // proceeds independently of the main thread, so this may or may not still
+  // be true.
+  ASSERT_EQ(0, Z);
+#endif
+
+  F3.wait();
+  ASSERT_EQ(1, X);
+  ASSERT_EQ(1, Y);
+  ASSERT_EQ(1, Z);
+}
+
+TEST_F(TaskQueueTest, Wait) {
+  // Test that calling wait() on a TaskQueue processes all outstanding work.
+  TaskQueue TQ;
+  std::atomic<int> X = 0;
+  std::atomic<int> Y = 0;
+  std::atomic<int> Z = 0;
+  TQ.async([&] { X = 1; });
+  TQ.async([&] { Y = 1; });
+  TQ.async([&] { Z = 1; });
+
+  TQ.wait();
+  ASSERT_EQ(1, X);
+  ASSERT_EQ(1, Y);
+  ASSERT_EQ(1, Z);
+}
+
+TEST_F(TaskQueueTest, WaitAgain) {
+  // Test that calling wait() on a TaskQueue and queueing up more work processes
+  // the new work as it should.
+  TaskQueue TQ;
+  std::atomic<int> X = 0;
+  std::atomic<int> Y = 0;
+  std::atomic<int> Z = 0;
+  TQ.async([&] { ++X; });
+  TQ.async([&] { ++Y; });
+  TQ.async([&] { ++Z; });
+
+  TQ.wait();
+  ASSERT_EQ(1, X);
+  ASSERT_EQ(1, Y);
+  ASSERT_EQ(1, Z);
+
+  TQ.async([&] { ++X; });
+  TQ.async([&] { ++Y; });
+  TQ.async([&] { ++Z; });
+
+  TQ.wait();
+  ASSERT_EQ(2, X);
+  ASSERT_EQ(2, Y);
+  ASSERT_EQ(2, Z);
+}
+
+TEST_F(TaskQueueTest, QueueWaitAfterFutureWait) {
+  // Test that waiting on the result of some work and then waiting on the
+  // entire queue processes each job exactly once, and in the correct
+  // order.
+  TaskQueue TQ;
+  std::atomic<int> X = 0;
+  std::atomic<int> Y = 0;
+  std::atomic<int> Z = 0;
+  std::atomic<int> A = 0;
+  std::shared_future<void> F1 = TQ.async([&] { ++X; });
+  std::shared_future<void> F2 = TQ.async([&] { ++Y; });
+  std::shared_future<void> F3 = TQ.async([&] { ++Z; });
+  std::shared_future<void> F4 = TQ.async([&] { ++A; });
+
+  F2.wait();
+  ASSERT_EQ(1, X);
+  ASSERT_EQ(1, Y);
+#if !LLVM_ENABLE_THREADS
+  // We can't check this in the LLVM_ENABLE_THREADS case since the work thread
+  // proceeds independently of the main thread, so this may or may not still
+  // be true.
+  ASSERT_EQ(0, Z);
+  ASSERT_EQ(0, A);
+#endif
+
+  TQ.wait();
+  ASSERT_EQ(1, X);
+  ASSERT_EQ(1, Y);
+  ASSERT_EQ(1, Z);
+  ASSERT_EQ(1, A);
+}