MultithreadedMeasurement.h

Go to the documentation of this file.

/*
 * Multithreadedmeasurement.h
 *
 *  Created on: May 17, 2018
 *      Author: mschefer
 */
 
#ifndef _SEIMPLEMENTATION_OUTPUT_MULTITHREADEDMEASUREMENT_H_
#define _SEIMPLEMENTATION_OUTPUT_MULTITHREADEDMEASUREMENT_H_
 
#include <atomic>
#include <thread>
#include <mutex>
#include <condition_variable>
#include <atomic>
#include "AlexandriaKernel/ThreadPool.h"
#include "AlexandriaKernel/Semaphore.h"
#include "SEFramework/Pipeline/Measurement.h"
 
namespace SourceXtractor {
 
class MultithreadedMeasurement : public Measurement {
public:
 
  using SourceToRowConverter = std::function<Euclid::Table::Row(const SourceInterface&)>;
  MultithreadedMeasurement(SourceToRowConverter source_to_row, const std::shared_ptr<Euclid::ThreadPool>& thread_pool,
                           unsigned max_queue_size)
      : m_source_to_row(source_to_row),
        m_thread_pool(thread_pool),
        m_group_counter(0),
        m_input_done(false), m_abort_raised(false), m_semaphore(max_queue_size) {}
 
  ~MultithreadedMeasurement() override;
 
  void receiveSource(std::unique_ptr<SourceGroupInterface> source_group) override;
  void receiveProcessSignal(const ProcessSourcesEvent& event) override;
 
  void startThreads() override;
  void stopThreads() override;
  void synchronizeThreads() override;
 
private:
  static void outputThreadStatic(MultithreadedMeasurement* measurement);
  void outputThreadLoop();
 
  SourceToRowConverter m_source_to_row;
  std::shared_ptr<Euclid::ThreadPool> m_thread_pool;
  std::unique_ptr<std::thread> m_output_thread;
 
  int m_group_counter;
  std::atomic_bool m_input_done, m_abort_raised;
 
  std::condition_variable m_new_output;
  std::list<std::pair<int, std::unique_ptr<SourceGroupInterface>>> m_output_queue;
  std::mutex m_output_queue_mutex;
  Euclid::Semaphore m_semaphore;
};
 
}
 
#endif /* _SEIMPLEMENTATION_OUTPUT_MULTITHREADEDMEASUREMENT_H_ */