thread_management.h

// ---------------------------------------------------------------------
//
// Copyright (C) 2000 - 2018 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 at
// the top level of the deal.II distribution.
//
// ---------------------------------------------------------------------

#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/template_constraints.h>
#include <deal.II/base/multithread_info.h>

#ifdef DEAL_II_WITH_THREADS
#  include <thread>
#  include <mutex>
#  include <condition_variable>
#endif

#include <iterator>
#include <vector>
#include <list>
#include <utility>
#include <functional>
#include <memory>
#include <tuple>


#ifdef DEAL_II_WITH_THREADS
#  ifdef DEAL_II_USE_MT_POSIX
#    include <pthread.h>
#  endif
#  include <tbb/task.h>
#  include <tbb/tbb_stddef.h>
#endif



DEAL_II_NAMESPACE_OPEN



namespace Threads
{
  class DummyThreadMutex
  {
  public:
    class ScopedLock
    {
    public:
      ScopedLock (DummyThreadMutex &) {}

      ~ScopedLock () {}
    };

    inline void acquire () const {}

    inline void release () const {}
  };



  class DummyThreadCondition
  {
  public:
    inline void signal () const {}

    inline void broadcast () const {}

    inline void wait (DummyThreadMutex &) const {}
  };



  class DummyBarrier
  {
  public:
    DummyBarrier (const unsigned int  count,
                  const char         *name = nullptr,
                  void               *arg  = nullptr);

    inline int wait () const
    {
      return 0;
    }

    inline void dump () const {}

    DeclException1 (ExcBarrierSizeNotUseful,
                    int,
                    << "In single-thread mode, barrier sizes other than 1 are not "
                    << "useful. You gave " << arg1 << ".");

  };


#ifdef DEAL_II_WITH_THREADS

  class Mutex
  {
  public:
    class ScopedLock
    {
    public:
      ScopedLock (Mutex &m) : mutex(m)
      {
        mutex.acquire();
      }

      ~ScopedLock ()
      {
        mutex.release ();
      }

    private:
      Mutex &mutex;
    };

    Mutex () = default;

    Mutex (const Mutex &)
      :
      mutex()
    {}


    Mutex &operator = (const Mutex &)
    {
      return *this;
    }


    inline void acquire ()
    {
      mutex.lock();
    }

    inline void release ()
    {
      mutex.unlock();
    }

  private:
    std::mutex mutex;

    friend class ConditionVariable;
  };


  class 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.mutex, std::adopt_lock);
      condition_variable.wait (lock);
    }

  private:
    std::condition_variable condition_variable;
  };


  class 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
  };


  typedef PosixThreadBarrier Barrier;

#else

  typedef DummyThreadMutex     Mutex;

  typedef DummyThreadCondition ConditionVariable;

  typedef DummyBarrier         Barrier;
#endif

}


namespace Threads
{

  unsigned int n_existing_threads ();

  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 ();
  }

}   // end declarations of 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)
  {
    typedef std::pair<ForwardIterator,ForwardIterator> IteratorPair;

    // 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;
  }
}

#endif // DOXYGEN

namespace Threads
{
  namespace internal
  {
    template <typename RT> struct return_value
    {
    private:
      RT value;
    public:
      typedef RT &reference_type;

      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:
      typedef RT &reference_type;

      inline return_value ()
        : value(nullptr)
      {}

      inline reference_type get () const
      {
        return *value;
      }

      inline void set (RT &v)
      {
        value = &v;
      }
    };


    template <> struct return_value<void>
    {
      typedef void reference_type;

      static inline void get () {}
    };
  }



  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
  {
#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
  }


  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);
      }
    };
  }



// ----------- 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())>
  {
    typedef decltype(function_object()) return_type;
    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)
  {
    return
      new_thread (std::function<RT ()>
                  (std::bind(fun_ptr,
                             internal::maybe_make_ref<Args>::act(args)...)));
  }



  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)
  {
    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)
  {
    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 ()
      {
        // 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 &);

      ~TaskDescriptor ();

      TaskDescriptor &
      operator = (const TaskDescriptor &);

      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);

      // in earlier versions of the TBB, task::spawn was a regular
      // member function; however, in later versions, it was converted
      // into a static function. we could always call it as a regular member
      // function of *task, but that appears to confuse the NVidia nvcc
      // compiler. consequently, the following work-around:
#if TBB_VERSION_MAJOR >= 4
      tbb::task::spawn (*worker);
#else
      task->spawn (*worker);
#endif
    }



    template <typename RT>
    TaskDescriptor<RT>::TaskDescriptor ()
      :
      task_is_done (false)
    {
      Assert (false, ExcInternalError());
    }



    template <typename RT>
    TaskDescriptor<RT>::TaskDescriptor (const TaskDescriptor &)
      :
      task_is_done (false)
    {
      // we shouldn't be getting here -- task descriptors
      // can't be copied
      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>
    TaskDescriptor<RT> &
    TaskDescriptor<RT>::operator = (const TaskDescriptor &)
    {
      // we shouldn't be getting here -- task descriptors
      // can't be copied
      Assert (false, ExcInternalError());
      return *this;
    }


    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

  }



  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 (const Task<RT> &t)
      :
      task_descriptor (t.task_descriptor)
    {}


    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())>
  {
    typedef decltype(function_object()) return_type;
    ::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)
  {
    return
      new_task (std::function<RT ()>
                (std::bind(fun_ptr,
                           internal::maybe_make_ref<Args>::act(args)...)));
  }



  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)
  {
    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)
  {
    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;
    }

    void join_all () const
    {
      for (typename std::list<Task<RT> >::const_iterator
           t=tasks.begin(); t!=tasks.end(); ++t)
        t->join ();
    }

  private:
    std::list<Task<RT> > tasks;
  };

}   // end of implementation of namespace Threads

//---------------------------------------------------------------------------
DEAL_II_NAMESPACE_CLOSE
// end of #ifndef dealii_thread_management_h
#endif
//---------------------------------------------------------------------------