JPlatypus_t.hh

Go to the documentation of this file.

#ifndef __JACOUSTICS__JPLATYPUS_T__
#define __JACOUSTICS__JPLATYPUS_T__

#include <string>
#include <type_traits>
#include <functional>
#include <future>
#include <mutex>
#include <thread>
#include <vector>
#include <queue>

#include "JTools/JHashMap.hh"
#include "JTools/JQuantile.hh"

#include "JAcoustics/JHit.hh"
#include "JAcoustics/JEmitter.hh"
#include "JAcoustics/JFitParameters.hh"
#include "JAcoustics/JKatoomba_t.hh"
#include "JAcoustics/JSuperEvt.hh"
#include "JAcoustics/JSuperEvtToolkit.hh"


namespace JACOUSTICS {}
namespace JPP { using namespace JACOUSTICS; }

namespace JACOUSTICS {

  typedef JTOOLS::JHashMap<int, JEmitter>  JEmitters;
  
  /**
   * Thread pool for global fits.
   */
  class JPlatypus {
  public:
    /**
     * Constructor.
     *
     * \param  geometry           detector geometry
     * \param  emitters           emitters geometry
     * \param  velocity           sound velocity
     * \param  parameters         parameters
     * \param  ns                 number of threads
     */
    JPlatypus(const JGeometry&      geometry,
              const JEmitters&      emitters,
              const JSoundVelocity& velocity,
              const JFitParameters& parameters,
              const size_t          ns) :
      stop(false)
    {
      using namespace std;
      using namespace JPP;

      Q.reset();

      for (size_t i = 0; i < ns; ++i) {

        thread worker([this, geometry, emitters, velocity, parameters]() {

          vector<JHit> data;

          for (JKatoomba<JGandalf> katoomba(geometry, velocity, parameters.option); ; ) {

            katoomba.estimator.reset(getMEstimator(parameters.mestimator));
 
            {
              unique_lock<mutex> lock(in);

              cv.wait(lock, [this]() { return stop || !input.empty(); });

              if (stop && input.empty()) {
                return;
              }

              const JSuperEvt& evt = input.front();

              data.clear();
              
              for (JSuperEvt::rx_type::const_iterator hit = evt.rx.begin(); hit != evt.rx.end(); ++hit) {
                
                data.push_back(JHit(emitters[hit->id],
                                    hit->counter,
                                    JLocation(hit->string, hit->floor),
                                    hit->toa,
                                    parameters.sigma_s,
                                    hit->weight));
              }

              katoomba.value = getModel(evt);

              input.pop();
            }

            const double chi2 = katoomba (data.begin(), data.end()) / katoomba.estimator->getRho(1.0);
            const double ndf  = getWeight(data.begin(), data.end()) - katoomba.value.getN();

            {
              unique_lock<mutex> lock(out);

              Q.put(chi2 / ndf);
            }
          }
        });

        workers.emplace_back(move(worker));
      }
    }


    /**
     * Destructor.
     */   
    ~JPlatypus()
    {
      using namespace std;

      {
        unique_lock<mutex> lock(in);

        stop = true;
      }
      
      cv.notify_all();

      for (auto& worker : workers) {
        worker.join();
      }
    }


    /**
     * Get number of pending data.
     *
     * \return                number of pending data
     */
    size_t backlog()
    {
      using namespace std;

      {
        unique_lock<mutex> lock(in);

        return input.size();
      }
    }


    /**
     * Queue super event.
     *
     * \param  evt            super event
     */
    void enqueue(const JSuperEvt& evt)
    {
      using namespace std;
    
      {
        unique_lock<mutex> lock(in);

        if (stop) {
          throw runtime_error("The thread pool has been stopped.");
        }

        input.emplace(evt);
      }

      cv.notify_one();
    }

    static JTOOLS::JQuantile  Q;               //!< chi2/NDF
    
  private:
    std::vector<std::thread>  workers;
    std::queue <JSuperEvt>    input;
    std::mutex                in;
    std::mutex                out;
    std::condition_variable   cv;
    bool                      stop;    
  };
}

#endif