tensor_runtime.hpp

 
#ifndef EXATN_RUNTIME_TENSOR_RUNTIME_HPP_
#define EXATN_RUNTIME_TENSOR_RUNTIME_HPP_
 
#include "tensor_graph_executor.hpp"
#include "tensor_graph.hpp"
#include "tensor_operation.hpp"
#include "tensor_method.hpp"
 
#include <map>
#include <list>
#include <string>
#include <vector>
#include <memory>
#include <thread>
#include <atomic>
#include <future>
#include <mutex>
 
namespace exatn {
namespace runtime {
 
class TensorRuntime final {
 
public:
  TensorRuntime(const std::string & graph_executor_name = "eager-dag-executor",  //DAG executor kind
                const std::string & node_executor_name = "talsh-node-executor"); //DAG node executor kind
  TensorRuntime(const TensorRuntime &) = delete;
  TensorRuntime & operator=(const TensorRuntime &) = delete;
  TensorRuntime(TensorRuntime &&) noexcept = delete;
  TensorRuntime & operator=(TensorRuntime &&) noexcept = delete;
  ~TensorRuntime();
 
  void resetLoggingLevel(int level = 0);
 
  void openScope(const std::string & scope_name);
 
  void pauseScope();
 
  void resumeScope(const std::string & scope_name);
 
  void closeScope();
 
  inline bool currentScopeIsSet() const {return !(current_scope_.empty());}
 
  VertexIdType submit(std::shared_ptr<TensorOperation> op);
 
  bool sync(TensorOperation & op,
            bool wait = true);
 
  bool sync(const Tensor & tensor,
            bool wait = true);
 
  std::future<std::shared_ptr<talsh::Tensor>> getLocalTensor(std::shared_ptr<Tensor> tensor, //in: exatn::numerics::Tensor to get slice of (by copy)
                            const std::vector<std::pair<DimOffset,DimExtent>> & slice_spec); //in: tensor slice specification
 
private:
  class TensorDataReq{
  public:
   std::promise<std::shared_ptr<talsh::Tensor>> slice_promise_;
   std::vector<std::pair<DimOffset,DimExtent>> slice_specs_;
   std::shared_ptr<Tensor> tensor_;
 
   TensorDataReq(std::promise<std::shared_ptr<talsh::Tensor>> && slice_promise,
                 const std::vector<std::pair<DimOffset,DimExtent>> & slice_specs,
                 std::shared_ptr<Tensor> tensor):
    slice_promise_(std::move(slice_promise)), slice_specs_(slice_specs), tensor_(tensor){}
 
   TensorDataReq(const TensorDataReq & req) = delete;
   TensorDataReq & operator=(const TensorDataReq & req) = delete;
   TensorDataReq(TensorDataReq && req) noexcept = default;
   TensorDataReq & operator=(TensorDataReq && req) noexcept = default;
   ~TensorDataReq() = default;
  };
 
  void launchExecutionThread();
  void executionThreadWorkflow();
  void processTensorDataRequests();
 
  inline void lockDataReqQ(){data_req_mtx_.lock();}
  inline void unlockDataReqQ(){data_req_mtx_.unlock();}
 
  std::string graph_executor_name_;
  std::string node_executor_name_;
  std::shared_ptr<TensorGraphExecutor> graph_executor_;
  std::map<std::string, std::shared_ptr<TensorGraph>> dags_;
  std::string current_scope_;
  std::shared_ptr<TensorGraph> current_dag_; //pointer to the current DAG
  std::list<TensorDataReq> data_req_queue_;
  int logging_;
  std::atomic<bool> executing_; //TRUE while the execution thread is executing the current DAG
  std::atomic<bool> alive_; //TRUE while the main thread is accepting new operations from Client
  std::thread exec_thread_;
  std::mutex data_req_mtx_;
};
 
} // namespace runtime
} // namespace exatn
 
#endif //EXATN_RUNTIME_TENSOR_RUNTIME_HPP_