# include <condition_variable>
# include <functional>
+# include <future>
# include <iterator>
# include <list>
# include <memory>
{
value = std::move(v);
}
+
+ inline void
+ set_from(std::future<RT> &v)
+ {
+ value = std::move(v.get());
+ }
};
{
value = &v;
}
+
+ inline void
+ set_from(std::future<RT &> &v)
+ {
+ value = &v.get();
+ }
};
static inline void
get()
{}
+
+
+ inline void
+ set_from(std::future<void> &)
+ {}
};
} // namespace internal
};
- 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
* [`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
*/
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
*/
Task() = default;
-
-
/**
* Join the task represented by this object, i.e. wait for it to finish.
*
void
join() const
{
+ // Make sure we actually have a task that we can wait for.
AssertThrow(joinable(), ExcNoTask());
- task_descriptor->join();
+
+ task_data->wait();
}
/**
bool
joinable() const
{
- return (task_descriptor !=
- std::shared_ptr<internal::TaskDescriptor<RT>>());
+ return (task_data != nullptr);
}
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();
}
//@}
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;
};
projects / dealii.git / commitdiff