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_ */