From 657bc70d6295aeb7f4329266eef79c24f7b7f988 Mon Sep 17 00:00:00 2001 From: Wolfgang Bangerth Date: Tue, 26 May 2020 18:53:36 -0600 Subject: [PATCH] Rewrite the Threads::Task class using C++11 features. --- include/deal.II/base/thread_management.h | 419 +++++++---------------- 1 file changed, 123 insertions(+), 296 deletions(-) diff --git a/include/deal.II/base/thread_management.h b/include/deal.II/base/thread_management.h index dfa6751af5..73bc007643 100644 --- a/include/deal.II/base/thread_management.h +++ b/include/deal.II/base/thread_management.h @@ -26,6 +26,7 @@ # include # include +# include # include # include # include @@ -281,6 +282,12 @@ namespace Threads { value = std::move(v); } + + inline void + set_from(std::future &v) + { + value = std::move(v.get()); + } }; @@ -317,6 +324,12 @@ namespace Threads { value = &v; } + + inline void + set_from(std::future &v) + { + value = &v.get(); + } }; @@ -336,6 +349,11 @@ namespace Threads static inline void get() {} + + + inline void + set_from(std::future &) + {} }; } // namespace internal @@ -924,284 +942,6 @@ namespace Threads }; - template - class Task; - - - namespace internal - { -# ifdef DEAL_II_WITH_THREADS - - template - struct TaskDescriptor; - - /** - * The task class for TBB that is used by the TaskDescriptor class. - */ - template - struct TaskEntryPoint : public tbb::task - { - TaskEntryPoint(TaskDescriptor &task_descriptor) - : task_descriptor(task_descriptor) - {} - - virtual tbb::task * - execute() override - { - // call the function object and put the return value into the - // proper place - try - { - call(task_descriptor.function, task_descriptor.ret_val); - } - catch (const std::exception &exc) - { - internal::handle_std_exception(exc); - } - catch (...) - { - internal::handle_unknown_exception(); - } - return nullptr; - } - - /** - * A reference to the descriptor object of this task. - */ - TaskDescriptor &task_descriptor; - }; - - /** - * @internal - * - * Base class describing a task. This is the basic class abstracting the - * Threading Building Blocks implementation of tasks. It provides a - * mechanism to start a new task, as well as for joining it. - * - * Internally, the way things are implemented is that all Task<> objects - * keep a shared pointer to the task descriptor. When the last Task<> - * object goes out of scope, the destructor of the descriptor is called. - * Since tasks can not be abandoned, the destructor makes sure that the - * task is finished before it can continue to destroy the object. - * - * Note that unlike threads, tasks are not always started right away, and - * so the starting thread can't rely on the fact that the started task can - * copy things off the spawning thread's stack frame. As a consequence, - * the task description needs to include a way to store the function and - * its arguments that shall be run on the task. - * - * @author Wolfgang Bangerth, 2009 - */ - template - struct TaskDescriptor - { - private: - /** - * The function and its arguments that are to be run on the task. - */ - std::function function; - - /** - * Variable holding the data the TBB needs to work with a task. Set by - * the queue_up_task() function. Note that the object behind this - * pointer will be deleted upon termination of the task, so we do not - * have to do so ourselves. In particular, if all objects with pointers - * to this task_description object go out of scope then no action is - * needed on our behalf. - */ - tbb::task *task; - - /** - * A place where the task will deposit its return value. - */ - return_value ret_val; - - /** - * A flag indicating whether the task has terminated. - */ - bool task_is_done; - - public: - /** - * Constructor. Take the function to be run on this task as argument. - */ - TaskDescriptor(const std::function &function); - - /** - * Default constructor. Throws an exception since we want to queue a - * task immediately upon construction of these objects to make sure that - * each TaskDescriptor object corresponds to exactly one task. - */ - TaskDescriptor(); - - /** - * Copy constructor. Objects of this type can not be copied, and so this - * constructor is `delete`d and can't be called. - */ - TaskDescriptor(const TaskDescriptor &) = delete; - - /** - * Destructor. - */ - ~TaskDescriptor(); - - /** - * Copy operator. Objects of this type can not be copied, and so this - * operator is `delete`d and can't be called. - */ - TaskDescriptor & - operator=(const TaskDescriptor &) = delete; - - /** - * Queue up the task to the scheduler. We need to do this in a separate - * function since the new tasks needs to access objects from the current - * object and that can only reliably happen if the current object is - * completely constructed already. - */ - void - queue_task(); - - /** - * Join a task, i.e. wait for it to finish. This function can safely be - * called from different threads at the same time, and can also be - * called more than once. - */ - void - join(); - - - template - friend struct TaskEntryPoint; - friend class dealii::Threads::Task; - }; - - - - template - inline TaskDescriptor::TaskDescriptor( - const std::function &function) - : function(function) - , task(nullptr) - , task_is_done(false) - {} - - - template - inline void - TaskDescriptor::queue_task() - { - // use the pattern described in the TBB book on pages 230/231 - // ("Start a large task in parallel with the main program") - task = new (tbb::task::allocate_root()) tbb::empty_task; - task->set_ref_count(2); - - tbb::task *worker = - new (task->allocate_child()) TaskEntryPoint(*this); - - tbb::task::spawn(*worker); - } - - - - template - TaskDescriptor::TaskDescriptor() - : task_is_done(false) - { - Assert(false, ExcInternalError()); - } - - - - template - inline TaskDescriptor::~TaskDescriptor() - { - // wait for the task to complete for sure - join(); - - // now destroy the empty task structure. the book recommends to - // spawn it as well and let the scheduler destroy the object - // when done, but this has the disadvantage that the scheduler - // may not get to actually finishing the task before it goes out - // of scope (at the end of the program, or if a thread is done - // on which it was run) and then we would get a hard-to-decipher - // warning about unfinished tasks when the scheduler "goes out - // of the arena". rather, let's explicitly destroy the empty - // task object. before that, make sure that the task has been - // shut down, expressed by a zero reference count - AssertNothrow(task != nullptr, ExcInternalError()); - AssertNothrow(task->ref_count() == 0, ExcInternalError()); - task->destroy(*task); - } - - - template - inline void - TaskDescriptor::join() - { - // if the task is already done, just return. this makes sure we - // call tbb::Task::wait_for_all() exactly once, as required by - // TBB. we could also get the reference count of task for doing - // this, but that is usually slower. note that this does not - // work when the thread calling this function is not the same as - // the one that initialized the task. - // - // TODO: can we assert that no other thread tries to end the - // task? - if (task_is_done == true) - return; - - // let TBB wait for the task to complete. - task_is_done = true; - task->wait_for_all(); - } - - - -# else // no threading enabled - - /** - * A way to describe tasks. Since we are in non-MT mode at this place, - * things are a lot simpler than in MT mode. - */ - template - struct TaskDescriptor - { - /** - * A place where the task will deposit its return value. - */ - return_value ret_val; - - /** - * Constructor. Call the given function and emplace the return value - * into the slot reserved for this purpose. - */ - TaskDescriptor(const std::function &function) - { - call(function, ret_val); - } - - /** - * Wait for the task to return. Since we are in non-MT mode here, there - * is nothing to do. - */ - static void - join() - {} - - /** - * Run the task. Since we are here in non-MT mode, there is nothing to - * do that the constructor hasn't already done. - */ - static void - queue_task() - {} - }; - -# endif - - } // namespace internal - - /** * This class describes a task object, i.e., what one obtains by calling @@ -1221,7 +961,11 @@ namespace Threads * [`std::async`](https://en.cppreference.com/w/cpp/thread/async) (which is * itself similar to what Threads::new_task() does). The principal conceptual * difference is that one can only call `std::future::get()` once, whereas one - * can call Threads::Task::return_value() as many times as desired. + * can call Threads::Task::return_value() as many times as desired. It is, + * thus, comparable to the + * [`std::shared_future`](https://en.cppreference.com/w/cpp/thread/shared_future) + * class. However, `std::shared_future` can not be used for types that can not + * be copied -- a particular restriction for `std::unique_ptr`, for example. * * @author Wolfgang Bangerth, 2009, 2020 * @ingroup threads @@ -1239,15 +983,10 @@ namespace Threads */ Task(const std::function &function_object) { - // create a task descriptor and tell it to queue itself up with - // the scheduling system - task_descriptor = - std::make_shared>(function_object); - task_descriptor->queue_task(); + task_data = std::make_shared( + std::move(std::async(std::launch::async, function_object))); } - - /** * Default constructor. You can't do much with a task object constructed * this way, except for assigning it a task object that holds data created @@ -1258,8 +997,6 @@ namespace Threads */ Task() = default; - - /** * Join the task represented by this object, i.e. wait for it to finish. * @@ -1274,8 +1011,10 @@ namespace Threads void join() const { + // Make sure we actually have a task that we can wait for. AssertThrow(joinable(), ExcNoTask()); - task_descriptor->join(); + + task_data->wait(); } /** @@ -1293,8 +1032,7 @@ namespace Threads bool joinable() const { - return (task_descriptor != - std::shared_ptr>()); + return (task_data != nullptr); } @@ -1344,8 +1082,12 @@ namespace Threads typename internal::return_value::reference_type return_value() { - join(); - return task_descriptor->ret_val.get(); + // Make sure we actually have a task that we can wait for. + AssertThrow(joinable(), ExcNoTask()); + + // Then return the promised object. If necessary, wait for the promise to + // be set. + return task_data->get(); } @@ -1364,10 +1106,95 @@ namespace Threads //@} private: /** - * Shared pointer to the object representing the task. This makes sure that - * the object lives as long as there is at least one subscriber to it. + * A data structure that holds a std::future into which the task deposits + * its return value. Since one can only call std::future::get() once, + * we do so in the get() member function and then move the returned object + * into the `returned_object` member variable from where we can read it + * multiple times and from where it can also be moved away if it is not + * copyable. + */ + class TaskData + { + public: + TaskData(std::future &&future) + : future(std::move(future)) + , task_has_finished(false) + {} + + void + wait() + { + // If we have previously already moved the result, then we don't + // need a lock and can just return. + if (task_has_finished) + return; + + // Else, we need to go under a lock and try again. A different thread + // may have waited and finished the task since then, so we have to try + // a second time. (This is Schmidt's double-checking pattern.) + std::lock_guard lock(mutex); + if (task_has_finished) + return; + else + { + // Wait for the task to finish and then move its + // result. (We could have made the set_from() function + // that we call here wait for the future to be ready -- + // which happens implicitly when it calls future.get() -- + // but that would have required putting an explicit + // future.wait() into the implementation of + // internal::return_value::set_from(), which is a + // bit awkward: that class doesn't actually need to set + // anything, and so it looks odd to have the explicit call + // to future.wait() in the set_from() function. Avoid the + // issue by just explicitly calling future.wait() here.) + future.wait(); + returned_object.set_from(future); + + // Now we can safely set the flag and return. + task_has_finished = true; + } + } + + + + typename internal::return_value::reference_type + get() + { + wait(); + return returned_object.get(); + } + + private: + /** + * A mutex used to synchronize access to the data structures of this + * class. + */ + std::mutex mutex; + + /** + * The promise associated with the task that is represented by the current + * class. + */ + std::future future; + + /** + * A boolean indicating whether the task in question has finished. + */ + bool task_has_finished; + + /** + * The place where the returned value is moved to once the std::future + * has delivered. + */ + internal::return_value returned_object; + }; + + /** + * A pointer to a descriptor of the object that described the task + * and its return value. */ - std::shared_ptr> task_descriptor; + std::shared_ptr task_data; }; -- 2.39.5