worker.hpp

#pragma once
 
#include "declarations.hpp"
#include "wsq.hpp"
#include "nonblocking_notifier.hpp"
#include "atomic_notifier.hpp"

 
namespace tf {
 
// ----------------------------------------------------------------------------
// Default Notifier
// Our experiments show that NonblockingNotifier has the most stable performance
// ----------------------------------------------------------------------------

#ifdef TF_ENABLE_ATOMIC_NOTIFIER
  using DefaultNotifier = AtomicNotifier;
#else
  using DefaultNotifier = NonblockingNotifier;
#endif
 
// ----------------------------------------------------------------------------
// Class Definition: Worker
// ----------------------------------------------------------------------------

class Worker {
 
  friend class Executor;
  friend class Runtime;
  friend class WorkerView;
  
  using wsq_type = BoundedWSQ<Node*>;
 
  public:

  inline size_t id() const { return _id; }

  inline size_t queue_size() const { return _wsq.size(); }
  
  inline size_t queue_capacity() const { return static_cast<size_t>(_wsq.capacity()); }

  std::thread& thread() { return _thread; }
 
  private:
  
  alignas(TF_CACHELINE_SIZE) std::atomic_flag _done = ATOMIC_FLAG_INIT; 
 
  size_t _id;
  size_t _sticky_victim;
  Xorshift<uint32_t> _rdgen; 
  std::thread _thread;
  wsq_type _wsq;
  //std::vector<Node*> _pool;
};
 
// ----------------------------------------------------------------------------
// Class Definition: WorkerView
// ----------------------------------------------------------------------------

class WorkerView {
 
  friend class Executor;
 
  public:

  size_t id() const;

  size_t queue_size() const;

  size_t queue_capacity() const;
 
  private:
 
  WorkerView(const Worker&);
  WorkerView(const WorkerView&) = default;
 
  const Worker& _worker;
 
};
 
// Constructor
inline WorkerView::WorkerView(const Worker& w) : _worker{w} {
}
 
// function: id
inline size_t WorkerView::id() const {
  return _worker._id;
}
 
// Function: queue_size
inline size_t WorkerView::queue_size() const {
  return _worker._wsq.size();
}
 
// Function: queue_capacity
inline size_t WorkerView::queue_capacity() const {
  return static_cast<size_t>(_worker._wsq.capacity());
}
 
// ----------------------------------------------------------------------------
// Class Definition: WorkerInterface
// ----------------------------------------------------------------------------

class WorkerInterface {
 
  public:
  
  virtual ~WorkerInterface() = default;

  virtual void scheduler_prologue(Worker& worker) = 0;

  virtual void scheduler_epilogue(Worker& worker, std::exception_ptr ptr) = 0;
 
};

template <typename T, typename... ArgsT>
std::shared_ptr<T> make_worker_interface(ArgsT&&... args) {
  static_assert(
    std::is_base_of_v<WorkerInterface, T>,
    "T must be derived from WorkerInterface"
  );
  return std::make_shared<T>(std::forward<ArgsT>(args)...);
}
 
 
                                                                                 
                                                                                 
}  // end of namespact tf ------------------------------------------------------