multiprocthreads.h

Go to the documentation of this file.

/* This file is part of the Tomographer project, which is distributed under the
 * terms of the MIT license.
 *
 * The MIT License (MIT)
 *
 * Copyright (c) 2016 ETH Zurich, Institute for Theoretical Physics, Philippe Faist
 * Copyright (c) 2017 Caltech, Institute for Quantum Information and Matter, Philippe Faist
 *
 * Permission is hereby granted, free of charge, to any person obtaining a copy
 * of this software and associated documentation files (the "Software"), to deal
 * in the Software without restriction, including without limitation the rights
 * to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
 * copies of the Software, and to permit persons to whom the Software is
 * furnished to do so, subject to the following conditions:
 *
 * The above copyright notice and this permission notice shall be included in
 * all copies or substantial portions of the Software.
 *
 * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
 * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
 * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
 * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
 * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
 * OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
 * SOFTWARE.
 */

#ifndef MULTIPROCTHREADS_H
#define MULTIPROCTHREADS_H

#include <chrono>
#include <stdexcept>

#include <boost/exception/diagnostic_information.hpp>

#include <tomographer/tools/loggers.h>
#include <tomographer/tools/cxxutil.h> // tomographer_assert()
#include <tomographer/tools/needownoperatornew.h>
#include <tomographer/multiproc.h>
#include <tomographer/multiprocthreadcommon.h>

#include <thread>
#include <mutex>
// Include these here only, because on MinGW with mingw-std-threads, this
// includes windows headers which messes up tomographer/tools/logger.h by
// defining ERROR preprocessor symbols and other sh***t...
#ifdef TOMOGRAPHER_USE_MINGW_STD_THREAD
#  include <mingw.thread.h>
#  include <mingw.mutex.h>
#endif



namespace Tomographer {
namespace MultiProc {
namespace CxxThreads {


template<typename BaseLogger>
class TOMOGRAPHER_EXPORT ThreadSanitizerLogger
  : public Logger::LoggerBase<ThreadSanitizerLogger<BaseLogger> >
{
public:
  static constexpr bool IsBaseLoggerThreadSafe = Logger::LoggerTraits<BaseLogger>::IsThreadSafe;
private:
  BaseLogger & _baselogger;

  std::mutex * _mutex;
public:

  ThreadSanitizerLogger(BaseLogger & logger, std::mutex * mutex)
    // NOTE: pass the baselogger's level on here. The ThreadSanitizerLogger's
    // level is this one, and is fixed and cannot be changed while running.
    : Logger::LoggerBase<ThreadSanitizerLogger<BaseLogger> >(logger.level()),
    _baselogger(logger),
    _mutex(mutex)
  {
  }
    
  ~ThreadSanitizerLogger()
  {
  }

    
  TOMOGRAPHER_ENABLED_IF(IsBaseLoggerThreadSafe)
  inline void emitLog(int level, const char * origin, const std::string& msg)
  {
    _baselogger.emitLog(level, origin, msg);
  }

  TOMOGRAPHER_ENABLED_IF(Logger::LoggerTraits<BaseLogger>::HasFilterByOrigin &&
                         IsBaseLoggerThreadSafe)
  bool filterByOrigin(int level, const char * origin) const
  {
    return _baselogger.filterByOrigin(level, origin);
  }

  TOMOGRAPHER_ENABLED_IF(!IsBaseLoggerThreadSafe)
  inline void emitLog(int level, const char * origin, const std::string& msg)
  {
    std::lock_guard<std::mutex> lock(*_mutex);
    _baselogger.emitLog(level, origin, msg);
  }

  TOMOGRAPHER_ENABLED_IF(Logger::LoggerTraits<BaseLogger>::HasFilterByOrigin &&
                         !IsBaseLoggerThreadSafe)
  bool filterByOrigin(int level, const char * origin) const
  {
    std::lock_guard<std::mutex> lock(*_mutex);
    return _baselogger.filterByOrigin(level, origin);
  }

};

} // namespace CxxThreads
} // namespace MultiProc

namespace Logger {
template<typename BaseLogger>
struct TOMOGRAPHER_EXPORT LoggerTraits<MultiProc::CxxThreads::ThreadSanitizerLogger<BaseLogger> >
  : public LoggerTraits<BaseLogger>
{
  enum {
    HasOwnGetLevel = 0,
    IsThreadSafe = 1
  };
};
} // namespace Logger


namespace MultiProc {
namespace CxxThreads {
    

template<typename TaskType_, typename TaskCData_,
         typename LoggerType_, typename TaskCountIntType_ = int>
class TOMOGRAPHER_EXPORT TaskDispatcher
  : public Tomographer::MultiProc::ThreadCommon::TaskDispatcherBase<
      TaskType_,
      TaskCountIntType_
    >
{
public:
  typedef Tomographer::MultiProc::ThreadCommon::TaskDispatcherBase<TaskType_, TaskCountIntType_> Base;

  using typename Base::TaskType;
  using typename Base::TaskResultType;
  using typename Base::TaskStatusReportType;
  using typename Base::TaskCountIntType;
  using typename Base::FullStatusReportType;

  typedef TaskCData_ TaskCData;

  typedef LoggerType_ LoggerType;

  typedef ThreadSanitizerLogger<LoggerType_> TaskLoggerType;

  using typename Base::FullStatusReportCallbackType;

private:

  typedef typename Base::template ThreadSharedData<TaskCData, LoggerType>
    ThreadSharedDataType;

  ThreadSharedDataType shared_data;

  struct CriticalSectionManager {
    std::mutex user_mutex;
    
    std::mutex schedule_mutex;
    
    std::mutex status_report_mutex;

    template<typename Fn>
    inline void critical_status_report(Fn && fn) {
      std::lock_guard<std::mutex> lck(status_report_mutex);
      fn();
    }
    template<typename Fn>
    inline void critical_status_report_and_user_fn(Fn && fn) {
      std::lock(status_report_mutex, user_mutex);
      std::lock_guard<std::mutex> lck1(status_report_mutex, std::adopt_lock);
      std::lock_guard<std::mutex> lck2(user_mutex, std::adopt_lock);
      fn();
    }
    template<typename Fn>
    inline void critical_status_report_and_schedule(Fn && fn) {
      std::lock(status_report_mutex, schedule_mutex);
      std::lock_guard<std::mutex> lck1(status_report_mutex, std::adopt_lock);
      std::lock_guard<std::mutex> lck2(schedule_mutex, std::adopt_lock);
      fn();
    }
    template<typename Fn>
    inline void critical_schedule(Fn && fn) {
      std::lock_guard<std::mutex> lck(schedule_mutex);
      fn();
    }
  };

  CriticalSectionManager critical;
  
  typedef typename Base::template ThreadPrivateData<
    ThreadSharedDataType,
    Tomographer::Logger::LocalLogger<TaskLoggerType>,
    CriticalSectionManager
    >
    ThreadPrivateDataType;

public:
  TaskDispatcher(TaskCData * pcdata, LoggerType & logger,
                 TaskCountIntType num_total_runs,
                 int num_threads = (int)std::thread::hardware_concurrency())
    : shared_data(pcdata, logger, num_total_runs, num_threads)
  {
  }

  TaskDispatcher(TaskDispatcher && other)
    : shared_data(std::move(other.shared_data))
    // critical(std::move(other.critical)) -- mutexes are not movable, so just
    //                                        use new ones...  ugly :(
  {
  }

  ~TaskDispatcher()
  {
  }

  void run()
  {
    auto logger = Tomographer::Logger::makeLocalLogger(TOMO_ORIGIN,
                                                       shared_data.logger);
    logger.debug("Let's go!");

    shared_data.time_start = Base::StdClockType::now();
    
    logger.debug("Preparing for parallel runs");

    auto worker_fn_id = [&](const int thread_id) noexcept(true) {
      
      // construct a thread-safe logger we can use
      TaskLoggerType threadsafelogger(shared_data.logger, & critical.user_mutex);

      Tomographer::Logger::LocalLogger<TaskLoggerType> locallogger(
          logger.originPrefix()+logger.glue()+"worker",
          threadsafelogger);
      
      ThreadPrivateDataType private_data(thread_id, & shared_data, locallogger,
                                         critical);

      locallogger.longdebug([&](std::ostream & stream) {
          stream << "Thread #" << thread_id
                 << ": thread-safe logger and private thread data set up";
        }) ;
      
      {
        // active working region. This thread now takes care of handling tasks.
        this->run_worker_enter(private_data, shared_data);
        auto _f0 = Tools::finally([&]() {
            this->run_worker_exit(private_data, shared_data);
          });

        for ( ;; ) {
          // continue doing stuff until we stop

          if (shared_data.schedule.interrupt_requested) {
            break;
          }

          // get new task to perform
          critical.critical_schedule([&]() {
              if (shared_data.schedule.num_launched ==
                  shared_data.schedule.num_total_runs) {
                private_data.task_id = -1; // all tasks already launched ->
                                           // nothing else to do
                return;
              }
              private_data.task_id = shared_data.schedule.num_launched;
              ++ shared_data.schedule.num_launched ;
            }) ;

          if ( private_data.task_id < 0 ) {
            // all tasks already launched -> nothing else to do
            break;
          }

          // run this task.

          this->run_task(private_data, shared_data) ;

        } // for(;;)

      } // end of active working region, thread on longer serves to run tasks
        // (--num_active_working_threads is executed at this point)

      // only master thread should make sure it continues to serve status report
      // requests
      if (thread_id == 0 && !shared_data.schedule.interrupt_requested) {

        this->master_continue_monitoring_status(private_data, shared_data) ;

      }
      
    } ; // worker_fn_id

    //
    // now, prepare & launch the workers
    //

    logger.debug("About to launch threads");

    std::vector<std::thread> threads;

    // thread_id = 0 is reserved for ourselves.
    for (int thread_id = 1; thread_id < shared_data.schedule.num_threads;
         ++thread_id) {
      // NOTE: do NOT capture thread_id by reference!
      threads.push_back( std::thread( [thread_id,worker_fn_id]() {
            worker_fn_id(thread_id);
          } ) );
    }

    // also run stuff as master thread
    worker_fn_id(0);

    std::for_each(threads.begin(), threads.end(),
                  [](std::thread & thread) { thread.join(); }) ;

    logger.debug("Threads finished");

    this->run_epilog(shared_data, logger);

    logger.debug("All done.");

  } // run()

  

  inline TaskCountIntType numTaskRuns() const {
    return shared_data.schedule.num_total_runs;
  }

  inline const std::vector<TaskResultType*> & collectedTaskResults() const {
    return shared_data.results;
  }

  inline const TaskResultType & collectedTaskResult(std::size_t k) const {
    return *shared_data.results[(std::size_t)k];
  }


  inline void setStatusReportHandler(FullStatusReportCallbackType fnstatus)
  {
    std::lock_guard<std::mutex> lck(critical.status_report_mutex) ;
    shared_data.status_report.user_fn = fnstatus;
  }

  inline void requestStatusReport()
  {
    //
    // This function can be called from a signal handler. We essentially can't
    // do anything here because the state of the program can be pretty much
    // anything, including inside a malloc() or thread lock.
    //
    // So just increment an atomic int.
    //

    ++ shared_data.status_report.event_counter_user;
  }

  inline void requestPeriodicStatusReport(int milliseconds)
  {
    std::lock_guard<std::mutex> lck(critical.status_report_mutex) ;
    shared_data.status_report.periodic_interval = milliseconds;
  }

  inline void requestInterrupt()
  {
    // set the atomic int
    shared_data.schedule.interrupt_requested = 1;
  }
    
}; // class TaskDispatcher



template<typename TaskType_, typename TaskCData_,
         typename LoggerType_, typename TaskCountIntType_ = int>
inline
TaskDispatcher<TaskType_, TaskCData_, LoggerType_, TaskCountIntType_>
mkTaskDispatcher(TaskCData_ * pcdata_,
                 LoggerType_ & logger_,
                 TaskCountIntType_ num_total_runs_,
                 int num_threads_ = (int)std::thread::hardware_concurrency())
{
  return TaskDispatcher<TaskType_, TaskCData_, LoggerType_, TaskCountIntType_>(
      pcdata_, logger_, num_total_runs_, num_threads_
      ) ;
}



} // namespace CxxThreads
} // namespace MultiProc

} // namespace Tomographer





#endif