syclflow.hpp

Go to the documentation of this file.

#pragma once
 
#include "../taskflow.hpp"
#include "sycl_task.hpp"
 
namespace tf {
 
// ----------------------------------------------------------------------------
// class definition: syclFlow
// ----------------------------------------------------------------------------
 
class syclFlow {
 
  friend class Executor;
 
  struct External {
    syclGraph graph;
  };
 
  struct Internal {
    Executor& executor;
    Internal(Executor& e) : executor {e} {}
  };
 
  using handle_t = std::variant<External, Internal>;
 
  public:
   
    syclFlow(sycl::queue& queue);
    
    ~syclFlow() = default;
 
    bool empty() const;
 
    size_t num_tasks() const;
    
    void dump(std::ostream& os) const;
 
    void clear();
 
    // ------------------------------------------------------------------------
    // Generic device operations
    // ------------------------------------------------------------------------
 
    template <typename F, std::enable_if_t<
      std::is_invocable_r_v<void, F, sycl::handler&>, void>* = nullptr
    >
    syclTask on(F&& func);
    
    template <typename F, std::enable_if_t<
      std::is_invocable_r_v<void, F, sycl::handler&>, void>* = nullptr
    >
    void on(syclTask task, F&& func);
    
    syclTask memcpy(void* tgt, const void* src, size_t bytes);
    
    syclTask memset(void* ptr, int value, size_t bytes);
    
    template <typename T>
    syclTask fill(void* ptr, const T& pattern, size_t count);
    
    template <typename T,
      std::enable_if_t<!std::is_same_v<T, void>, void>* = nullptr
    >
    syclTask copy(T* target, const T* source, size_t count);
    
    template <typename...ArgsT>
    syclTask parallel_for(ArgsT&&... args);
    
    // ------------------------------------------------------------------------
    // algorithms
    // ------------------------------------------------------------------------
    
    template <typename F>
    syclTask single_task(F&& func);
    
    template <typename I, typename C>
    syclTask for_each(I first, I last, C&& callable);
    
    template <typename I, typename C>
    syclTask for_each_index(I first, I last, I step, C&& callable);
    
    template <typename I, typename C, typename... S>
    syclTask transform(I first, I last, C&& callable, S... srcs);
    
    template <typename I, typename T, typename C>
    syclTask reduce(I first, I last, T* result, C&& op);
    
    template <typename I, typename T, typename C>
    syclTask uninitialized_reduce(I first, I last, T* result, C&& op);
    
    // ------------------------------------------------------------------------
    // offload methods
    // ------------------------------------------------------------------------
 
    template <typename P>
    void offload_until(P&& predicate);
    
    void offload_n(size_t N);
 
    void offload();
    
    // ------------------------------------------------------------------------
    // update methods
    // ------------------------------------------------------------------------
    
 
    void memcpy(syclTask task, void* tgt, const void* src, size_t bytes);
    
    void memset(syclTask task, void* ptr, int value, size_t bytes);
    
    template <typename T>
    void fill(syclTask task, void* ptr, const T& pattern, size_t count);
    
    template <typename T,
      std::enable_if_t<!std::is_same_v<T, void>, void>* = nullptr
    >
    void copy(syclTask task, T* target, const T* source, size_t count);
    
    template <typename...ArgsT>
    void parallel_for(syclTask task, ArgsT&&... args);
 
    template <typename F>
    void single_task(syclTask task, F&& func);
    
  private:
 
    syclFlow(Executor&, syclGraph&, sycl::queue&);
    
    sycl::queue& _queue;
    
    handle_t _handle;
    
    syclGraph& _graph;
  
    std::vector<syclNode*> _tpg;
    std::queue<syclNode*> _bfs;
};
 
// constructor
inline syclFlow::syclFlow(sycl::queue& queue) :
  _queue  {queue}, 
  _handle {std::in_place_type_t<External>{}},
  _graph  {std::get_if<External>(&_handle)->graph} {
}
 
// Construct the syclFlow from executor (internal graph)
inline syclFlow::syclFlow(Executor& e, syclGraph& g, sycl::queue& queue) :
  _queue  {queue},
  _handle {std::in_place_type_t<Internal>{}, e},
  _graph  {g} {
}
 
// Function: empty
inline bool syclFlow::empty() const {
  return _graph._nodes.empty();
}
 
// Function: num_tasks
inline size_t syclFlow::num_tasks() const {
  return _graph._nodes.size();
}
 
// Procedure: dump
inline void syclFlow::dump(std::ostream& os) const {
  _graph.dump(os, nullptr, "");
}
 
// Procedure: clear
inline void syclFlow::clear() {
  _graph.clear();
}
 
// Function: memcpy
inline syclTask syclFlow::memcpy(void* tgt, const void* src, size_t bytes) {
  return on([=](sycl::handler& h){ h.memcpy(tgt, src, bytes); });
}
 
// Function: memset
inline syclTask syclFlow::memset(void* ptr, int value, size_t bytes) {
  return on([=](sycl::handler& h){ h.memset(ptr, value, bytes); });
}
 
// Function: fill
template <typename T>
syclTask syclFlow::fill(void* ptr, const T& pattern, size_t count) {
  return on([=](sycl::handler& h){ h.fill(ptr, pattern, count); });
}
 
// Function: copy
template <typename T,
  std::enable_if_t<!std::is_same_v<T, void>, void>*
>
syclTask syclFlow::copy(T* target, const T* source, size_t count) {
  return on([=](sycl::handler& h){ h.memcpy(target, source, count*sizeof(T)); });
}
 
// Function: on
template <typename F, std::enable_if_t<
  std::is_invocable_r_v<void, F, sycl::handler&>, void>*
>
syclTask syclFlow::on(F&& f) {
  auto node = _graph.emplace_back(_graph, 
    std::in_place_type_t<syclNode::CGH>{}, std::forward<F>(f)
  );
  return syclTask(node);
}
 
// Function: single_task
template <typename F>
syclTask syclFlow::single_task(F&& func) {
  return on([f=std::forward<F>(func)] (sycl::handler& h) {
    h.single_task(f);
  });
}
 
// Function: parallel_for
template <typename...ArgsT>
syclTask syclFlow::parallel_for(ArgsT&&... args) {
  return on([args...] (sycl::handler& h) { h.parallel_for(args...); });
}
 
// Procedure: offload_until
template <typename P>
void syclFlow::offload_until(P&& predicate) {
  
  if(!(_graph._state & syclGraph::TOPOLOGY_CHANGED)) {
    goto offload;
  }
 
  // levelize the graph
  _tpg.clear();
 
  // insert the first level of nodes into the queue
  for(auto& u : _graph._nodes) {
    u->_level = u->_dependents.size();
    if(u->_level == 0) {
      _bfs.push(u.get());
    }
  }
  
  while(!_bfs.empty()) {
    auto u = _bfs.front();
    _bfs.pop();
    _tpg.push_back(u);
    for(auto v : u->_successors) {
      if(--(v->_level) == 0) {
        v->_level = u->_level + 1;
        _bfs.push(v);
      }
    }
  }
 
  offload:
  
  // offload the syclFlow graph
  bool in_order = _queue.is_in_order();
  
  while(!predicate()) {
 
    // traverse node in a topological order
    for(auto u : _tpg) {
      
      switch(u->_handle.index()) {
        // task type 1: command group handler 
        case syclNode::COMMAND_GROUP_HANDLER:
          u->_event = _queue.submit([u, in_order](sycl::handler& h){
            // wait on all predecessors
            if(!in_order) {
              for(auto p : u->_dependents) {
                h.depends_on(p->_event);
              }
            }
            std::get_if<syclNode::CGH>(&u->_handle)->work(h);
          });
        break;
      }
    }
    
    // synchronize the execution
    _queue.wait();
  }
 
  _graph._state = syclGraph::OFFLOADED;
}
 
// Procedure: offload_n
inline void syclFlow::offload_n(size_t n) {
  offload_until([repeat=n] () mutable { return repeat-- == 0; });
}
 
// Procedure: offload
inline void syclFlow::offload() {
  offload_until([repeat=1] () mutable { return repeat-- == 0; });
}
 
// Function: on
template <typename F, std::enable_if_t<
  std::is_invocable_r_v<void, F, sycl::handler&>, void>*
>
void syclFlow::on(syclTask task, F&& f) {
  std::get_if<syclNode::CGH>(&task._node->_handle)->work = 
    std::forward<F>(f);
}
 
// Function: memcpy
inline void syclFlow::memcpy(
  syclTask task, void* tgt, const void* src, size_t bytes
) {
  on(task, [=](sycl::handler& h){ h.memcpy(tgt, src, bytes); });
}
 
// Function: memset
inline void syclFlow::memset(
  syclTask task, void* ptr, int value, size_t bytes
) {
  on(task, [=](sycl::handler& h){ h.memset(ptr, value, bytes); });
}
 
// Function: fill
template <typename T>
void syclFlow::fill(
  syclTask task, void* ptr, const T& pattern, size_t count
) {
  on(task, [=](sycl::handler& h){ h.fill(ptr, pattern, count); });
}
 
// Function: copy
template <typename T,
  std::enable_if_t<!std::is_same_v<T, void>, void>*
>
void syclFlow::copy(
  syclTask task, T* target, const T* source, size_t count
) {
  on(task, [=](sycl::handler& h){ 
    h.memcpy(target, source, count*sizeof(T));}
  );
}
 
// Function: parallel_for
template <typename...ArgsT>
void syclFlow::parallel_for(syclTask task, ArgsT&&... args) {
  on(task, [args...] (sycl::handler& h) { h.parallel_for(args...); });
}
    
// Function: single_task
template <typename F>
void syclFlow::single_task(syclTask task, F&& func) {
  on(task, [f=std::forward<F>(func)] (sycl::handler& h) { h.single_task(f); });
}
 
// ############################################################################
// Forward declaration: FlowBuilder
// ############################################################################
    
// FlowBuilder::emplace_on
template <typename C, typename Q, std::enable_if_t<is_syclflow_task_v<C>, void>*>
Task FlowBuilder::emplace_on(C&& callable, Q&& q) {
  auto n = _graph._emplace_back(
    std::in_place_type_t<Node::syclFlow>{},
    [c=std::forward<C>(callable), queue=std::forward<Q>(q)] 
    (Executor& e, Node* p) mutable {
      e._invoke_syclflow_task_entry(p, c, queue);
    },
    std::make_unique<syclGraph>()
  );
  return Task(n);
}
 
// FlowBuilder::emplace
template <typename C, std::enable_if_t<is_syclflow_task_v<C>, void>*>
Task FlowBuilder::emplace(C&& callable) {
  return emplace_on(std::forward<C>(callable), sycl::queue{});
}
 
// ############################################################################
// Forward declaration: Executor
// ############################################################################
 
// Procedure: _invoke_syclflow_task_entry (syclFlow)
template <typename C, typename Q,
  std::enable_if_t<is_syclflow_task_v<C>, void>*
>
void Executor::_invoke_syclflow_task_entry(Node* node, C&& c, Q& queue) {
 
  auto h = std::get_if<Node::syclFlow>(&node->_handle);
 
  syclGraph* g = dynamic_cast<syclGraph*>(h->graph.get());
  
  g->clear();
 
  syclFlow sf(*this, *g, queue);
 
  c(sf); 
 
  if(!(g->_state & syclGraph::OFFLOADED)) {
    sf.offload();
  }
}
 
}  // end of namespace tf -----------------------------------------------------