thread_management.h

Go to the documentation of this file.

// ---------------------------------------------------------------------
//
// Copyright (C) 2000 - 2022 by the deal.II authors
//
// This file is part of the deal.II library.
//
// The deal.II library is free software; you can use it, redistribute
// it, and/or modify it under the terms of the GNU Lesser General
// Public License as published by the Free Software Foundation; either
// version 2.1 of the License, or (at your option) any later version.
// The full text of the license can be found in the file LICENSE.md at
// the top level directory of deal.II.
//
// ---------------------------------------------------------------------
 
#ifndef dealii_thread_management_h
#  define dealii_thread_management_h
 
 
#  include <deal.II/base/config.h>
 
#  include <deal.II/base/exceptions.h>
#  include <deal.II/base/multithread_info.h>
#  include <deal.II/base/std_cxx17/tuple.h>
#  include <deal.II/base/template_constraints.h>
 
#  include <atomic>
#  include <functional>
#  include <future>
#  include <list>
#  include <memory>
#  include <mutex>
#  include <thread>
#  include <utility>
#  include <vector>
 
#  ifdef DEAL_II_WITH_TBB
DEAL_II_DISABLE_EXTRA_DIAGNOSTICS
#    include <tbb/task_group.h>
DEAL_II_ENABLE_EXTRA_DIAGNOSTICS
#  endif
 
DEAL_II_NAMESPACE_OPEN
 
 
 
namespace Threads
{
  class Mutex : public std::mutex
  {
  public:
    Mutex() = default;
 
    Mutex(const Mutex &)
      : std::mutex()
    {}
 
    Mutex &
    operator=(const Mutex &)
    {
      return *this;
    }
  };
} // namespace Threads
 
 
namespace Threads
{
  template <typename ForwardIterator>
  std::vector<std::pair<ForwardIterator, ForwardIterator>>
  split_range(const ForwardIterator &begin,
              const ForwardIterator &end,
              const unsigned int     n_intervals);
 
  std::vector<std::pair<unsigned int, unsigned int>>
  split_interval(const unsigned int begin,
                 const unsigned int end,
                 const unsigned int n_intervals);
 
  namespace internal
  {
    [[noreturn]] void
    handle_std_exception(const std::exception &exc);
 
    [[noreturn]] void
    handle_unknown_exception();
  } // namespace internal
 
} // namespace Threads
 
/* ----------- implementation of functions in namespace Threads ---------- */
#  ifndef DOXYGEN
namespace Threads
{
  template <typename ForwardIterator>
  std::vector<std::pair<ForwardIterator, ForwardIterator>>
  split_range(const ForwardIterator &begin,
              const ForwardIterator &end,
              const unsigned int     n_intervals)
  {
    using IteratorPair = std::pair<ForwardIterator, ForwardIterator>;
 
    // in non-multithreaded mode, we often have the case that this
    // function is called with n_intervals==1, so have a shortcut here
    // to handle that case efficiently
 
    if (n_intervals == 1)
      return (std::vector<IteratorPair>(1, IteratorPair(begin, end)));
 
    // if more than one interval requested, do the full work
    const unsigned int n_elements              = std::distance(begin, end);
    const unsigned int n_elements_per_interval = n_elements / n_intervals;
    const unsigned int residual                = n_elements % n_intervals;
 
    std::vector<IteratorPair> return_values(n_intervals);
 
    return_values[0].first = begin;
    for (unsigned int i = 0; i < n_intervals; ++i)
      {
        if (i != n_intervals - 1)
          {
            return_values[i].second = return_values[i].first;
            // note: the cast is performed to avoid a warning of gcc
            // that in the library `dist>=0' is checked (dist has a
            // template type, which here is unsigned if no cast is
            // performed)
            std::advance(return_values[i].second,
                         static_cast<signed int>(n_elements_per_interval));
            // distribute residual in division equally among the first
            // few subintervals
            if (i < residual)
              ++return_values[i].second;
 
            return_values[i + 1].first = return_values[i].second;
          }
        else
          return_values[i].second = end;
      }
    return return_values;
  }
} // namespace Threads
 
#  endif // DOXYGEN
 
namespace Threads
{
  namespace internal
  {
    template <typename RT>
    struct return_value
    {
    private:
      RT value;
 
    public:
      using reference_type = RT &;
 
      inline return_value()
        : value()
      {}
 
      inline reference_type
      get()
      {
        return value;
      }
 
      inline void
      set(RT &&v)
      {
        value = std::move(v);
      }
 
      inline void
      set_from(std::future<RT> &v)
      {
        value = std::move(v.get());
      }
    };
 
 
    template <typename RT>
    struct return_value<RT &>
    {
    private:
      RT *value;
 
    public:
      using reference_type = RT &;
 
      inline return_value()
        : value(nullptr)
      {}
 
      inline reference_type
      get() const
      {
        return *value;
      }
 
      inline void
      set(RT &v)
      {
        value = &v;
      }
 
      inline void
      set_from(std::future<RT &> &v)
      {
        value = &v.get();
      }
    };
 
 
    template <>
    struct return_value<void>
    {
      using reference_type = void;
 
      static inline void
      get()
      {}
 
 
      inline void
      set_from(std::future<void> &)
      {}
    };
  } // namespace internal
 
 
 
  namespace internal
  {
    template <typename RT>
    inline void
    call(const std::function<RT()> & function,
         internal::return_value<RT> &ret_val)
    {
      ret_val.set(function());
    }
 
 
    inline void
    call(const std::function<void()> &function, internal::return_value<void> &)
    {
      function();
    }
  } // namespace internal
 
 
 
  namespace internal
  {
    template <typename RT>
    struct ThreadDescriptor
    {
      std::thread thread;
 
      std::shared_ptr<return_value<RT>> ret_val;
 
      std::atomic<bool> thread_is_active;
 
      Mutex thread_is_active_mutex;
 
      ThreadDescriptor()
        : thread_is_active(false)
      {}
 
      ~ThreadDescriptor()
      {
        if (!thread_is_active)
          return;
        thread.detach();
        thread_is_active = false;
      }
 
      void
      start(const std::function<RT()> &function)
      {
        thread_is_active = true;
        ret_val          = std::make_shared<return_value<RT>>();
        thread           = std::thread(thread_entry_point, function, ret_val);
      }
 
 
      void
      join()
      {
        // see if the thread hasn't been joined yet. if it has, then
        // join() is a no-op. use schmidt's double-checking strategy
        // to use the mutex only when necessary
        if (thread_is_active == false)
          return;
 
        std::lock_guard<std::mutex> lock(thread_is_active_mutex);
        if (thread_is_active == true)
          {
            Assert(thread.joinable(), ExcInternalError());
            thread.join();
            thread_is_active = false;
          }
      }
 
    private:
      static void
      thread_entry_point(const std::function<RT()> &       function,
                         std::shared_ptr<return_value<RT>> ret_val)
      {
        // call the function in question. since an exception that is
        // thrown from one of the called functions will not propagate
        // to the main thread, it will kill the program if not treated
        // here before we return to the operating system's thread
        // library
        try
          {
            call(function, *ret_val);
          }
        catch (const std::exception &exc)
          {
            internal::handle_std_exception(exc);
          }
        catch (...)
          {
            internal::handle_unknown_exception();
          }
      }
    };
  } // namespace internal
 
 
  template <typename RT = void>
  class Thread
  {
  public:
    DEAL_II_DEPRECATED
    Thread(const std::function<RT()> &function)
      : thread_descriptor(new internal::ThreadDescriptor<RT>())
    {
      // in a second step, start the thread.
      thread_descriptor->start(function);
    }
 
    DEAL_II_DEPRECATED
    Thread() = default;
 
    DEAL_II_DEPRECATED
    Thread(const Thread<RT> &t)
      : thread_descriptor(t.thread_descriptor)
    {}
 
    void
    join() const
    {
      if (thread_descriptor)
        thread_descriptor->join();
    }
 
    typename internal::return_value<RT>::reference_type
    return_value()
    {
      join();
      return thread_descriptor->ret_val->get();
    }
 
    bool
    valid() const
    {
      return static_cast<bool>(thread_descriptor);
    }
 
 
    bool
    operator==(const Thread &t) const
    {
      return thread_descriptor == t.thread_descriptor;
    }
 
  private:
    std::shared_ptr<internal::ThreadDescriptor<RT>> thread_descriptor;
  };
 
 
  namespace internal
  {
    template <typename T>
    struct maybe_make_ref
    {
      static T
      act(T &t)
      {
        return t;
      }
    };
 
 
 
    template <typename T>
    struct maybe_make_ref<T &>
    {
      static std::reference_wrapper<T>
      act(T &t)
      {
        return std::ref(t);
      }
    };
  } // namespace internal
 
 
 
  // ----------- thread starters for functions not taking any parameters
 
  template <typename RT>
  DEAL_II_DEPRECATED inline Thread<RT>
  new_thread(const std::function<RT()> &function)
  {
    // Here and below we need to disable deprecation warnings for calling the
    // constructor in this function - as this function itself is deprecated
    // these warnings are not helpful. This problem only appears in some
    // configurations (e.g., Debian 11 with GCC-10).
    DEAL_II_DISABLE_EXTRA_DIAGNOSTICS
    return Thread<RT>(function);
    DEAL_II_ENABLE_EXTRA_DIAGNOSTICS
  }
 
 
 
  template <typename FunctionObjectType>
  DEAL_II_DEPRECATED inline auto
  new_thread(FunctionObjectType function_object)
    -> Thread<decltype(function_object())>
  {
    // See the comment in the first new_thread() implementation
    DEAL_II_DISABLE_EXTRA_DIAGNOSTICS
    using return_type = decltype(function_object());
    return Thread<return_type>(std::function<return_type()>(function_object));
    DEAL_II_ENABLE_EXTRA_DIAGNOSTICS
  }
 
 
 
  template <typename RT, typename... Args>
  DEAL_II_DEPRECATED inline Thread<RT>
  new_thread(RT (*fun_ptr)(Args...), typename identity<Args>::type... args)
  {
    // See the comment in the first new_thread() implementation
    DEAL_II_DISABLE_EXTRA_DIAGNOSTICS
    auto dummy = std::make_tuple(internal::maybe_make_ref<Args>::act(args)...);
    return new_thread(
      [dummy, fun_ptr]() -> RT { return std_cxx17::apply(fun_ptr, dummy); });
    DEAL_II_ENABLE_EXTRA_DIAGNOSTICS
  }
 
 
 
  template <typename RT, typename C, typename... Args>
  DEAL_II_DEPRECATED inline Thread<RT>
  new_thread(RT (C::*fun_ptr)(Args...),
             typename identity<C>::type &c,
             typename identity<Args>::type... args)
  {
    // NOLINTNEXTLINE(modernize-avoid-bind) silence clang-tidy
    return new_thread(std::function<RT()>(std::bind(
      fun_ptr, std::ref(c), internal::maybe_make_ref<Args>::act(args)...)));
  }
 
  template <typename RT, typename C, typename... Args>
  DEAL_II_DEPRECATED inline Thread<RT>
  new_thread(RT (C::*fun_ptr)(Args...) const,
             typename identity<const C>::type &c,
             typename identity<Args>::type... args)
  {
    // See the comment in the first new_thread() implementation
    DEAL_II_DISABLE_EXTRA_DIAGNOSTICS
    // NOLINTNEXTLINE(modernize-avoid-bind) silence clang-tidy
    return new_thread(std::function<RT()>(std::bind(
      fun_ptr, std::cref(c), internal::maybe_make_ref<Args>::act(args)...)));
    DEAL_II_ENABLE_EXTRA_DIAGNOSTICS
  }
 
  // ------------------------ ThreadGroup -------------------------------------
 
  template <typename RT = void>
  class DEAL_II_DEPRECATED ThreadGroup
  {
  public:
    ThreadGroup &
    operator+=(const Thread<RT> &t)
    {
      threads.push_back(t);
      return *this;
    }
 
    void
    join_all() const
    {
      for (auto &t : threads)
        t.join();
    }
 
  private:
    std::list<Thread<RT>> threads;
  };
 
 
  namespace internal
  {
    template <typename RT, typename Function>
    void
    evaluate_and_set_promise(Function &function, std::promise<RT> &promise)
    {
      promise.set_value(function());
    }
 
 
    template <typename Function>
    void
    evaluate_and_set_promise(Function &function, std::promise<void> &promise)
    {
      function();
      promise.set_value();
    }
  } // namespace internal
 
 
 
  template <typename RT = void>
  class Task
  {
  public:
    Task(const std::function<RT()> &function_object)
    {
      if (MultithreadInfo::n_threads() > 1)
        {
#  ifdef DEAL_II_WITH_TBB
          // Create a promise object and from it extract a future that
          // we can use to refer to the outcome of the task. For reasons
          // explained below, we can't just create a std::promise object,
          // but have to make do with a pointer to such an object.
          std::unique_ptr<std::promise<RT>> promise =
            std::make_unique<std::promise<RT>>();
          task_data =
            std::make_shared<TaskData>(std::move(promise->get_future()));
 
          // Then start the task, using a task_group object (for just this one
          // task) that is associated with the TaskData object. Note that we
          // have to *copy* the function object being executed so that it is
          // guaranteed to live on the called thread as well -- the copying is
          // facilitated by capturing the 'function_object' variable by value.
          //
          // We also have to *move* the promise object into the new task's
          // memory space because promises can not be copied and we can't refer
          // to it by reference because it's a local variable of the current
          // (surrounding) function that may go out of scope before the promise
          // is ultimately set. This leads to a conundrum: if we had just
          // declared 'promise' as an object of type std::promise, then we could
          // capture it in the lambda function via
          //     [..., promise=std::move(promise)]() {...}
          // and set the promise in the body of the lambda. But setting a
          // promise is a non-const operation on the promise, and so we would
          // actually have to declare the lambda function as 'mutable' because
          // by default, lambda captures are 'const'. That is, we would have
          // to write
          //     [..., promise=std::move(promise)]() mutable {...}
          // But this leads to other problems: It turns out that the
          // tbb::task_group::run() function cannot take mutable lambdas as
          // argument :-(
          //
          // We work around this issue by not declaring the 'promise' variable
          // as an object of type std::promise, but as a pointer to such an
          // object. This pointer we can move, and the *pointer* itself can
          // be 'const' (meaning we can leave the lambda as non-mutable)
          // even though we modify the object *pointed to*. One would think
          // that a std::unique_ptr would be the right choice for this, but
          // that's not true: the resulting lambda function can then be
          // non-mutable, but the lambda function object is not copyable
          // and at least some TBB variants require that as well. So
          // instead we move the std::unique_ptr used above into a
          // std::shared_ptr to be stored within the lambda function object.
          task_data->task_group.run(
            [function_object,
             promise =
               std::shared_ptr<std::promise<RT>>(std::move(promise))]() {
              try
                {
                  internal::evaluate_and_set_promise(function_object, *promise);
                }
              catch (...)
                {
                  try
                    {
                      // store anything thrown in the promise
                      promise->set_exception(std::current_exception());
                    }
                  catch (...)
                    {
                      // set_exception() may throw too. But ignore this on
                      // the task.
                    }
                }
            });
 
#  else
          // If no threading library is supported, just fall back onto C++11
          // facilities. The problem with this is that the standard does
          // not actually say what std::async should do. The first
          // argument to that function can be std::launch::async or
          // std::launch::deferred, or both. The *intent* of the standard's
          // authors was probably that if one sets it to
          //   std::launch::async | std::launch::deferred,
          // that the task is run in a thread pool. But at least as of
          // 2021, GCC doesn't do that: It just runs it on a new thread.
          // If one chooses std::launch::deferred, it runs the task on
          // the same thread but only when one calls join() on the task's
          // std::future object. In the former case, this leads to
          // oversubscription, in the latter case to undersubscription of
          // resources. We choose oversubscription here.
          //
          // The issue illustrates why relying on external libraries
          // with task schedulers is the way to go.
          task_data = std::make_shared<TaskData>(
            std::async(std::launch::async | std::launch::deferred,
                       function_object));
#  endif
        }
      else
        {
          // Only one thread allowed. So let the task run to completion
          // and just emplace a 'ready' future.
          //
          // The design of std::promise/std::future is unclear, but it
          // seems that the intent is to obtain the std::future before
          // we set the std::promise. So create the TaskData object at
          // the top and then run the task and set the returned
          // value. Since everything here happens sequentially, it
          // really doesn't matter in which order all of this is
          // happening.
          std::promise<RT> promise;
          task_data = std::make_shared<TaskData>(promise.get_future());
          try
            {
              internal::evaluate_and_set_promise(function_object, promise);
            }
          catch (...)
            {
              try
                {
                  // store anything thrown in the promise
                  promise.set_exception(std::current_exception());
                }
              catch (...)
                {
                  // set_exception() may throw too. But ignore this on
                  // the task.
                }
            }
        }
    }
 
    Task() = default;
 
    void
    join() const
    {
      // Make sure we actually have a task that we can wait for.
      AssertThrow(joinable(), ExcNoTask());
 
      task_data->wait();
    }
 
    bool
    joinable() const
    {
      return (task_data != nullptr);
    }
 
 
    typename internal::return_value<RT>::reference_type
    return_value()
    {
      // 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();
    }
 
 
    DeclExceptionMsg(ExcNoTask,
                     "The current object is not associated with a task that "
                     "can be joined. It may have been detached, or you "
                     "may have already joined it in the past.");
  private:
    class TaskData
    {
    public:
      TaskData(std::future<RT> &&future) noexcept
        : future(std::move(future))
        , task_has_finished(false)
      {}
 
      TaskData(const TaskData &) = delete;
 
      TaskData(TaskData &&) = delete;
 
      TaskData &
      operator=(const TaskData &) = delete;
 
      TaskData &
      operator=(TaskData &&) = delete;
 
      ~TaskData() noexcept
      {
        // Explicitly wait for the results to be ready. This class stores
        // a std::future object, and we could just let the compiler generate
        // the destructor which would then call the destructor of std::future
        // which *may* block until the future is ready. As explained in
        //   https://en.cppreference.com/w/cpp/thread/future/~future
        // this is only a *may*, not a *must*. (The standard does not
        // appear to say anything about it at all.) As a consequence,
        // let's be explicit about waiting.
        //
        // One of the corner cases we have to worry about is that if a task
        // ends by throwing an exception, then wait() will re-throw that
        // exception on the thread that calls it, the first time around
        // someone calls wait() (or the return_value() function of the
        // surrounding class). So if we get to this constructor and an exception
        // is thrown by wait(), then that means that the last Task object
        // referring to a task is going out of scope with nobody having
        // ever checked the return value of the task itself. In that case,
        // one could argue that they would also not have cared about whether
        // an exception is thrown, and that we should simply ignore the
        // exception. This is what we do here. It is also the simplest solution,
        // because we don't know what one should do with the exception to begin
        // with: destructors aren't allowed to throw exceptions, so we can't
        // just rethrow it here if one had been triggered.
        try
          {
            wait();
          }
        catch (...)
          {}
      }
 
      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
          {
#  ifdef DEAL_II_WITH_TBB
            // If we build on the TBB, then we can't just wait for the
            // std::future object to get ready. Apparently the TBB happily
            // enqueues a task into an arena and then just sits on it without
            // ever executing it unless someone expresses an interest in the
            // task. The way to avoid this is to add the task to a
            // tbb::task_group, and then here wait for the single task
            // associated with that task group.
            task_group.wait();
#  endif
 
            // 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:
      std::mutex mutex;
 
      std::future<RT> future;
 
      std::atomic<bool> task_has_finished;
 
      internal::return_value<RT> returned_object;
 
#  ifdef DEAL_II_WITH_TBB
      tbb::task_group task_group;
 
      friend class Task<RT>;
#  endif
    };
 
    std::shared_ptr<TaskData> task_data;
  };
 
 
 
  template <typename RT>
  inline Task<RT>
  new_task(const std::function<RT()> &function)
  {
    return Task<RT>(function);
  }
 
 
 
  template <typename FunctionObjectType>
  inline auto
  new_task(FunctionObjectType function_object)
    -> Task<decltype(function_object())>
  {
    using return_type = decltype(function_object());
    ::MultithreadInfo::initialize_multithreading();
    return new_task(std::function<return_type()>(function_object));
  }
 
 
 
  template <typename RT, typename... Args>
  inline Task<RT>
  new_task(RT (*fun_ptr)(Args...), typename identity<Args>::type... args)
  {
    auto dummy = std::make_tuple(internal::maybe_make_ref<Args>::act(args)...);
    return new_task(
      [dummy, fun_ptr]() -> RT { return std_cxx17::apply(fun_ptr, dummy); });
  }
 
 
 
  template <typename RT, typename C, typename... Args>
  inline Task<RT>
  new_task(RT (C::*fun_ptr)(Args...),
           typename identity<C>::type &c,
           typename identity<Args>::type... args)
  {
    // NOLINTNEXTLINE(modernize-avoid-bind) silence clang-tidy
    return new_task(std::function<RT()>(std::bind(
      fun_ptr, std::ref(c), internal::maybe_make_ref<Args>::act(args)...)));
  }
 
  template <typename RT, typename C, typename... Args>
  inline Task<RT>
  new_task(RT (C::*fun_ptr)(Args...) const,
           typename identity<const C>::type &c,
           typename identity<Args>::type... args)
  {
    // NOLINTNEXTLINE(modernize-avoid-bind) silence clang-tidy
    return new_task(std::function<RT()>(std::bind(
      fun_ptr, std::cref(c), internal::maybe_make_ref<Args>::act(args)...)));
  }
 
 
  // ------------------------ TaskGroup -------------------------------------
 
  template <typename RT = void>
  class TaskGroup
  {
  public:
    TaskGroup &
    operator+=(const Task<RT> &t)
    {
      tasks.push_back(t);
      return *this;
    }
 
 
    std::size_t
    size() const
    {
      return tasks.size();
    }
 
 
    void
    join_all() const
    {
      for (auto &t : tasks)
        t.join();
    }
 
  private:
    std::list<Task<RT>> tasks;
  };
 
} // namespace Threads
 
//---------------------------------------------------------------------------
DEAL_II_NAMESPACE_CLOSE
// end of #ifndef dealii_thread_management_h
#endif
//---------------------------------------------------------------------------