]> https://gitweb.dealii.org/ - dealii.git/commitdiff
Rewrite the Threads::Task class using C++11 features.
authorWolfgang Bangerth <bangerth@colostate.edu>
Wed, 27 May 2020 00:53:36 +0000 (18:53 -0600)
committerWolfgang Bangerth <bangerth@colostate.edu>
Wed, 27 May 2020 18:47:12 +0000 (12:47 -0600)
include/deal.II/base/thread_management.h

index dfa6751af57f95a56bf094592a33603a39bb015d..73bc007643f848059cc6a2c9a304ddaac315262f 100644 (file)
@@ -26,6 +26,7 @@
 
 #  include <condition_variable>
 #  include <functional>
+#  include <future>
 #  include <iterator>
 #  include <list>
 #  include <memory>
@@ -281,6 +282,12 @@ namespace Threads
       {
         value = std::move(v);
       }
+
+      inline void
+      set_from(std::future<RT> &v)
+      {
+        value = std::move(v.get());
+      }
     };
 
 
@@ -317,6 +324,12 @@ namespace Threads
       {
         value = &v;
       }
+
+      inline void
+      set_from(std::future<RT &> &v)
+      {
+        value = &v.get();
+      }
     };
 
 
@@ -336,6 +349,11 @@ namespace Threads
       static inline void
       get()
       {}
+
+
+      inline void
+      set_from(std::future<void> &)
+      {}
     };
   } // namespace internal
 
@@ -924,284 +942,6 @@ namespace Threads
   };
 
 
-  template <typename>
-  class Task;
-
-
-  namespace internal
-  {
-#  ifdef DEAL_II_WITH_THREADS
-
-    template <typename>
-    struct TaskDescriptor;
-
-    /**
-     * The task class for TBB that is used by the TaskDescriptor class.
-     */
-    template <typename RT>
-    struct TaskEntryPoint : public tbb::task
-    {
-      TaskEntryPoint(TaskDescriptor<RT> &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<RT> &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 <typename RT>
-    struct TaskDescriptor
-    {
-    private:
-      /**
-       * The function and its arguments that are to be run on the task.
-       */
-      std::function<RT()> 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<RT> 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<RT()> &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 <typename>
-      friend struct TaskEntryPoint;
-      friend class dealii::Threads::Task<RT>;
-    };
-
-
-
-    template <typename RT>
-    inline TaskDescriptor<RT>::TaskDescriptor(
-      const std::function<RT()> &function)
-      : function(function)
-      , task(nullptr)
-      , task_is_done(false)
-    {}
-
-
-    template <typename RT>
-    inline void
-    TaskDescriptor<RT>::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<RT>(*this);
-
-      tbb::task::spawn(*worker);
-    }
-
-
-
-    template <typename RT>
-    TaskDescriptor<RT>::TaskDescriptor()
-      : task_is_done(false)
-    {
-      Assert(false, ExcInternalError());
-    }
-
-
-
-    template <typename RT>
-    inline TaskDescriptor<RT>::~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 <typename RT>
-    inline void
-    TaskDescriptor<RT>::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 <typename RT>
-    struct TaskDescriptor
-    {
-      /**
-       * A place where the task will deposit its return value.
-       */
-      return_value<RT> ret_val;
-
-      /**
-       * Constructor. Call the given function and emplace the return value
-       * into the slot reserved for this purpose.
-       */
-      TaskDescriptor(const std::function<RT()> &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<RT()> &function_object)
     {
-      // create a task descriptor and tell it to queue itself up with
-      // the scheduling system
-      task_descriptor =
-        std::make_shared<internal::TaskDescriptor<RT>>(function_object);
-      task_descriptor->queue_task();
+      task_data = std::make_shared<TaskData>(
+        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<internal::TaskDescriptor<RT>>());
+      return (task_data != nullptr);
     }
 
 
@@ -1344,8 +1082,12 @@ namespace Threads
     typename internal::return_value<RT>::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<RT> &&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<std::mutex> 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<void>::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<RT>::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<RT> 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<RT> returned_object;
+    };
+
+    /**
+     * A pointer to a descriptor of the object that described the task
+     * and its return value.
      */
-    std::shared_ptr<internal::TaskDescriptor<RT>> task_descriptor;
+    std::shared_ptr<TaskData> task_data;
   };
 
 

In the beginning the Universe was created. This has made a lot of people very angry and has been widely regarded as a bad move.

Douglas Adams


Typeset in Trocchi and Trocchi Bold Sans Serif.