thread_management.h

Go to the documentation of this file.

// ---------------------------------------------------------------------
//
// Copyright (C) 2000 - 2020 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 <condition_variable>
#  include <functional>
#  include <iterator>
#  include <list>
#  include <memory>
#  include <mutex>
#  include <tuple>
#  include <utility>
#  include <vector>
DEAL_II_DISABLE_EXTRA_DIAGNOSTICS
#  ifdef DEAL_II_WITH_THREADS
#    include <thread>
#    ifdef DEAL_II_USE_MT_POSIX
#      include <pthread.h>
#    endif
#    define TBB_SUPPRESS_DEPRECATED_MESSAGES 1
#    include <tbb/task.h>
#    undef TBB_SUPPRESS_DEPRECATED_MESSAGES
#    include <tbb/tbb_stddef.h>
#  endif
DEAL_II_ENABLE_EXTRA_DIAGNOSTICS
 
 
 
DEAL_II_NAMESPACE_OPEN
 
 
 
namespace Threads
{
  class Mutex : public std::mutex
  {
  public:
    using ScopedLock DEAL_II_DEPRECATED = std::lock_guard<std::mutex>;
 
    Mutex() = default;
 
    Mutex(const Mutex &)
      : std::mutex()
    {}
 
 
    Mutex &
    operator=(const Mutex &)
    {
      return *this;
    }
 
 
    DEAL_II_DEPRECATED
    inline void
    acquire()
    {
      this->lock();
    }
 
    DEAL_II_DEPRECATED
    inline void
    release()
    {
      this->unlock();
    }
  };
 
 
  class DEAL_II_DEPRECATED ConditionVariable
  {
  public:
    inline void
    signal()
    {
      condition_variable.notify_one();
    }
 
    inline void
    broadcast()
    {
      condition_variable.notify_all();
    }
 
    inline void
    wait(Mutex &mutex)
    {
      std::unique_lock<std::mutex> lock(mutex, std::adopt_lock);
      condition_variable.wait(lock);
    }
 
  private:
    std::condition_variable condition_variable;
  };
 
#  ifdef DEAL_II_USE_MT_POSIX
 
  class DEAL_II_DEPRECATED PosixThreadBarrier
  {
  public:
    PosixThreadBarrier(const unsigned int count,
                       const char *       name = nullptr,
                       void *             arg  = nullptr);
 
    ~PosixThreadBarrier();
 
    int
    wait();
 
  private:
#    ifndef DEAL_II_USE_MT_POSIX_NO_BARRIERS
    pthread_barrier_t barrier;
#    else
    unsigned int count;
#    endif
  };
 
 
  using Barrier DEAL_II_DEPRECATED = PosixThreadBarrier;
#  endif
} // namespace Threads
 
 
namespace Threads
{
  DEAL_II_DEPRECATED
  unsigned int
  n_existing_threads();
 
  DEAL_II_DEPRECATED
  unsigned int
  this_thread_id();
 
  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();
 
    void
    register_thread();
 
    void
    deregister_thread();
  } // 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);
      }
    };
 
 
    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;
      }
    };
 
 
    template <>
    struct return_value<void>
    {
      using reference_type = void;
 
      static inline void
      get()
      {}
    };
  } // 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
  {
#  ifdef DEAL_II_WITH_THREADS
 
    template <typename RT>
    struct ThreadDescriptor
    {
      std::thread thread;
 
      std::shared_ptr<return_value<RT>> ret_val;
 
      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;
 
        Mutex::ScopedLock 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
        internal::register_thread();
        try
          {
            call(function, *ret_val);
          }
        catch (const std::exception &exc)
          {
            internal::handle_std_exception(exc);
          }
        catch (...)
          {
            internal::handle_unknown_exception();
          }
        internal::deregister_thread();
      }
    };
 
#  else
 
    template <typename RT>
    struct ThreadDescriptor
    {
      std::shared_ptr<return_value<RT>> ret_val;
 
      void
      start(const std::function<RT()> &function)
      {
        ret_val = std::make_shared<return_value<RT>>();
        call(function, *ret_val);
      }
 
      void
      join()
      {}
    };
 
#  endif
  } // namespace internal
 
 
  template <typename RT = void>
  class Thread
  {
  public:
    Thread(const std::function<RT()> &function)
      : thread_descriptor(new internal::ThreadDescriptor<RT>())
    {
      // in a second step, start the thread.
      thread_descriptor->start(function);
    }
 
    Thread() = default;
 
    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>
  inline Thread<RT>
  new_thread(const std::function<RT()> &function)
  {
    return Thread<RT>(function);
  }
 
 
 
  template <typename FunctionObjectType>
  inline auto
  new_thread(FunctionObjectType function_object)
    -> Thread<decltype(function_object())>
  {
    using return_type = decltype(function_object());
    return Thread<return_type>(std::function<return_type()>(function_object));
  }
 
 
 
  template <typename RT, typename... Args>
  inline Thread<RT>
  new_thread(RT (*fun_ptr)(Args...), typename identity<Args>::type... args)
  {
    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); });
  }
 
 
 
  template <typename RT, typename C, typename... Args>
  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)...)));
  }
 
#  ifndef DEAL_II_CONST_MEMBER_DEDUCTION_BUG
 
  template <typename RT, typename C, typename... Args>
  inline Thread<RT>
  new_thread(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_thread(std::function<RT()>(std::bind(
      fun_ptr, std::cref(c), internal::maybe_make_ref<Args>::act(args)...)));
  }
#  endif
 
  // ------------------------ ThreadGroup -------------------------------------
 
  template <typename RT = void>
  class ThreadGroup
  {
  public:
    ThreadGroup &
    operator+=(const Thread<RT> &t)
    {
      threads.push_back(t);
      return *this;
    }
 
    void
    join_all() const
    {
      for (typename std::list<Thread<RT>>::const_iterator t = threads.begin();
           t != threads.end();
           ++t)
        t->join();
    }
 
  private:
    std::list<Thread<RT>> threads;
  };
 
 
  template <typename>
  class Task;
 
 
  namespace internal
  {
#  ifdef DEAL_II_WITH_THREADS
 
    template <typename>
    struct TaskDescriptor;
 
    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;
      }
 
      TaskDescriptor<RT> &task_descriptor;
    };
 
    template <typename RT>
    struct TaskDescriptor
    {
    private:
      std::function<RT()> function;
 
      tbb::task *task;
 
      return_value<RT> ret_val;
 
      bool task_is_done;
 
    public:
      TaskDescriptor(const std::function<RT()> &function);
 
      TaskDescriptor();
 
      TaskDescriptor(const TaskDescriptor &) = delete;
 
      ~TaskDescriptor();
 
      TaskDescriptor &
      operator=(const TaskDescriptor &) = delete;
 
      void
      queue_task();
 
      void
      join();
 
 
      template <typename>
      friend struct TaskEntryPoint;
      friend class ::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
 
    template <typename RT>
    struct TaskDescriptor
    {
      return_value<RT> ret_val;
 
      TaskDescriptor(const std::function<RT()> &function)
      {
        call(function, ret_val);
      }
 
      static void
      join()
      {}
 
      static void
      queue_task()
      {}
    };
 
#  endif
 
  } // namespace internal
 
 
 
  template <typename RT = void>
  class Task
  {
  public:
    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() = default;
 
 
 
    void
    join() const
    {
      AssertThrow(joinable(), ExcNoTask());
      task_descriptor->join();
    }
 
    bool
    joinable() const
    {
      return (task_descriptor !=
              std::shared_ptr<internal::TaskDescriptor<RT>>());
    }
 
 
    typename internal::return_value<RT>::reference_type
    return_value()
    {
      join();
      return task_descriptor->ret_val.get();
    }
 
 
    bool
    operator==(const Task &t) const
    {
      AssertThrow(joinable(), ExcNoTask());
      return task_descriptor == t.task_descriptor;
    }
 
    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:
    std::shared_ptr<internal::TaskDescriptor<RT>> task_descriptor;
  };
 
 
 
  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 Task<return_type>(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)...)));
  }
 
#  ifndef DEAL_II_CONST_MEMBER_DEDUCTION_BUG
 
  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)...)));
  }
#  endif
 
 
  // ------------------------ 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
//---------------------------------------------------------------------------