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/SourceGrouping.h"
#include "SEUtils/Observable.h"
 
namespace SourceXtractor {
 
class Prefetcher : public Observer<std::shared_ptr<SourceInterface>>,
                   public Observable<std::shared_ptr<SourceInterface>>,
                   public Observer<ProcessSourcesEvent>,
                   public Observable<ProcessSourcesEvent> {
public:
 
  Prefetcher(const std::shared_ptr<Euclid::ThreadPool>& thread_pool, unsigned max_queue_size);
 
  virtual ~Prefetcher();
 
  void handleMessage(const std::shared_ptr<SourceInterface>& message) override;
 
  void handleMessage(const ProcessSourcesEvent& message) override;
 
  template<typename Container>
  void requestProperties(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::shared_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_