ModelManager.cpp

#include "ModelManager.h"
 
#include "Context.h"
#include "Model.h"
 
#include "DataFetcherManager.h"
#include "ResourceStore.h"
 
#include "Services/TaskManager.h"
#include "Services/Variables.h"
#include "Services/Features.h"
 
#include "Platform/Instrumentation.h"
 
#include "Foundation/Logging/Logger.h"
#include "Foundation/Platform/Timer.h"
 
 
namespace
{
  Cogs::Logging::Log logger = Cogs::Logging::getLogger("ModelManager");
 
  constexpr Cogs::StringView forceFetchesInMainThreadName = "resources.models.forceFetchesInMainThread";
  constexpr Cogs::StringView timeLimitName = "resources.models.mainThreadTimeLimitMs";
  constexpr Cogs::StringView itemLimitName = "resources.models.mainThreadItemLimit";
 
}
 
namespace Cogs::Core
{
  struct ModelManagerData
  {
    TaskId modelsGroup;
 
    std::vector<Model *> deletionModels;
 
    bool updateAsync = false;
  };
}
 
Cogs::Core::ModelManager::ModelManager(Context * context) : ResourceManager(context),
  main(std::this_thread::get_id()),
  data(std::make_unique<ModelManagerData>())
{
  data->modelsGroup = context->taskManager->createGroup(TaskManager::ResourceQueue);
  if (!context->variables->exist(timeLimitName)) {
    context->variables->set(timeLimitName, 1.f);
  }
  if (!context->variables->exist(itemLimitName)) {
    context->variables->set(itemLimitName, 10);
  }
}
 
Cogs::Core::ModelManager::~ModelManager()
{
  context->taskManager->wait(data->modelsGroup);
 
  if (!context->variables->exist(forceFetchesInMainThreadName)) {
    context->variables->set(forceFetchesInMainThreadName, false);
  }
  reportLeaks("Model");
}
 
Cogs::Core::ModelHandle Cogs::Core::ModelManager::loadModel(const StringView & resourceName, ResourceId id, ModelLoadFlags flags)
{
  ModelLoadInfo & loadInfo = *createLoadInfo();
  loadInfo.resourceId = id;
  loadInfo.resourcePath = resourceName.to_string();
  loadInfo.loadFlags = (ResourceLoadFlags)flags;
  loadInfo.modelFlags = flags;
  
  if (context->variables->get("resources.models.autoReload", false)) {
    loadInfo.loadFlags |= ResourceLoadFlags::AutoReload;
  }
 
  return loadResource(&loadInfo);
}
 
void Cogs::Core::ModelManager::postProcessLoading()
{
  if (fetchedItems.empty()) return;
 
  double timeLimitSeconds = 0.001 * context->variables->get(timeLimitName, 0.f);
  int itemLimit = context->variables->get(itemLimitName, 0);
 
  int modelsLoaded = 0;
  Cogs::Timer processTimer = Cogs::Timer::startNew();
  while (!fetchedItems.empty()) {
    FetchedItem item = std::move(fetchedItems.front());
    fetchedItems.pop();
    if (processFetchedItem(item.loadedLoader, item.loadInfo, std::move(item.data))) {
 
      // Model loaded successfully
      modelsLoaded++;
      if ((0 < itemLimit) && (itemLimit <= modelsLoaded)) {
        // Hit item limit, stop processing this frame.
        break;
      }
      if ((0.f < timeLimitSeconds) && (timeLimitSeconds <= processTimer.elapsedSeconds())) {
        // Hit time limit, stop processing this frame.
        break;
      }
    }
  }
 
  // If we are not done, trigger a new frame so we can continue.
  if (!fetchedItems.empty()) {
    // Not done, we need another frame where we can continue.
    context->engine->setDirty();
  }
 
}
 
bool Cogs::Core::ModelManager::processFetchedItem(ILoadedModelLoader* loadedLoader, ModelLoadInfo* loadInfo, std::unique_ptr<FileContents> data)
{
  bool notCancelled = false;
  { // Remove entry of fetch id
    LockGuard guard(fetchIds.lock);
    if (auto it = fetchIds.map.find(reinterpret_cast<size_t>(loadInfo->handle.get())); it != fetchIds.map.end()) {
      fetchIds.map.erase(it);
      notCancelled = true;
    }
  }
  
  bool success = false;
  Model* model = get(loadInfo->handle);
  if (loadInfo->handle->referenceCount() <= 1) {
    LOG_TRACE(logger, "Abandoned model received in async callback, skipping further processing");
    loadInfo->handle->setFailedLoad();
  }
  else if (!data) {
    if (notCancelled) {
      LOG_ERROR(logger, "Error loading model from %s.", loadInfo->resourceName.empty() ? loadInfo->resourcePath.c_str() : loadInfo->resourceName.c_str());
    }
    else {
      LOG_TRACE(logger, "Cancelled model from %s.", loadInfo->resourceName.empty() ? loadInfo->resourcePath.c_str() : loadInfo->resourceName.c_str());
    }
    model->setFailedLoad();
  }
  else if (loadedLoader->load(context, *loadInfo, std::move(data))) {
    success = true;
  }
  else {
    LOG_ERROR(logger, "Error loading model from %s.", loadInfo->resourceName.empty() ? loadInfo->resourcePath.c_str() : loadInfo->resourceName.c_str());
    model->setFailedLoad();
  }
  setProcessed(loadInfo, !loadInfo->loadSync());
  return success;
}
 
 
void Cogs::Core::ModelManager::handleLoad(ModelLoadInfo * loadInfo)
{
  // If no-one has a reference to this model except us, we just abandon it.
  if (loadInfo->handle->referenceCount() <= 1) {
    LOG_TRACE(logger, "Abandoned model, skipping loading");
    loadInfo->handle->setFailedLoad();
    setProcessed(loadInfo, false);
    return;
  }
 
  IModelLoader* loader = nullptr;
  if (!loadInfo->loadSync()
      && context->features->prefers(PlatformPreference::AsyncFetch)
      && !context->resourceStore->hasResource(loadInfo->resourcePath)
      && loadInfo->protocol != ResourceProtocol::Archive)
  {
    // We will try to load this asynchronously if the loader supports it.
 
    std::string path;
    Cogs::FileContentsHints contentsHints = Cogs::FileContentsHints::None;
    ILoadedModelLoader* loadedLoader = nullptr;
 
    // If model has a compressions suffix we can automatically decompress, we can try to
    // load without it
    if (size_t off = loadInfo->resourcePath.find_last_of('.'); off != std::string::npos) {
 
      bool stripSuffix = true;
      switch (Cogs::StringView(loadInfo->resourcePath.data() + off).hashLowercase()) {
      case Cogs::hash(".cogsbin"):
        // We allow .cogsbin files that are really .cogsbin.zst files inside asset hierarchies,
        // see Documentation/AssetSystem.md for more details.
        contentsHints = Cogs::FileContentsHints::ZStdDecompress;
        stripSuffix = false;
        break;
      case Cogs::hash(".zst"):
      case Cogs::hash(".zstd"):
        contentsHints = Cogs::FileContentsHints::ZStdDecompress;
        break;
      case Cogs::hash(".br"):
        contentsHints = Cogs::FileContentsHints::BrotliDecompress;
        break;
      default:
        break;
      }
 
      // Try to find loader that handle file without suffix
      if (contentsHints != Cogs::FileContentsHints::None) {
        if (stripSuffix) {
          path = std::move(loadInfo->resourcePath);
          loadInfo->resourcePath = path.substr(0, off);
        }
        if (loader = findLoader(loadInfo); loader) {
          loadedLoader = dynamic_cast<ILoadedModelLoader*>(loader);
        }
        if (loadedLoader == nullptr) {
          // No supported loader, revert changes
          contentsHints = Cogs::FileContentsHints::None;
          if (stripSuffix) {
            loadInfo->resourcePath = std::move(path);
          }
          path.clear();
          loader = nullptr;
        }
      }
    }
 
    if (loadedLoader == nullptr) {
      // Try to find loader with full suffix
      loader = findLoader(loadInfo);
      if (loader == nullptr) {
        LOG_ERROR(logger, "No loader found for model %s.", loadInfo->resourcePath.c_str());
        Model* model = get(loadInfo->handle);
        model->setFailedLoad();
        return;
      }
      loadedLoader = dynamic_cast<ILoadedModelLoader*>(loader);
    }
 
    if (loadedLoader) {
 
      // We have a loader that supports to get a memory blob instead of reading a file from disk or resource store.
 
      // TODO: Investigate chain of data propagation. There is a copy in the emscripten fetch handler, a subsequent copy
      // when model is loaded. And the loader may spawn tasks (though they are currently synchronous in emscripten), so
      // we really should take ownership somehow.
      FileContents::Callback handleResult = [this, ctx = context, loadedLoader, loadInfo](std::unique_ptr<FileContents> data)
      {
          // If we are in the main thread, we queue the response to be processed during the engine update
          if (main == std::this_thread::get_id()) {
            fetchedItems.push(FetchedItem{ .data = std::move(data), .loadInfo = loadInfo, .loadedLoader = loadedLoader });
            ctx->engine->setDirty();
            return;
          }
          processFetchedItem(loadedLoader, loadInfo, std::move(data));
      };
 
      // Opt-in option to force fetches onto the main thread, useful for debugging that code path on desktop with threads availabe.
      if (context->variables->get(forceFetchesInMainThreadName, false)) {
        FileContents::Callback handleResultInMainThread = [ctx = context, handleResult](std::unique_ptr<FileContents> data) {
          auto task = [handleResult, dataPtr = data.release()]() { handleResult(std::unique_ptr<Cogs::FileContents>(dataPtr)); };
          ctx->engine->runTaskInMainThread(std::move(task));
        };
        handleResult = handleResultInMainThread;
      }
 
      // We have a potential race condition since the callback that removes the
      // cancellation id can either run during the fetch call or after, so we add
      // an item now so we can detect and handle this situation.
      size_t modelKey = reinterpret_cast<size_t>(loadInfo->handle.get());
      {
        LockGuard guard(fetchIds.lock);
        fetchIds.map[modelKey] = DataFetcherManager::NoFetchId;
      }
 
      DataFetcherManager::FetchId fetchId = DataFetcherManager::fetchAsync(context, path.empty() ? loadInfo->resourcePath : path, handleResult, 0, 0, true, contentsHints);
      {
        LockGuard guard(fetchIds.lock);
        if (auto it = fetchIds.map.find(modelKey); it != fetchIds.map.end()) {
          // callback has not yet removed the entry, update it.
          it->second = fetchId;
        }
      }
 
      return; // handleResult continuation will update ResourceState
    }
    else {
 
#ifdef EMSCRIPTEN
      // We have a loader that requires reading from disk or resource store,
      // check if we have it already or trigger a fetch and re-queue model
      if (!checkPreloaded(loadInfo)) {
        return; // Not yet in resource store
      }
#endif
 
      // Model in resource store, load it with the code below
    }
  }
 
  if (loader == nullptr) {
    loader = findLoader(loadInfo);
    if (loader == nullptr) {
      LOG_ERROR(logger, "No loader found for model %s.", loadInfo->resourceName.c_str());
      Model* model = get(loadInfo->handle);
      model->setFailedLoad();
      return;
    }
  }
 
  auto loadResource = [=, this]() mutable
  {
    Model* model = get(loadInfo->handle);
    if (loadInfo->handle->referenceCount() <= 1) {
      LOG_DEBUG(logger, "Abandoned model read, skipping further processing");
      loadInfo->handle->setFailedLoad();
    }
    else {
      bool success = false;
      if (!loader) {
        loader = findLoader(loadInfo);
        if (!loader) {
          LOG_ERROR(logger, "No loader found for model %s.", loadInfo->resourceName.c_str());
        }
      }
 
      if (loader) {
        success = loader->load(context, *loadInfo);
      }
 
      if (success == false) {
        LOG_ERROR(logger, "Error loading model from %s.", loadInfo->resourceName.c_str());
        model->setFailedLoad();
      }
    }
    setProcessed(loadInfo, !loadInfo->loadSync());
  };
 
  if (loadInfo->loadSync() || !context->features->prefers(PlatformPreference::BackgroundTasks)) {
    loadResource();
  }
  else {
    context->taskManager->enqueue(TaskManager::ResourceQueue, loadResource);
  }
}
 
 
void  Cogs::Core::ModelManager::cancelModelLoad(ModelHandle handle)
{
  if (!handle) return;
 
  DataFetcherManager::FetchId id = DataFetcherManager::NoFetchId;
  {
    LockGuard guard(fetchIds.lock);
    if (auto it = fetchIds.map.find(reinterpret_cast<size_t>(handle.get())); it != fetchIds.map.end()) {
      id = it->second;
      fetchIds.map.erase(it);
    }
    else return;
  }
  DataFetcherManager::cancelAsyncFetch(context, id);
}
 
 
void Cogs::Core::ModelManager::handleReload(ResourceHandleBase handle)
{
  ModelHandle model(handle);
 
  ModelLoadInfo & loadInfo = *createLoadInfo();
  loadInfo.resourceId = model->getId();
  loadInfo.resourcePath = model->getSource().to_string();
  loadInfo.resourceName = model->getName().to_string();
  loadInfo.loadFlags = ResourceLoadFlags::Reload;
  loadInfo.handle = model;
 
  loadResource(&loadInfo);
}
 
void Cogs::Core::ModelManager::processDeletion()
{
  CpuInstrumentationScope(SCOPE_RESOURCES, "ModelManager::processDeletion");
 
  data->updateAsync = context->variables->get("resources.models.updateAsync", false);
 
  auto t = Timer::startNew();
 
  ResourceManager::processDeletion();
 
  if (!data->deletionModels.empty()) {
    context->taskManager->enqueueChild(data->modelsGroup, [this, deletionModels = std::move(data->deletionModels)]()
    {
      CpuInstrumentationScope(SCOPE_RESOURCES, "ModelManager::processDeletionTask");
 
      for (auto model : deletionModels) {
        destroyLocked(model);
      }
    });
  }
 
  auto elapsed = t.elapsedSeconds();
 
  if (elapsed > 0.005) {
    LOG_DEBUG(logger, "ModelManager::processDeletion elapsed: %f", elapsed);
  }
}
 
void Cogs::Core::ModelManager::handleDeletion(Model * model)
{
  if (!data->updateAsync) {
    ResourceManager::handleDeletion(model);
  } else {
    data->deletionModels.emplace_back(model);
  }
}
 
void Cogs::Core::ModelManager::destroyInternal(ResourceBase * resource)
{
  if (!data->updateAsync) {
    ResourceManager::destroyInternal(resource);
  }
}