resource_manager.h

Go to the documentation of this file.

// -----------------------------------------------------------------------------
//
// Copyright (C) 2021 CERN & University of Surrey for the benefit of the
// BioDynaMo collaboration. All Rights Reserved.
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
//
// See the LICENSE file distributed with this work for details.
// See the NOTICE file distributed with this work for additional information
// regarding copyright ownership.
//
// -----------------------------------------------------------------------------
 
#ifndef CORE_RESOURCE_MANAGER_H_
#define CORE_RESOURCE_MANAGER_H_
 
#include <omp.h>
#include <sched.h>
#include <algorithm>
#include <cmath>
#include <limits>
#include <memory>
#include <ostream>
#include <set>
#include <string>
#include <unordered_map>
#include <utility>
#include <vector>
 
#include "core/agent/agent.h"
#include "core/agent/agent_handle.h"
#include "core/agent/agent_uid.h"
#include "core/agent/agent_uid_generator.h"
#include "core/container/agent_uid_map.h"
#include "core/diffusion/continuum_interface.h"
#include "core/diffusion/diffusion_grid.h"
#include "core/functor.h"
#include "core/operation/operation.h"
#include "core/simulation.h"
#include "core/type_index.h"
#include "core/util/numa.h"
#include "core/util/root.h"
#include "core/util/thread_info.h"
#include "core/util/type.h"
 
namespace bdm {
 
class ResourceManager {
 public:
  explicit ResourceManager(TRootIOCtor* r) {}
 
  ResourceManager();
 
  virtual ~ResourceManager();
 
  ResourceManager& operator=(ResourceManager&& other) noexcept {
    if (agents_.size() != other.agents_.size()) {
      Log::Fatal(
          "Restored ResourceManager has different number of NUMA nodes.");
    }
    for (auto& el : continuum_models_) {
      delete el.second;
    }
    for (auto& numa_agents : agents_) {
      for (auto* agent : numa_agents) {
        delete agent;
      }
    }
    agents_ = std::move(other.agents_);
    agents_lb_.resize(agents_.size());
    continuum_models_ = std::move(other.continuum_models_);
 
    RebuildAgentUidMap();
    // restore type_index_
    if (type_index_) {
      for (auto& numa_agents : agents_) {
        for (auto* agent : numa_agents) {
          type_index_->Add(agent);
        }
      }
    }
    return *this;
  }
 
  void RebuildAgentUidMap() {
    // rebuild uid_ah_map_
    uid_ah_map_.clear();
    auto* agent_uid_generator = Simulation::GetActive()->GetAgentUidGenerator();
    uid_ah_map_.resize(agent_uid_generator->GetHighestIndex() + 1);
    for (AgentHandle::NumaNode_t n = 0; n < agents_.size(); ++n) {
      for (AgentHandle::ElementIdx_t i = 0; i < agents_[n].size(); ++i) {
        auto* agent = agents_[n][i];
        this->uid_ah_map_.Insert(agent->GetUid(), AgentHandle(n, i));
      }
    }
  }
 
  Agent* GetAgent(const AgentUid& uid) {
    if (!uid_ah_map_.Contains(uid)) {
      return nullptr;
    }
    auto& ah = uid_ah_map_[uid];
    return agents_[ah.GetNumaNode()][ah.GetElementIdx()];
  }
 
  Agent* GetAgent(AgentHandle ah) {
    return agents_[ah.GetNumaNode()][ah.GetElementIdx()];
  }
 
  AgentHandle GetAgentHandle(const AgentUid& uid) const {
    return uid_ah_map_[uid];
  }
 
  void SwapAgents(std::vector<std::vector<Agent*>>* agents);
 
  [[deprecated("Use AddContinuum() instead")]] void AddDiffusionGrid(
      DiffusionGrid* dgrid) {
    AddContinuum(dgrid);
  }
 
  void AddContinuum(Continuum* cm) {
    auto tmp = cm->GetContinuumId();
    if (tmp < 0) {
      Log::Fatal("ResourceManager::AddContinuum",
                 "You tried to add a continuum model that was not properly "
                 "initialized. "
                 "Continuum id is negative but should be positive. "
                 "Continuum id is " +
                     std::to_string(tmp));
    }
    auto continuum_id = static_cast<uint64_t>(tmp);
    auto search = continuum_models_.find(continuum_id);
    if (search != continuum_models_.end()) {
      Log::Fatal("ResourceManager::AddContinuum",
                 "You tried to add a continuum model with an already existing "
                 "substance id. Please choose a different substance id.");
    } else {
      continuum_models_[continuum_id] = cm;
    }
    MarkEnvironmentOutOfSync();
  }
 
  [[deprecated("Use RemoveContinuum() instead")]] void RemoveDiffusionGrid(
      size_t substance_id) {
    RemoveContinuum(substance_id);
  }
 
  void RemoveContinuum(size_t continuum_id) {
    auto search = continuum_models_.find(continuum_id);
    if (search != continuum_models_.end()) {
      delete search->second;
      continuum_models_.erase(search);
    } else {
      Log::Error("ResourceManager::RemoveContinuum",
                 "You tried to remove a continuum model that does not exist.");
    }
  }
 
  DiffusionGrid* GetDiffusionGrid(size_t substance_id) const {
    auto* cm = GetContinuum(substance_id);
    auto* dgrid = dynamic_cast<DiffusionGrid*>(cm);
    if (!dgrid) {
      Log::Error("ResourceManager::GetDiffusionGrid",
                 "You tried to get a diffusion grid but the substance id "
                 "does not correspond to a diffusion grid.");
    }
    return dgrid;
  }
 
  Continuum* GetContinuum(size_t continuum_id) const {
    auto search = continuum_models_.find(continuum_id);
    if (search != continuum_models_.end()) {
      return search->second;
    } else {
      Log::Error("ResourceManager::GetContinuum",
                 "You tried to request continuum model '", continuum_id,
                 "', but it does not exist! Make sure that it's the correct id "
                 "and that the continuum model is registered.");
      return nullptr;
    }
  }
 
  DiffusionGrid* GetDiffusionGrid(const std::string& substance_name) const {
    auto* cm = GetContinuum(substance_name);
    auto* dgrid = dynamic_cast<DiffusionGrid*>(cm);
    if (!dgrid) {
      Log::Error("ResourceManager::GetDiffusionGrid",
                 "You tried to get a diffusion grid but the substance name "
                 "does not correspond to a diffusion grid.");
    }
    return dgrid;
  }
 
  Continuum* GetContinuum(const std::string& continuum_name) const {
    for (auto& el : continuum_models_) {
      auto& cm = el.second;
      if (cm->GetContinuumName() == continuum_name) {
        return cm;
      }
    }
    Log::Error("ResourceManager::GetContinuum",
               "You tried to request a continuum model named '", continuum_name,
               "', but it does not exist! Make sure that it's spelled "
               "correctly and that the continuum model is registered.");
    return nullptr;
  }
 
  template <typename TFunctor>
  void ForEachDiffusionGrid(TFunctor&& f) const {
    for (auto& el : continuum_models_) {
      auto* dgrid = dynamic_cast<DiffusionGrid*>(el.second);
      if (dgrid) {
        f(dgrid);
      }
    }
  }
 
  template <typename TFunctor>
  void ForEachContinuum(TFunctor&& f) const {
    for (auto& el : continuum_models_) {
      f(el.second);
    }
  }
 
  size_t GetNumAgents(int numa_node = -1) const {
    if (numa_node == -1) {
      size_t num_agents = 0;
      for (auto& numa_agents : agents_) {
        num_agents += numa_agents.size();
      }
      return num_agents;
    } else {
      return agents_[numa_node].size();
    }
  }
 
  size_t GetAgentVectorCapacity(int numa_node);
 
  virtual void ForEachAgent(const std::function<void(Agent*)>& function,
                            Functor<bool, Agent*>* filter = nullptr) {
    for (auto& numa_agents : agents_) {
      for (auto* agent : numa_agents) {
        if (!filter || (filter && (*filter)(agent))) {
          function(agent);
        }
      }
    }
  }
 
  virtual void ForEachAgent(
      const std::function<void(Agent*, AgentHandle)>& function,
      Functor<bool, Agent*>* filter = nullptr) {
    for (AgentHandle::NumaNode_t n = 0; n < agents_.size(); ++n) {
      auto& numa_agents = agents_[n];
      for (AgentHandle::ElementIdx_t i = 0; i < numa_agents.size(); ++i) {
        auto* a = numa_agents[i];
        if (!filter || (filter && (*filter)(a))) {
          function(a, AgentHandle(n, i));
        }
      }
    }
  }
 
  virtual void ForEachAgentParallel(Functor<void, Agent*>& function,
                                    Functor<bool, Agent*>* filter = nullptr);
 
  virtual void ForEachAgentParallel(Operation& op,
                                    Functor<bool, Agent*>* filter = nullptr);
 
  virtual void ForEachAgentParallel(
      Functor<void, Agent*, AgentHandle>& function,
      Functor<bool, Agent*>* filter = nullptr);
 
  virtual void ForEachAgentParallel(
      uint64_t chunk, Functor<void, Agent*, AgentHandle>& function,
      Functor<bool, Agent*>* filter = nullptr);
 
  void Reserve(size_t capacity) {
    for (auto& numa_agents : agents_) {
      numa_agents.reserve(capacity);
    }
    if (type_index_) {
      type_index_->Reserve(capacity);
    }
  }
 
  uint64_t GrowAgentContainer(size_t additional, size_t numa_node) {
    if (additional == 0) {
      return agents_[numa_node].size();
    }
    auto current = agents_[numa_node].size();
    if (current + additional < agents_[numa_node].size()) {
      agents_[numa_node].reserve((current + additional) * 1.5);
    }
    agents_[numa_node].resize(current + additional);
    return current;
  }
 
  bool ContainsAgent(const AgentUid& uid) const {
    return uid_ah_map_.Contains(uid);
  }
 
  void ClearAgents() {
    uid_ah_map_.clear();
    for (auto& numa_agents : agents_) {
      for (auto* agent : numa_agents) {
        delete agent;
      }
      numa_agents.clear();
    }
    if (type_index_) {
      type_index_->Clear();
    }
  }
 
  virtual void LoadBalance();
 
  void DebugNuma() const;
 
  void AddAgent(Agent* agent,  // NOLINT
                typename AgentHandle::NumaNode_t numa_node = 0) {
    auto uid = agent->GetUid();
    if (uid.GetIndex() >= uid_ah_map_.size()) {
      uid_ah_map_.resize(uid.GetIndex() + 1);
    }
    agents_[numa_node].push_back(agent);
    uid_ah_map_.Insert(
        uid, AgentHandle(numa_node, static_cast<AgentHandle::ElementIdx_t>(
                                        agents_[numa_node].size() - 1u)));
    if (type_index_) {
      type_index_->Add(agent);
    }
    MarkEnvironmentOutOfSync();
  }
 
  void ResizeAgentUidMap() {
    auto* agent_uid_generator = Simulation::GetActive()->GetAgentUidGenerator();
    auto highest_idx = agent_uid_generator->GetHighestIndex();
    auto new_size = highest_idx * 1.5 + 1;
    if (highest_idx >= uid_ah_map_.size()) {
      uid_ah_map_.resize(new_size);
    }
    if (type_index_) {
      type_index_->Reserve(new_size);
    }
  }
 
  virtual void EndOfIteration() {}
 
  virtual void AddAgents(typename AgentHandle::NumaNode_t numa_node,
                         uint64_t offset,
                         const std::vector<Agent*>& new_agents) {
    uint64_t i = 0;
    for (auto* agent : new_agents) {
      auto uid = agent->GetUid();
      uid_ah_map_.Insert(
          uid, AgentHandle(numa_node,
                           static_cast<AgentHandle::ElementIdx_t>(offset + i)));
      agents_[numa_node][offset + i] = agent;
      i++;
    }
    if (type_index_) {
#pragma omp critical
      for (auto* agent : new_agents) {
        type_index_->Add(agent);
      }
    }
#pragma omp single
    if (new_agents.size() != 0) {
      MarkEnvironmentOutOfSync();
    }
  }
 
  void RemoveAgent(const AgentUid& uid) {
    // remove from map
    if (uid_ah_map_.Contains(uid)) {
      auto ah = uid_ah_map_[uid];
      uid_ah_map_.Remove(uid);
      // remove from vector
      auto& numa_agents = agents_[ah.GetNumaNode()];
      Agent* agent = nullptr;
      if (ah.GetElementIdx() == numa_agents.size() - 1) {
        agent = numa_agents.back();
        numa_agents.pop_back();
      } else {
        // swap
        agent = numa_agents[ah.GetElementIdx()];
        auto* reordered = numa_agents.back();
        numa_agents[ah.GetElementIdx()] = reordered;
        numa_agents.pop_back();
        uid_ah_map_.Insert(reordered->GetUid(), ah);
      }
      if (type_index_) {
        type_index_->Remove(agent);
      }
      delete agent;
      MarkEnvironmentOutOfSync();
    }
    Simulation::GetActive()->GetAgentUidGenerator()->ReuseAgentUid(uid);
  }
 
  // \param uids: one vector for each thread containing one vector for each numa
  //              node
  void RemoveAgents(const std::vector<std::vector<AgentUid>*>& uids);
 
  const TypeIndex* GetTypeIndex() const { return type_index_; }
 
 protected:
  void MarkEnvironmentOutOfSync() const;
 
  AgentUidMap<AgentHandle> uid_ah_map_ = AgentUidMap<AgentHandle>(100u);  
  std::vector<std::vector<Agent*>> agents_;
  std::vector<std::vector<Agent*>> agents_lb_;  
 
  ThreadInfo* thread_info_ = ThreadInfo::GetInstance();  
 
  TypeIndex* type_index_ = nullptr;
 
  struct ParallelRemovalAuxData {
    std::vector<std::vector<uint64_t>> to_right;
    std::vector<std::vector<uint64_t>> not_to_left;
  };
 
  ParallelRemovalAuxData parallel_remove_;  
 
  friend class SimulationBackup;
  friend std::ostream& operator<<(std::ostream& os, const ResourceManager& rm);
 
 private:
  std::unordered_map<uint64_t, Continuum*> continuum_models_;
 
  BDM_CLASS_DEF_NV(ResourceManager, 2);
};
 
inline std::ostream& operator<<(std::ostream& os, const ResourceManager& rm) {
  os << "\033[1mAgents per numa node\033[0m" << std::endl;
  uint64_t cnt = 0;
  for (auto& numa_agents : rm.agents_) {
    os << "numa node " << cnt++ << " -> size: " << numa_agents.size()
       << std::endl;
  }
  return os;
}
 
}  // namespace bdm
 
#endif  // CORE_RESOURCE_MANAGER_H_