Prefetcher.h

Go to the documentation of this file.

#ifndef _SEIMPLEMENTATION_MEASUREMENT_PREFETCHER_H_
#define _SEIMPLEMENTATION_MEASUREMENT_PREFETCHER_H_

#include <condition_variable>
#include "AlexandriaKernel/ThreadPool.h"
#include "AlexandriaKernel/Semaphore.h"
#include "SEFramework/Source/SourceInterface.h"
#include "SEFramework/Pipeline/PipelineStage.h"

namespace SourceXtractor {

class Prefetcher : public PipelineReceiver<SourceInterface>, public PipelineEmitter<SourceInterface> {
public:

  Prefetcher(const std::shared_ptr<Euclid::ThreadPool>& thread_pool, unsigned max_queue_size);

  virtual ~Prefetcher();

  void receiveSource(std::unique_ptr<SourceInterface> source) override;

  void receiveProcessSignal(const ProcessSourcesEvent& event) override;

  template<typename Container>
  void requestProperties(const Container& properties) {
    for (auto& p : properties) {
      requestProperty(p);
    }
  }

  void wait();

  void synchronize();


private:
  struct EventType {
    enum Type {
      SOURCE, PROCESS_SOURCE
    } m_event_type;
    intptr_t m_source_addr;

    explicit EventType(Type type, intptr_t source_addr = -1)
      : m_event_type(type), m_source_addr(source_addr) {}
  };

  std::shared_ptr<Euclid::ThreadPool> m_thread_pool;
  std::set<PropertyId> m_prefetch_set;
  std::unique_ptr<std::thread> m_output_thread;
  std::condition_variable m_new_output;
  std::map<intptr_t, std::unique_ptr<SourceInterface>> m_finished_sources;
  std::deque<ProcessSourcesEvent> m_event_queue;
  std::deque<EventType> m_received;

  std::mutex m_queue_mutex;

  std::atomic_bool m_stop;

  Euclid::Semaphore m_semaphore;

  void requestProperty(const PropertyId& property_id);
  void outputLoop();
};

} // end of namespace SourceXtractor

#endif // _SEIMPLEMENTATION_MEASUREMENT_PREFETCHER_H_