thread.h

Go to the documentation of this file.

/* Copyright (c) 2008-2022 the MRtrix3 contributors.
 *
 * This Source Code Form is subject to the terms of the Mozilla Public
 * License, v. 2.0. If a copy of the MPL was not distributed with this
 * file, You can obtain one at http://mozilla.org/MPL/2.0/.
 *
 * Covered Software is provided under this License on an "as is"
 * basis, without warranty of any kind, either expressed, implied, or
 * statutory, including, without limitation, warranties that the
 * Covered Software is free of defects, merchantable, fit for a
 * particular purpose or non-infringing.
 * See the Mozilla Public License v. 2.0 for more details.
 *
 * For more details, see http://www.mrtrix.org/.
 */
 
#ifndef __mrtrix_thread_h__
#define __mrtrix_thread_h__
 
#include <thread>
#include <future>
#include <mutex>
 
#include "debug.h"
#include "mrtrix.h"
#include "exception.h"
 
namespace MR
{
  namespace Thread
  {
 
    class __Backend { NOMEMALIGN
      public:
        __Backend();
        ~__Backend();
 
        static void register_thread () {
          std::lock_guard<std::mutex> lock (mutex);
          if (!backend)
            backend = new __Backend;
          ++backend->refcount;
        }
        static void unregister_thread () {
          assert (backend);
          std::lock_guard<std::mutex> lock (mutex);
          if (!(--backend->refcount)) {
            delete backend;
            backend = nullptr;
          }
        }
 
        static bool valid() { return backend; }
 
        static void thread_print_func (const std::string& msg);
        static void thread_report_to_user_func (const std::string& msg, int type);
 
        static void (*previous_print_func) (const std::string& msg);
        static void (*previous_report_to_user_func) (const std::string& msg, int type);
 
      protected:
        size_t refcount;
 
        static __Backend* backend;
        static std::mutex mutex;
    };
 
 
    namespace {
 
      class __thread_base { NOMEMALIGN
        public:
          __thread_base (const std::string& name = "unnamed") : name (name) { __Backend::register_thread(); }
          __thread_base (const __thread_base&) = delete;
          __thread_base (__thread_base&&) = default;
          ~__thread_base () { __Backend::unregister_thread(); }
 
 
        protected:
          const std::string name;
      };
 
 
      class __single_thread : public __thread_base { NOMEMALIGN
        public:
          template <class Functor>
            __single_thread (Functor&& functor, const std::string& name = "unnamed") :
            __thread_base (name) {
              DEBUG ("launching thread \"" + name + "\"...");
              using F = typename std::remove_reference<Functor>::type;
              thread = std::async (std::launch::async, &F::execute, &functor);
            }
          __single_thread (const __single_thread&) = delete;
          __single_thread (__single_thread&&) = default;
 
          bool finished () const
          {
            return thread.wait_for(std::chrono::microseconds(0)) == std::future_status::ready;
          }
 
          void wait () noexcept (false) {
            DEBUG ("waiting for completion of thread \"" + name + "\"...");
            thread.get();
            DEBUG ("thread \"" + name + "\" completed OK");
          }
 
          ~__single_thread () {
            if (thread.valid()) {
              try { wait(); }
              catch (Exception& E) { E.display(); }
            }
          }
 
        protected:
          std::future<void> thread;
      };
 
 
      template <class Functor>
        class __multi_thread : public __thread_base { NOMEMALIGN
          public:
            __multi_thread (Functor& functor, size_t nthreads, const std::string& name = "unnamed") :
              __thread_base (name), functors ( (nthreads>0 ? nthreads-1 : 0), functor) {
                DEBUG ("launching " + str (nthreads) + " threads \"" + name + "\"...");
                using F = typename std::remove_reference<Functor>::type;
                threads.reserve (nthreads);
                for (auto& f : functors)
                  threads.push_back (std::async (std::launch::async, &F::execute, &f));
                threads.push_back (std::async (std::launch::async, &F::execute, &functor));
              }
 
            __multi_thread (const __multi_thread&) = delete;
            __multi_thread (__multi_thread&&) = default;
 
            void wait () noexcept (false) {
              DEBUG ("waiting for completion of threads \"" + name + "\"...");
              bool exception_thrown = false;
              for (auto& t : threads) {
                if (!t.valid())
                  continue;
                try { t.get(); }
                catch (Exception& E) {
                  exception_thrown = true;
                  E.display();
                }
              }
              if (exception_thrown)
                throw Exception ("exception thrown from one or more threads \"" + name + "\"");
              DEBUG ("threads \"" + name + "\" completed OK");
            }
 
            bool finished () const {
              for (auto& t : threads)
                if (t.wait_for (std::chrono::microseconds(0)) != std::future_status::ready)
                  return false;
              return true;
            }
 
            bool any_valid () const {
              for (auto& t : threads)
                if (t.valid())
                  return true;
              return false;
            }
 
            ~__multi_thread () {
              if (any_valid()) {
                try { wait(); }
                catch (Exception& E) { E.display(); }
              }
            }
          protected:
            vector<std::future<void>> threads;
            vector<typename std::remove_reference<Functor>::type> functors;
 
        };
 
 
      template <class Functor>
        class __Multi { NOMEMALIGN
          public:
            __Multi (Functor& object, size_t number) : functor (object), num (number) { }
            __Multi (__Multi&& m) = default;
            template <class X> bool operator() (const X&) { assert (0); return false; }
            template <class X> bool operator() (X&) { assert (0); return false; }
            template <class X, class Y> bool operator() (const X&, Y&) { assert (0); return false; }
            typename std::remove_reference<Functor>::type& functor;
            size_t num;
        };
 
      template <class Functor>
        class __run { NOMEMALIGN
          public:
            using type = __single_thread;
            type operator() (Functor& functor, const std::string& name) {
              return { functor, name };
            }
        };
 
      template <class Functor>
        class __run<__Multi<Functor>> { NOMEMALIGN
          public:
            using type = __multi_thread<Functor>;
            type operator() (__Multi<Functor>& functor, const std::string& name) {
              return { functor.functor, functor.num, name };
            }
        };
 
    }
 
 
 
    size_t number_of_threads ();
 
    enum class nthreads_t { UNINITIALISED, EXPLICIT, IMPLICIT };
    nthreads_t type_nthreads ();
 
    size_t threads_to_execute ();
 
 
 
 
    template <class Functor>
      inline __Multi<typename std::remove_reference<Functor>::type>
      multi (Functor&& functor, size_t nthreads = threads_to_execute())
      {
        return { functor, nthreads };
      }
 
 
 
 
    template <class Functor>
      inline typename __run<Functor>::type run (Functor&& functor, const std::string& name = "unnamed")
      {
        return __run<typename std::remove_reference<Functor>::type>() (functor, name);
      }
 
  }
}
 
#endif