multiprocomp.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 MULTIPROCOMP_H
#define MULTIPROCOMP_H

#include <csignal>
#include <chrono>
#include <stdexcept>

#ifdef _OPENMP
#include <omp.h>
#else
inline constexpr int omp_get_thread_num() { return 0; }
inline constexpr int omp_get_num_threads() { return 1; }
#endif

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


namespace Tomographer {
namespace MultiProc {
namespace OMP {


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

public:

  template<typename... MoreArgs>
    ThreadSanitizerLogger(BaseLogger & logger, MoreArgs && ...)
    // 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)
  {
  }
    
  ~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)
  {
#pragma omp critical
    {
      _baselogger.emitLog(level, origin, msg);
    }
  }

  TOMOGRAPHER_ENABLED_IF(Logger::LoggerTraits<BaseLogger>::HasFilterByOrigin &&
                         !IsBaseLoggerThreadSafe)
  bool filterByOrigin(int level, const char * origin) const
  {
    bool ok;
#pragma omp critical
    {
      ok = _baselogger.filterByOrigin(level, origin);
    }
    
    return ok;
  }

};

} // namespace OMP
} // namespace MultiProc

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


namespace MultiProc {
namespace OMP {
    

template<typename TaskType_, typename TaskCData_, typename LoggerType_,
         typename TaskCountIntType_ = int,
         typename TaskLoggerType_ = ThreadSanitizerLogger<LoggerType_> >
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 TaskLoggerType_ TaskLoggerType;

  using typename Base::FullStatusReportCallbackType;

private:

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

  ThreadSharedDataType shared_data;
  TaskCountIntType n_chunk;

  struct CriticalSectionManager {

    template<typename Fn>
    inline void critical_status_report(Fn && fn) {
#pragma omp critical
      {
        fn();
      }
    }
    template<typename Fn>
    inline void critical_status_report_and_user_fn(Fn && fn) {
#pragma omp critical
      {
        fn();
      }
    }
    template<typename Fn>
    inline void critical_status_report_and_schedule(Fn && fn) {
#pragma omp critical
      {
        fn();
      }
    }
    template<typename Fn>
    inline void critical_schedule(Fn && fn) {
#pragma omp critical
      {
        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_,
                 TaskCountIntType n_chunk_ = 1)
    : shared_data(pcdata_, logger_, num_total_runs_, 0), n_chunk(n_chunk_)
  {
  }

  TaskDispatcher(TaskDispatcher && x)
    : shared_data(std::move(x.shared_data)),
      n_chunk(x.n_chunk)
  {
  }

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

#ifndef _OPENMP
    logger.warning("OpenMP is disabled; tasks will run serially.");
#endif

    // declaring these as "const" causes a weird compiler error
    // "`n_chunk' is predetermined `shared' for `shared'"
    TaskCountIntType num_total_runs = shared_data.schedule.num_total_runs;
    TaskCountIntType n_chunk_ = n_chunk;
    (void)n_chunk_; // silence "unused variable" warning when compiling without OMP support

    TaskCountIntType k = 0;

    ThreadSharedDataType * shdat = & shared_data;

    const std::string logger_prefix = logger.originPrefix()+logger.glue()+"worker";
    const std::string * logger_prefix_ptr = &logger_prefix;

    logger.debug("About to start parallel section");

#pragma omp parallel default(none) private(k) shared(shdat, logger_prefix_ptr, num_total_runs, n_chunk_)
    {
      // construct a thread-safe logger we can use
      TaskLoggerType threadsafelogger(shared_data.logger, shdat->pcdata, k);

      Tomographer::Logger::LocalLogger<TaskLoggerType> locallogger(*logger_prefix_ptr, threadsafelogger);

      ThreadPrivateDataType private_data(omp_get_thread_num(), & shared_data, locallogger, critical);

      private_data.shared_data = shdat;
      private_data.task_id = -1;

      // master thread sets shared_data.schedule.num_threads ...
#pragma omp master
      {
        shdat->schedule.num_threads = omp_get_num_threads();
      }

      // ... while other threads wait for master to be done
#pragma omp barrier
#pragma omp flush


      //
      // Register new parallel worker
      //
      this->run_worker_enter(private_data, *shdat);

      //
      // The main, parallel FOR loop:
      //
#pragma omp for schedule(dynamic,n_chunk) nowait
      for (k = 0; k < num_total_runs; ++k) {

        private_data.task_id = k;

        this->run_task(private_data, shared_data);
          
      } // omp for

      //
      // De-register parallel worker
      //
      this->run_worker_exit(private_data, *shdat);

#pragma omp master
      {
        this->master_continue_monitoring_status(private_data, *shdat) ;
      }
      
    } // omp parallel

    logger.debug("OpenMP parallel section finished");
    
    this->run_epilog(shared_data, logger) ;

    logger.debug("Done.");
  }

  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[k];
  }


  
  inline void setStatusReportHandler(FullStatusReportCallbackType fnstatus)
  {
#pragma omp critical
    {
      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 gomp lock. So can't call any
    // function which needs malloc or a #pragma omp critical.
    //
    // So just increment an atomic int.
    //

    ++ shared_data.status_report.event_counter_user;

  }

  inline void requestPeriodicStatusReport(int milliseconds)
  {
#pragma omp critical
    {
      shared_data.status_report.periodic_interval = milliseconds;
    }
  }

  inline void requestInterrupt()
  {
    shared_data.schedule.interrupt_requested = 1;
  }

    
}; // class TaskDispatcher


template<typename TaskType_, typename TaskCData_,
         typename LoggerType_, typename TaskCountIntType_ = int>
inline TaskDispatcher<TaskType_, TaskCData_,
                      LoggerType_, TaskCountIntType_>
makeTaskDispatcher(TaskCData_ * pcdata_, LoggerType_ & logger_,
                   TaskCountIntType_ num_total_runs_, TaskCountIntType_ n_chunk_ = 1)
{
  // RVO should be rather obvious to the compiler
  return TaskDispatcher<TaskType_, TaskCData_,
                        LoggerType_, TaskCountIntType_>(
                            pcdata_, logger_, num_total_runs_, n_chunk_
                            );
}



} // namespace OMP
} // namespace MultiProc

} // namespace Tomographer





#endif