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) 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 MULTIPROCOMP_H
#define MULTIPROCOMP_H

#include <csignal>

#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 <tomographer2/tools/loggers.h>
#include <tomographer2/tools/cxxutil.h> // tomographer_assert()
#include <tomographer2/tools/needownoperatornew.h>
#include <tomographer2/multiproc.h>


namespace Tomographer {
namespace MultiProc {
namespace OMP {

namespace tomo_internal {

template<typename BaseLogger, bool baseLoggerIsThreadSafe>
struct ThreadSanitizerLoggerHelper
{
  static inline void emitLog(BaseLogger & baselogger, int level, const char * origin, const std::string & msg)
  {
#pragma omp critical
    {
      //printf("ThreadSanitizerLoggerHelper::emitLog(%d, %s, %s) -- OMP CRITICAL\n", level, origin, msg.c_str());
      baselogger.emitLog(level, origin, msg);
    }
  }
  static inline bool filterByOrigin(BaseLogger & baselogger, int level, const char * origin)
  {
    bool ok = true;
#pragma omp critical
    {
      //printf("ThreadSanitizerLoggerHelper::filterByOrigin(%d, %s) -- OMP CRITICAL\n", level, origin);
      ok = baselogger.filterByOrigin(level, origin);
    }
    return ok;
  }
};

//
// specialize the helper for when logging to a thread-safe base logger. No critical
// section needed because the logger is already thread-safe.
//
template<typename BaseLogger>
struct ThreadSanitizerLoggerHelper<BaseLogger, true>
{
  static inline void emitLog(BaseLogger & baselogger, int level, const char * origin, const std::string & msg)
  {
    //printf("ThreadSanitizerLoggerHelper::emitLog(%d, %s, %s) -- NORMAL\n", level, origin, msg.c_str());
    baselogger.emitLog(level, origin, msg);
  }
  static inline bool filterByOrigin(BaseLogger & baselogger, int level, const char * origin)
  {
    //printf("ThreadSanitizerLoggerHelper::filterByOrigin(%d, %s) -- NORMAL\n", level, origin);
    return baselogger.filterByOrigin(level, origin);
  }
};

} // namespace tomo_internal


template<typename BaseLogger>
class ThreadSanitizerLogger : public Logger::LoggerBase<ThreadSanitizerLogger<BaseLogger> >
{
  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)
  {
    // when you have to debug the log mechanism.... lol
    //printf("ThreadSanitizerLogger(): object created\n");
    //_baselogger.debug("ThreadSanitizerLogger()", "log from constructor.");
    //emitLog(Logger::DEBUG, "ThreadSanitizerLogger!", "emitLog from constructor");
    //LoggerBase<ThreadSanitizerLogger<BaseLogger> >::debug("ThreadSanitizerLogger", "debug from constructor");
  }
    
  ~ThreadSanitizerLogger()
  {
  }

    
  inline void emitLog(int level, const char * origin, const std::string& msg)
  {
    //printf("ThreadSanitizerLogger::emitLog(%d, %s, %s)\n", level, origin, msg.c_str());
    tomo_internal::ThreadSanitizerLoggerHelper<BaseLogger,
                                               Logger::LoggerTraits<BaseLogger>::IsThreadSafe>
      ::emitLog(
          _baselogger, level, origin, msg
          );
  }

  template<bool dummy = true>
  inline typename std::enable_if<dummy && Logger::LoggerTraits<BaseLogger>::HasFilterByOrigin, bool>::type
    filterByOrigin(int level, const char * origin) const
  {
    return tomo_internal::ThreadSanitizerLoggerHelper<BaseLogger,
                                                      Logger::LoggerTraits<BaseLogger>::IsThreadSafe>
      ::filterByOrigin(
          _baselogger, level, origin
          );
  }
};

} // namespace OMP
} // namespace MultiProc

namespace Logger {
template<typename BaseLogger>
struct 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 ResultsCollector_,
         typename LoggerType_, typename CountIntType_ = int,
         typename TaskLoggerType_ = ThreadSanitizerLogger<LoggerType_> >
class TaskDispatcher
{
public:
  typedef TaskType_ TaskType;
  typedef typename TaskType::StatusReportType TaskStatusReportType;
  typedef TaskCData_ TaskCData;
  typedef ResultsCollector_ ResultsCollector;
  typedef LoggerType_ LoggerType;
  typedef CountIntType_ CountIntType;
  typedef TaskLoggerType_ TaskLoggerType;
  typedef FullStatusReport<TaskStatusReportType> FullStatusReportType;

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

private:

  struct thread_shared_data {
    thread_shared_data(const TaskCData * pcdata_, ResultsCollector * results_, LoggerType & logger_,
                       CountIntType num_total_runs_, CountIntType n_chunk_)
      : pcdata(pcdata_),
        results(results_),
        logger(logger_),
        status_report_underway(false),
        status_report_initialized(false),
        status_report_counter(0),
        status_report_numreportsrecieved(0),
        status_report_full(),
        status_report_user_fn(),
        num_total_runs(num_total_runs_), n_chunk(n_chunk_), num_completed(0),
        num_active_working_threads(0)
    { }

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

    bool status_report_underway;
    bool status_report_initialized;
    volatile std::sig_atomic_t status_report_counter;
    CountIntType status_report_numreportsrecieved;

    FullStatusReportType status_report_full;
    FullStatusReportCallbackType status_report_user_fn;

    CountIntType num_total_runs;
    CountIntType n_chunk;
    CountIntType num_completed;

    CountIntType num_active_working_threads;
  };
  struct thread_private_data
  {
    thread_shared_data * shared_data;

    TaskLoggerType * logger;

    CountIntType kiter;
    CountIntType local_status_report_counter;

    inline bool statusReportRequested()
    {
      //fprintf(stderr, "statusReportRequested(), shared_data=%p\n", shared_data);
      return (int)local_status_report_counter != (int)shared_data->status_report_counter;
    }

    inline void submitStatusReport(const TaskStatusReportType &statreport)
    {
      if ((int)local_status_report_counter == (int)shared_data->status_report_counter) {
        // error: task submitted unsollicited report
        logger->warning("OMP TaskDispatcher/taskmanageriface", "Task submitted unsollicited status report");
        return;
      }

#pragma omp critical
      {
        bool ok = true; // whether to proceed or not

        // we've reacted to the given "signal"
        local_status_report_counter = shared_data->status_report_counter;
          
        // add our status report to being-prepared status report in the  shared data
        int threadnum = omp_get_thread_num();

        // Important: access the original logger, not the thread-safe-wrapped logger!!
        // Otherwise this could lead to deadlocks because of nested critical blocks.
        shared_data->logger.longdebug("OMP TaskDispatcher/taskmanageriface", [&](std::ostream & stream) {
            stream << "status report received for thread #" << threadnum << ", treating it ...";
          });

        //
        // If we're the first reporting thread, we need to initiate the status reporing
        // procedure and initialize the general data
        //
        if (!shared_data->status_report_initialized) {

          //
          // Check that we indeed have to submit a status report.
          //
          if (shared_data->status_report_underway) {
            // status report already underway!
            // Important: access the original logger, not the thread-safe-wrapped logger!!
            // Otherwise this could lead to deadlocks because of nested critical blocks.
            shared_data->logger.warning("OMP TaskDispatcher/taskmanageriface",
                                        "status report already underway!");
            ok = false;
          }
          if (!shared_data->status_report_user_fn) {
            // no user handler set
            // Important: access the original logger, not the thread-safe-wrapped logger!!
            // Otherwise this could lead to deadlocks because of nested critical blocks.
            shared_data->logger.warning("OMP TaskDispatcher/taskmanageriface",
                                        "no user status report handler set!"
                                        " call setStatusReportHandler() first.");
            ok = false;
          }

          // since we can't return out of a critical section(?), we use an if block.
          if (ok) {

            shared_data->status_report_underway = true;
            shared_data->status_report_initialized = true;
              
            // initialize status report object & overall data
            shared_data->status_report_full = FullStatusReportType();
            shared_data->status_report_full.num_completed = shared_data->num_completed;
            shared_data->status_report_full.num_total_runs = shared_data->num_total_runs;
            // shared_data->status_report_full.num_active_working_threads = shared_data->num_active_working_threads;
            int num_threads = omp_get_num_threads();
            // shared_data->status_report_full.num_threads = num_threads;
              
            // initialize task-specific reports
            // fill our lists with default-constructed values & set all running to false.
            shared_data->status_report_full.workers_running.clear();
            shared_data->status_report_full.workers_reports.clear();
            shared_data->status_report_full.workers_running.resize(num_threads, false);
            shared_data->status_report_full.workers_reports.resize(num_threads);

            // Important: access the original logger, not the thread-safe-wrapped logger!!
            // Otherwise this could lead to deadlocks because of nested critical blocks.
            shared_data->logger.debug("OMP TaskDispatcher/taskmanageriface", [&](std::ostream & stream) {
                stream << "vectors resized to workers_running.size()="
                       << shared_data->status_report_full.workers_running.size()
                       << " and workers_reports.size()="
                       << shared_data->status_report_full.workers_reports.size()
                       << ".";
              });
              
            shared_data->status_report_numreportsrecieved = 0;
          }

        } // status_report_initialized

          // if we're the first reporting thread, then maybe ok was set to false above, so
          // check again.
        if (ok) {

          //
          // Report the data corresponding to this thread.
          //
          // Important: access the original logger, not the thread-safe-wrapped logger!!
          // Otherwise this could lead to deadlocks because of nested critical blocks.
          shared_data->logger.debug("OMP TaskDispatcher/taskmanageriface", "threadnum=%ld, workers_reports.size()=%ld",
                                    (long)threadnum, (long)shared_data->status_report_full.workers_reports.size());

          tomographer_assert(0 <= threadnum &&
                             (std::size_t)threadnum < shared_data->status_report_full.workers_reports.size());

          shared_data->status_report_full.workers_running[threadnum] = true;
          shared_data->status_report_full.workers_reports[threadnum] = statreport;

          ++ shared_data->status_report_numreportsrecieved;

          if (shared_data->status_report_numreportsrecieved == shared_data->num_active_working_threads) {
            // the report is ready to be transmitted to the user: go!
            shared_data->status_report_user_fn(shared_data->status_report_full);
            // all reports recieved: done --> reset our status_report_* flags
            shared_data->status_report_numreportsrecieved = 0;
            shared_data->status_report_underway = false;
            shared_data->status_report_initialized = false;
            shared_data->status_report_full.workers_running.clear();
            shared_data->status_report_full.workers_reports.clear();
          }
        } // if ok
      } // omp critical

    }
  };

  thread_shared_data shared_data;

public:
  TaskDispatcher(TaskCData * pcdata_, ResultsCollector * results_, LoggerType & logger_,
                 CountIntType num_total_runs_, CountIntType n_chunk_)
    : shared_data(pcdata_, results_, logger_, num_total_runs_, n_chunk_)
  {
  }

  void run()
    TOMOGRAPHER_CXX_STACK_FORCE_REALIGN
  {
    shared_data.results->init(shared_data.num_total_runs, shared_data.n_chunk, shared_data.pcdata);
      
    shared_data.logger.debug("MultiProc::OMP::TaskDispatcher::run()", "preparing for parallel runs");

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

    CountIntType k = 0;

    thread_shared_data *shdat = &shared_data;
    thread_private_data privdat;

    shared_data.logger.debug("MultiProc::OMP::TaskDispatcher::run()", "About to start parallel section");

#pragma omp parallel default(none) private(k, privdat) shared(shdat, num_total_runs, n_chunk)
    {
      privdat.shared_data = shdat;
      privdat.kiter = 0;

#pragma omp for schedule(dynamic,n_chunk) nowait
      for (k = 0; k < num_total_runs; ++k) {

        // make separate function call, so that we can tell GCC to realign the stack on
        // platforms which don't do that automatically (yes, MinGW, it's you I'm looking
        // at)
        _run_task(privdat, shdat, k);

      }
    }
    
    shared_data.results->runsFinished(num_total_runs, shared_data.pcdata);
  }

private:
  void _run_task(thread_private_data & privdat, thread_shared_data * shdat, CountIntType k)
    TOMOGRAPHER_CXX_STACK_FORCE_REALIGN
  {

#pragma omp critical
    {
      ++ shdat->num_active_working_threads;
      privdat.local_status_report_counter = shdat->status_report_counter;
    }

    // construct a thread-safe logger we can use
    TaskLoggerType threadsafelogger(shdat->logger, shdat->pcdata, k);
      
    // not sure an std::ostream would be safe here threadwise...?
    threadsafelogger.longdebug("Tomographer::MultiProc::OMP::TaskDispatcher::_run_task()",
                               "Run #%lu: thread-safe logger set up", (unsigned long)k);
      
    // set up our thread-private data
    privdat.kiter = k;
    privdat.logger = &threadsafelogger;
      
    // not sure an std::ostream would be safe here threadwise...?
    threadsafelogger.longdebug("Tomographer::MultiProc::OMP::TaskDispatcher::_run_task()",
                               "Run #%lu: querying CData for task input", (unsigned long)k);

    auto input = shdat->pcdata->getTaskInput(k);
      
    // not sure an std::ostream would be safe here threadwise...?
    threadsafelogger.debug("Tomographer::MultiProc::OMP::TaskDispatcher::_run_task()",
                           "Running task #%lu ...", (unsigned long)k);
      
    // construct a new task instance
    TaskType t(input, shdat->pcdata, threadsafelogger);
      
    // not sure an std::ostream would be safe here threadwise...?
    threadsafelogger.longdebug("Tomographer::MultiProc::OMP::TaskDispatcher::_run_task()",
                               "Task #%lu set up.", (unsigned long)k);
      
    // and run it
    t.run(shdat->pcdata, threadsafelogger, &privdat);
      
#pragma omp critical
    {
      shdat->results->collectResult(k, t.getResult(), shdat->pcdata);
        
      if ((int)privdat.local_status_report_counter != (int)shdat->status_report_counter) {
        // status report request missed by task... do as if we had provided a
        // report, but don't provide report.
        ++ shdat->status_report_numreportsrecieved;
      }
        
      ++ shdat->num_completed;
      -- shdat->num_active_working_threads;
    }
  }

public:

  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_counter = (shared_data.status_report_counter + 1) & 0x7f;

  }

}; // class TaskDispatcher


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



} // namespace OMP
} // namespace MultiProc

} // namespace Tomographer





#endif