multiproc.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 MULTIPROC_H
#define MULTIPROC_H

#include <csignal>

#include <string>
#include <chrono>
#include <exception>

// task status reports etc. can be serialized with boost::serialization
#include <boost/serialization/serialization.hpp>

#include <tomographer/tools/fmt.h>
#include <tomographer/tools/needownoperatornew.h>



namespace Tomographer {

namespace MultiProc {



struct TOMOGRAPHER_EXPORT TaskStatusReport
{
  TaskStatusReport()
    : fraction_done(0), msg("<unknown>")
  { }
  TaskStatusReport(double fraction_done_, std::string msg_)
    : fraction_done(fraction_done_), msg(std::move(msg_))
  { }

  double fraction_done;
  std::string msg;

private:
  friend class boost::serialization::access;
  template<typename Archive>
  void serialize(Archive & a, unsigned int /*version*/)
  {
    a & fraction_done;
    a & msg;
  }
};

    
template<typename TaskStatusReportType, typename TaskCountIntType>
struct TOMOGRAPHER_EXPORT FullStatusReport
{
  FullStatusReport()
    : num_completed(0),
      num_total_runs(0),
      workers_running(),
      workers_reports(),
      elapsed(std::numeric_limits<double>::quiet_NaN())
  {
  }

  TaskCountIntType num_completed;

  TaskCountIntType num_total_runs;
    
  std::vector<bool> workers_running;

  std::vector<TaskStatusReportType,
              typename Tools::NeedOwnOperatorNew<TaskStatusReportType>::AllocatorType> workers_reports;

  double elapsed;


  inline double totalFractionDone() const
  {
    // idea: cumulate in f one unit per task completed.
    double f = num_completed;
    for (std::size_t k = 0; k < workers_running.size(); ++k) {
      if (workers_running[k]) {
        // partially completed tasks contribute a fraction
        f += workers_reports[k].fraction_done;
      }
    }
    // and f goes from zero to num_total_runs
    return f / num_total_runs;
  }

  inline std::string getHumanReport() const
  {
    std::string elapsed_s = Tomographer::Tools::fmtDuration(
        std::chrono::milliseconds(
            (std::chrono::milliseconds::rep)(elapsed*1000)
            ));
    std::stringstream ss;
    ss << "=========================== Intermediate Progress Report ============================\n"
       << "  "
       << elapsed_s << "s elapsed"
       << "  -  "
       << num_completed << "/" << num_total_runs << " runs completed"
       << "  -  "
       << std::fixed << std::setw(5) << std::setprecision(2) << totalFractionDone() * 100.0 << "% total done"
       << "\n";

    if (workers_running.size() == 0) {
      // no info
    } else if (workers_running.size() == 1) {
      if (workers_running[0]) {
        ss << "--> " << workers_reports[0].msg << "\n";
      }
    } else {
      ss << "Current Run(s) information (workers working/spawned "
         << std::count(workers_running.begin(), workers_running.end(), true)
         << "/" << workers_running.size() << "):\n";
      for (std::size_t k = 0; k < workers_running.size(); ++k) {
        ss << "=== " << std::setw(2) << k << ": ";
        if (!workers_running[k]) {
          ss << "<idle>\n";
        } else {
          ss << workers_reports[k].msg << "\n";
        }
      }
    }
    ss << "=====================================================================================\n";
    return ss.str();
  }


private:
  friend class boost::serialization::access;
  template<typename Archive>
  void serialize(Archive & a, unsigned int /*version*/)
  {
    a & num_completed;
    a & num_total_runs;
    a & workers_running;
    a & workers_reports;
    a & elapsed;
  }
};



class TOMOGRAPHER_EXPORT TasksInterruptedException : public std::exception
{
  std::string msg_;
public:
  TasksInterruptedException(std::string msg = "Tasks Interrupted.") : msg_(msg) { }
  virtual ~TasksInterruptedException() throw() { }
  const char * what() const throw() { return msg_.c_str(); }
};



namespace Sequential {

template<typename TaskType_, typename TaskCData_,
         typename LoggerType_, typename TaskCountIntType_ = int>
class TOMOGRAPHER_EXPORT TaskDispatcher
{
public:
  typedef TaskType_ TaskType;
  typedef typename TaskType::StatusReportType TaskStatusReportType;
  typedef TaskCData_ TaskCData;
  typedef LoggerType_ LoggerType;
  typedef TaskCountIntType_ TaskCountIntType;

  // not directly needed, but make sure TaskType::ResultType exists as part of testing the
  // task, cdata and result-collectors's correct type interface implementation
  typedef typename TaskType::ResultType TaskResultType;

  typedef FullStatusReport<TaskStatusReportType, TaskCountIntType> FullStatusReportType;

  typedef std::function<void(const FullStatusReportType&)> FullStatusReportCallbackType;

private:
  
  const TaskCData * pcdata;
  std::vector<TaskResultType*> results;
  LoggerType & logger;

  TaskCountIntType num_total_runs;
  
  TaskCountIntType task_k;

  typedef
#if defined(__GNUC__) && __GNUC__ == 4 && __GNUC_MINOR__ <= 6 && !defined(__clang__)
    std::chrono::monotonic_clock // for GCC/G++ 4.6
#else
    std::chrono::steady_clock
#endif
    StdClockType;


  struct TaskMgrIface {
    TaskMgrIface(TaskDispatcher * dispatcher_)
      : dispatcher(dispatcher_),
        interrupt_requested(0),
        status_report_requested(0),
        status_report_user_fn(),
        _tasks_start_time(StdClockType::now()),
        _last_status_report(StdClockType::now()),
        _status_report_periodic_interval(0)
    {
    }

  private:
    TaskDispatcher * dispatcher;

    volatile std::sig_atomic_t interrupt_requested; // could be written to by signal handler
    volatile std::sig_atomic_t status_report_requested; // could be written to by signal handler
    FullStatusReportCallbackType status_report_user_fn;

    const StdClockType::time_point _tasks_start_time;
    StdClockType::time_point _last_status_report;
    StdClockType::duration _status_report_periodic_interval;

    friend class TaskDispatcher;

    inline void _request_status_report() {
      status_report_requested = 1;
    }
    inline void _request_interrupt() {
      interrupt_requested = 1;
    }
    template<typename IntType>
    inline void _request_periodic_status_report(IntType milliseconds) {
      if ( milliseconds >= 0 ) {
        _status_report_periodic_interval = std::chrono::duration_cast<StdClockType::duration>(
            std::chrono::milliseconds(1+milliseconds)
            );
      } else {
        _status_report_periodic_interval = StdClockType::duration(0);
      }
    }

  public:
    inline bool statusReportRequested() const
    {
      if (interrupt_requested) {
        throw TasksInterruptedException();
      }
      if (_status_report_periodic_interval.count() > 0
          && (StdClockType::now() - (_last_status_report + _status_report_periodic_interval)).count() > 0) {
        return true;
      }
      return (bool) status_report_requested;
    }

    inline void submitStatusReport(const TaskStatusReportType &statreport)
    {
      FullStatusReportType fullstatus;
      
      fullstatus.num_completed = dispatcher->task_k;
      fullstatus.num_total_runs = dispatcher->num_total_runs;
              
      // initialize task-specific reports
      // fill our lists with default-constructed values & set all running to false.
      fullstatus.workers_running.clear();
      fullstatus.workers_reports.clear();

      fullstatus.workers_running.resize(1, false);
      fullstatus.workers_running[0] = true;

      fullstatus.workers_reports.resize(1);
      fullstatus.workers_reports[0] = statreport;

      fullstatus.elapsed = std::chrono::duration_cast<std::chrono::microseconds>(
          StdClockType::now() - _tasks_start_time
          ).count() * 1e-6;

      status_report_user_fn(std::move(fullstatus));

      status_report_requested = false;
      _last_status_report = StdClockType::now();
    }

  };

  TaskMgrIface mgriface;
  
public:
  TaskDispatcher(TaskCData * pcdata_, LoggerType & logger_, TaskCountIntType num_total_runs_)
    : pcdata(pcdata_), results(), logger(logger_), num_total_runs(num_total_runs_),
      mgriface(this)
  {
  }
  ~TaskDispatcher() {
    for ( auto r : results ) {
      if (r != NULL) {
        delete r;
      }
    }
  }
  
  void run()
  {
    results.resize((std::size_t)num_total_runs, NULL);
    
    logger.debug("MultiProc::Sequential::TaskDispatcher::run()", "preparing for sequential runs");
    
    for (task_k = 0; task_k < num_total_runs; ++task_k) {
      
      logger.debug("Tomographer::MultiProc::Sequential::TaskDispatcher::run()",
                   [&](std::ostream & stream) { stream << "Running task #" << task_k << " ..."; });
      
      auto input = pcdata->getTaskInput(task_k);
      
      // construct a new task instance
      TaskType t(input, pcdata, logger);

      // and run it
      t.run(pcdata, logger, &mgriface);
      
      // and collect the result
      logger.longdebug("MultiProc::Sequential::TaskDispatcher::run()", "collecting result");
      results[(std::size_t)task_k] = new TaskResultType(t.stealResult());
    }

    // all done
    logger.debug("MultiProc::Sequential::TaskDispatcher::run()", "all done");
  }

  inline TaskCountIntType numTaskRuns() const { return num_total_runs; }

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

  inline const TaskResultType & collectedTaskResult(std::size_t k) const { return *results[k]; }
  
  
  template<typename Fn>
  inline void setStatusReportHandler(Fn fnstatus)
  {
    mgriface.status_report_user_fn = fnstatus;
  }
  
  inline void requestStatusReport()
  {
    mgriface._request_status_report();
  }

  template<typename IntType>
  inline void requestPeriodicStatusReport(IntType milliseconds)
  {
    mgriface._request_periodic_status_report(milliseconds);
  }

  inline void requestInterrupt()
  {
    mgriface._request_interrupt();
  }
  
}; // 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_)
{
  return TaskDispatcher<TaskType_, TaskCData_, LoggerType_, TaskCountIntType_>(
      pcdata_, logger_, num_total_runs_
      );
}





} // namespace Sequential
} // namespace MultiProc
} // namespace Tomographer

#endif