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) 2015 ETH Zurich, Institute for Theoretical Physics, 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 <tomographer2/tools/needownoperatornew.h>



namespace Tomographer {

namespace MultiProc {



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

    
template<typename TaskStatusReportType>
struct FullStatusReport
{
  FullStatusReport()
    : num_completed(0),
      num_total_runs(0),
      workers_running(),
      workers_reports()
  {
  }

  int num_completed;

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

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




namespace Sequential {

template<typename TaskType_, typename TaskCData_, typename ResultsCollector_,
         typename LoggerType_, typename CountIntType_ = int>
class TaskDispatcher
{
public:
  typedef TaskType_ TaskType;
  typedef typename TaskType::StatusReportType TaskStatusReportType;
  typedef TaskCData_ TaskCData;
  typedef ResultsCollector_ ResultsCollector;
  typedef LoggerType_ LoggerType;
  typedef CountIntType_ CountIntType;

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

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

private:
  
  const TaskCData * pcdata;
  ResultsCollector * results;
  LoggerType & logger;

  CountIntType num_total_runs;
  
  CountIntType task_k;

  struct TaskMgrIface {
    TaskMgrIface(TaskDispatcher * dispatcher_)
      : dispatcher(dispatcher_),
        status_report_requested(false),
        status_report_user_fn()
    {
    }

    TaskDispatcher * dispatcher;

    volatile std::sig_atomic_t status_report_requested;
    FullStatusReportCallbackType status_report_user_fn;
  
    inline void _request_status_report() { status_report_requested = 1; }

    inline bool statusReportRequested() const
    {
      return status_report_requested;
    }

    inline void submitStatusReport(const TaskStatusReportType &statreport)
    {
      FullStatusReport<TaskStatusReportType> 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, false);
      fullstatus.workers_reports[0] = statreport;

      status_report_user_fn(fullstatus);

      status_report_requested = false;
    }

  };

  TaskMgrIface mgriface;
  
public:
  TaskDispatcher(TaskCData * pcdata_, ResultsCollector * results_, LoggerType & logger_,
                 CountIntType num_total_runs_)
    : pcdata(pcdata_), results(results_), logger(logger_), num_total_runs(num_total_runs_),
        mgriface(this)
  {
  }
  
  void run()
  {
    results->init(num_total_runs, CountIntType(1), pcdata);
    
    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
      results->collectResult(task_k, t.getResult(), pcdata);
    }
    
    results->runsFinished(num_total_runs, pcdata);
  }
  
  
  template<typename Fn>
  inline void setStatusReportHandler(Fn fnstatus)
  {
    mgriface.status_report_user_fn = fnstatus;
  }
  
  inline void requestStatusReport()
  {
    mgriface._request_status_report();
  }
  
}; // class TaskDispatcher


} // namespace Sequential

} // namespace MultiProc

} // namespace Tomographer

#endif