cpark.h

#ifndef CPARK_CPARK_H
#define CPARK_CPARK_H
 
#include <any>
#include <future>
#include <ostream>
#include <queue>
#include <set>
#include <shared_mutex>
#include <string>
#include <thread>
#include <unordered_map>
#include <unordered_set>
#include <vector>
 
#include "utils.h"
 
namespace cpark {
 
class Config {
public:
  enum class ParallelPolicy {
    Sequential,
    Thread,
#ifdef CPARK_DISTRIBUTED
    Distributed
#endif  // CPARK_DISTRIBUTED
  };
 
public:
  [[nodiscard]] constexpr const std::string& getDebugName() const noexcept { return debug_name_; }
 
  [[nodiscard]] constexpr size_t getParallelTaskNum() const noexcept { return parallel_task_num_; }
 
  [[nodiscard]] constexpr ParallelPolicy getParallelPolicy() const noexcept {
    return parallel_policy_;
  }
 
  [[nodiscard]] constexpr std::ostream* getLoggerPtr() const noexcept { return logger_; }
 
  [[nodiscard]] constexpr std::ostream& getLoggerOrNullStream() const noexcept {
    return logger_ ? *logger_ : utils::g_null_ostream;
  }
 
  Config& setDebugName(std::string name) noexcept {
    debug_name_ = std::move(name);
    return *this;
  }
 
  Config& setParallelTaskNum(size_t num = 0) noexcept {
    if (num == 0) {
      unsigned int n = std::thread::hardware_concurrency();
      if (n != 0) {
        parallel_task_num_ = n;
      }
    } else {
      parallel_task_num_ = num;
    }
    return *this;
  }
 
  Config& setParallelPolicy(ParallelPolicy policy) noexcept {
    parallel_policy_ = policy;
    return *this;
  }
 
  Config& setLogger(std::ostream* logger) noexcept {
    logger_ = logger;
    return *this;
  }
 
private:
  std::string debug_name_{};
  size_t parallel_task_num_{8};
  ParallelPolicy parallel_policy_{ParallelPolicy::Thread};
  std::ostream* logger_{nullptr};  // Should this be here?
 
#ifdef CPARK_DISTRIBUTED
public:
  struct Address {
    uint32_t ip_;
    uint16_t port_;
  };
 
public:
  Config& addWorker(const Address& address) {
    workers_.push_back(address);
    return *this;
  }
 
  Config& setMaster(const Address& address) noexcept {
    master_ = address;
    return *this;
  }
 
private:
  std::vector<Address> workers_;
  Address master_;
#endif  //CPARK_DISTRIBUTED
};
 
class ExecutionContext {
public:
  using RddId = uint32_t;
 
  using SplitId = uint32_t;
 
public:
  ExecutionContext() = default;
 
  explicit ExecutionContext(Config config) : config_{std::move(config)} {}
 
  void setConfig(Config config) { config_ = std::move(config); }
 
  const Config& getConfig() const noexcept { return config_; }
 
  RddId getAndIncRddId() { return next_rdd_id_++; }
 
  SplitId getAndIncSplitId() { return next_split_id_++; }
 
  bool splitShouldCache(SplitId split_id) const noexcept {
    std::shared_lock guard(cache_mutex_);
    return dependent_by_.contains(split_id) && dependent_by_.at(split_id).size() >= 2;
  }
 
  bool splitCached(SplitId split_id) const noexcept {
    std::shared_lock guard(cache_mutex_);
    return cache_done_.contains(split_id) &&
           cache_done_[split_id].wait_for(std::chrono::seconds(0)) == std::future_status::ready;
  }
 
  void markDependency(SplitId from, SplitId to) noexcept {
    std::shared_lock guard(cache_mutex_);
    dependent_by_[to].insert(from);
  }
 
  const std::any& getSplitCache(SplitId split_id) const {
    std::shared_lock guard(cache_mutex_);
    return cache_.at(split_id);
  }
 
  template <typename CacheType, typename OriginalIterator>
  std::shared_future<void> startCalculationOrGetFuture(SplitId split_id, OriginalIterator begin,
                                                       OriginalIterator end) {
    {
      std::shared_lock guard(cache_mutex_);
      if (cache_done_.contains(split_id)) {
        return cache_done_[split_id];
      }
    }
 
    std::promise<void> promise{};
 
    {
      std::unique_lock guard(cache_mutex_);
      if (cache_done_.contains(split_id)) {
        return cache_done_[split_id];
      }
      cache_done_[split_id] = promise.get_future();
    }
 
    CacheType cache;
    std::copy(begin, end, std::back_inserter(cache));
 
    {
      std::unique_lock guard(cache_mutex_);
      cache_[split_id] = std::move(cache);
      promise.set_value();
      return cache_done_[split_id];
    }
  }
 
private:
  Config config_{};
 
  // Using them to create incremental unique id for Rdd and Split.
  std::atomic<RddId> next_rdd_id_{};
  std::atomic<SplitId> next_split_id_{};
 
  // Which splits are relying on this one.
  std::unordered_map<SplitId, std::unordered_set<SplitId>> dependent_by_{};
 
  // Cache information for the Splits.
  std::unordered_map<SplitId, std::any> cache_{};
  mutable std::unordered_map<SplitId, std::shared_future<void>> cache_done_{};
  mutable std::shared_mutex cache_mutex_{};
 
  // Thread synchronization information.
 
  // Scheduler.
};
 
}  // namespace cpark
 
#endif  //CPARK_CPARK_H