Parallel EntityX prototype

entityx_tx2.cc

#include <utility>
#include <thread>
#include <limits>
#include <algorithm>
#include <cassert>
#include <vector>
#include <cstdint>
#include <iostream>
#include <mutex>
#include <future>
#include <atomic>
#include <tuple>
#include <map>
#include <string>
#include "optional.hh"

using std::cerr;
using std::endl;


constexpr std::uint32_t UNDEFINED_ENTITY = std::numeric_limits<std::uint32_t>::max();
constexpr std::uint32_t DELETED_ENTITY = std::numeric_limits<std::uint32_t>::max() - 1;


std::recursive_mutex log_lock;

template <typename A>
void log(const A &arg) {
  log_lock.lock();
  cerr << arg << endl;
  log_lock.unlock();
}

template <typename A0, typename A1, typename ... Args>
void log(const A0 &a0, const A1 &a1, const Args & ... args) {
  log_lock.lock();
  cerr << a0;
  log(a1, args...);
  log_lock.unlock();
}


constexpr bool valid_entity(std::uint32_t id) { return id < DELETED_ENTITY; }

struct Entity {
  explicit Entity(std::uint32_t id = UNDEFINED_ENTITY) : id(id) {}

  std::uint32_t id;
};


template <typename ... Components>
class EntityManager;


template <typename ... Components>
class Transaction;


template <typename ... Components>
class TransactionSet;


/**
 * All of the data surrounding a single transactional commit.
 */
template <typename ... Components>
struct Commit {
  using Tuple = std::tuple<std::experimental::optional<Components>...>;
  using IndexedTuple = std::pair<std::uint32_t, Tuple>;
  using Iterator = typename std::vector<IndexedTuple>::iterator;

  // Abort the commit, discarding all mutations.
  void abort() {
    lookup_.clear();
    components_.clear();
  }

  std::size_t size() const {
    return lookup_.size();
  }

  bool has(std::uint32_t id) const {
    return id < lookup_.size() && valid_entity(lookup_[id]);
  }

  std::tuple<std::experimental::optional<Components>...> &operator[] (std::uint32_t id) {
    return components_[lookup_[id]].second;
  }

  const std::tuple<std::experimental::optional<Components>...> &operator[] (std::uint32_t id) const {
    return components_[lookup_[id]].second;
  }

  template <typename C>
  const C *operator[] (std::uint32_t id) const {
    std::experimental::optional<C> &component = components_[lookup_[id]].second;
    return component ? &component.value() : nullptr;
  }

  template <typename C>
  C *operator[] (std::uint32_t id) {
    std::experimental::optional<C> &component = components_[lookup_[id]].second;
    return component ? &component.value() : nullptr;
  }

  std::vector<std::uint32_t> lookup_;
  std::vector<IndexedTuple> components_;
  std::vector<std::uint32_t> destroys_;
};


template <typename ... Components>
class TransactionSet {
public:
  TransactionSet(Transaction<Components...> &root, std::atomic<std::uint32_t> &id) : root_(root), id_(id) {}

  // No copying or moving.
  TransactionSet(TransactionSet<Components...> && other) = delete;
  TransactionSet(const TransactionSet<Components...> & other) = delete;
  TransactionSet &operator = (const TransactionSet &other) = delete;
  TransactionSet &operator = (TransactionSet &&other) = delete;

  ~TransactionSet() {
    merge();
  }

  /**
   * Execute a transaction inside a std::async block.
   */
  template <typename Function>
  TransactionSet<Components...> &async(Function f);

  /**
   * Execute a transaction synchronously.
   */
  template <typename Function>
  TransactionSet<Components...> &sync(Function f);

  /**
   * Wait for all outstanding async transactions to complete.
   */
  void wait() {
    for (auto &future : futures_) future.wait();
  }

private:
  friend class Transaction<Components...>;

  /**
   * Transactions call commit() when complete.
   */
  void commit(Commit<Components...> &&commit) {
    std::lock_guard<std::mutex> guard(lock_);
    commits_.push_back(std::move(commit));
  }

  void merge() {
    wait();

    // Transactions have completed.
    std::lock_guard<std::mutex> guard(lock_);

    auto &root = root_.commit_;

    // Merge components back into root.
    int count = 0;
    for (auto &commit : commits_) {
      for (auto &pair : commit.components_) {
        if (!valid_entity(pair.first)) continue;

        // Ensure lookup table has enough capacity.
        if (pair.first >= root.size()) {
          root.lookup_.resize(pair.first + 1, UNDEFINED_ENTITY);
        }
        // Ensure components have storage allocated.
        if (!root.has(pair.first)) {
          root.lookup_[pair.first] = root.components_.size();
          root.components_.emplace_back();
        }
        move<Components...>(pair.second, root[pair.first]);
        count++;
      }
    }

    // Destroy components.
    for (auto &commit : commits_) {
      for (auto id : commit.destroys_) {
        if (root.has(id)) {
          std::uint32_t index = root.lookup_[id];
          root.components_[index].first = DELETED_ENTITY;
          root.components_[index].second = typename Commit<Components...>::Tuple();
          root.lookup_[id] = DELETED_ENTITY;
        }
      }
    }
  }

  template <typename C>
  void move(std::tuple<std::experimental::optional<Components>...> &source, std::tuple<std::experimental::optional<Components>...> &dest) {
    if (std::get<std::experimental::optional<C>>(source))
      std::get<std::experimental::optional<C>>(source) = std::move(std::get<std::experimental::optional<C>>(dest));
  }

  // Move overridden components for a single entity from source transaction into dest.
  template <typename C0, typename C1, typename ... Cn>
  void move(std::tuple<std::experimental::optional<Components>...> &source, std::tuple<std::experimental::optional<Components>...> &dest) {
    move<C0>(source, dest);
    move<C1, Cn...>(source, dest);
  }

  std::mutex lock_;
  Transaction<Components...> &root_;
  std::atomic<std::uint32_t> &id_;
  std::vector<std::future<bool>> futures_;
  std::vector<Commit<Components...>> commits_;
};


template <typename ... Components>
class Transaction {
public:
  friend class TransactionSet<Components...>;

  class Iterator : public std::iterator<std::input_iterator_tag, std::uint32_t> {
  public:
    explicit Iterator(Transaction &tx, bool end) : tx_(tx), id_(end ? tx.commit_.size() : 0)  {
      next();
    }

    Iterator &operator ++() {
      id_++;
      next();
      return *this;
    }

    bool operator == (const Iterator &other) const { return id_ == other.id_; }
    bool operator != (const Iterator &other) const { return id_ != other.id_; }

    Entity operator * () { return Entity(id_); }
    const Entity operator * () const { return Entity(id_); }

  private:
    void next() {
      while (id_ < tx_.commit_.size() && !tx_.exists(id_)) id_++;
    }

    Transaction<Components...> &tx_;
    std::uint32_t id_;
  };

  explicit Transaction(std::atomic<std::uint32_t> &id, TransactionSet<Components...> *set = nullptr, Transaction *parent = nullptr) : set_(set), parent_(parent), id_(id) {
    if (parent) {
      commit_.lookup_ = std::vector<std::uint32_t>(parent->commit_.lookup_.size(), UNDEFINED_ENTITY);
    }
  }

  // This needs to exist, but never gets used. Very odd.
  Transaction(Transaction &&other) : set_(other.set_), parent_(std::move(other.parent_)), id_(other.id_), commit_(std::move(other.commit_)) {
  }

  ~Transaction() {
    if (parent_) commit();
  }

  Iterator begin() {
    return Iterator(*this, false);
  }

  Iterator end() {
    return Iterator(*this, true);
  }

  bool exists(std::uint32_t id) const {
    return commit_.has(id) || (parent_ ? parent_->exists(id) : false);
  }

  void destroy(Entity entity) {
    assert(exists(entity.id));
    commit_.destroys_.push_back(entity.id);
    if (commit_.has(entity.id)) {
      std::uint32_t index = commit_.lookup_[entity.id];
      commit_.components_[index].first = DELETED_ENTITY;
      commit_.components_[index].second = typename Commit<Components...>::Tuple();
      commit_.lookup_[entity.id] = DELETED_ENTITY;
    }
  }

  Entity create() {
    const std::uint32_t id = parent_->next_id();
    commit_.lookup_.resize(id + 1, UNDEFINED_ENTITY);
    commit_.lookup_[id] = commit_.components_.size();
    commit_.components_.emplace_back();
    commit_.components_[commit_.components_.size() - 1].first = id;
    return Entity(id);
  }

  template <typename C, typename ... Args>
  C *assign(Entity entity, Args && ... args) {
    std::get<std::experimental::optional<C>>(commit_[entity.id]).emplace(std::forward<Args>(args)...);
    return &std::get<std::experimental::optional<C>>(commit_[entity.id]).value();
  }

  /**
   * Return immutable component on entity.
   */
  template <typename C>
  const C * component(Entity entity) const {
    if (commit_.has(entity.id)) {
      const C *ours = std::get<C*>(commit_[entity.id]);
      if (ours)
        return ours;
    }
    return parent_ ? parent_->component<C>(entity) : nullptr;
  }

  /**
   * Return a mutable version of component C on entity.
   *
   * Returns nullptr if the component is not assigned to the entity.
   */
  template <typename C>
  C *component(Entity entity) {
    if (commit_.has(entity.id)) {
      auto &ours = std::get<std::experimental::optional<C>>(commit_[entity.id]);
      if (ours)
        return &ours.value();
    }
    C *theirs = parent_ ? parent_->component<C>(entity) : nullptr;
    // If parent has the component, take a copy.
    if (theirs)
      return assign<C>(entity, *theirs);
    return nullptr;
  }

  void abort() {
    commit_.abort();
  }

  void commit() {
    set_->commit(std::move(commit_));
    set_ = nullptr;
    parent_ = nullptr;
  }

private:
  uint32_t next_id() { return id_++; }

  TransactionSet<Components...> *set_;
  Transaction *parent_;
  std::atomic<uint32_t> &id_;
  Commit<Components...> commit_;
};


template <typename ... Components>
template <typename Function>
TransactionSet<Components...> &TransactionSet<Components...>::async(Function f) {
  std::future<bool> result = std::async([this, &f]() -> bool {
    Transaction<Components...> tx(id_, this, &root_);
    f(tx);
    return true;
  });
  std::lock_guard<std::mutex> guard(lock_);
  futures_.push_back(std::move(result));
  return *this;
}

template <typename ... Components>
template <typename Function>
TransactionSet<Components...> &TransactionSet<Components...>::sync(Function f) {
  Transaction<Components...> tx(id_, this, &root_);
  f(tx);
  return *this;
}


/**
 * A wrapper around a shared TransactionSet so we don't have to
 * hack around move semantics.
 */
template <typename ... Components>
class TransactionSetRef {
public:
  TransactionSetRef(Transaction<Components...> &root, std::atomic<std::uint32_t> &id) : set_(std::make_shared<TransactionSet<Components...>>(root, id)) {}

  /**
   * Execute a transaction inside a std::async block.
   */
  template <typename Function>
  TransactionSetRef<Components...> &async(Function f) {
    set_->async(f);
    return *this;
  }

  /**
   * Execute a transaction synchronously.
   */
  template <typename Function>
  TransactionSetRef<Components...> &sync(Function f) {
    set_->sync(f);
    return *this;
  }
private:
  std::shared_ptr<TransactionSet<Components...>> set_;
};


template <typename ... Components>
class EntityManager {
public:
  EntityManager() : id_(0), tx_(id_) {}
  ~EntityManager() {}

  TransactionSetRef<Components...> begin() {
    return TransactionSetRef<Components...>(tx_, id_);
  }

  template <typename F>
  TransactionSetRef<Components...> async(F f) {
    return TransactionSetRef<Components...>(tx_, id_).async(f);
  }

  template <typename F>
  TransactionSetRef<Components...> sync(F f) {
    return TransactionSetRef<Components...>(tx_, id_).sync(f);
  }

private:
  std::atomic<std::uint32_t> id_;
  Transaction<Components...> tx_;
};

struct Position {
  explicit Position(float x = 0, float y = 0) : x(x), y(y) {}
  float x, y;
};

struct Direction {
  explicit Direction(float x = 0, float y = 0) : x(x), y(y) {}
  float x, y;
};


using Entities = EntityManager<Position, Direction>;
using Tx = Transaction<Position, Direction>;


int main() {
  Entities entities;

  srand(time(0));

  // Run two transactions in parallel.
  entities
    .async([](Tx &tx) {
      for (int i = 0; i < 10; i++) {
        Entity entity = tx.create();
        tx.assign<Position>(entity, i, i * 2);
      }

      for (Entity entity : tx) {
        log(std::this_thread::get_id(), "(0): entity = ", entity.id);
      }
      log("DONE 0");
    })
    .async([](Tx &tx) {
      for (int i = 0; i < 10; i++) {
        Entity entity = tx.create();
        tx.assign<Position>(entity, i, i * 2);
        if (rand() % 2 == 0) tx.destroy(entity);
      }
      for (const Entity entity : tx) {
        log(std::this_thread::get_id(), "(1): entity = ", entity.id);
      }
      log("DONE 1");
    });

  log(std::this_thread::get_id(), ": waited");

  // Run a transaction synchronously.
  entities
    .sync([](Tx &tx) {
      for (Entity entity : tx) {
        log(std::this_thread::get_id(), ": entity = ", entity.id);
        if (rand() % 2 == 0) {
          log("destroying ", entity.id);
          tx.destroy(entity);
        }
      }
    });

  log(std::this_thread::get_id(), ": waited");

  // Run another transaction synchronously.
  entities
    .sync([](Tx &tx) {
      for (Entity entity : tx) {
        cerr << std::this_thread::get_id() << ": entity = " << entity.id << endl;
      }
    });

  return 0;
}