ResourceManagerBase.cpp

#include "ResourceManagerBase.h"
#include "Context.h"
#include "Engine.h"
#include "ResourceStore.h"
 
#include "Services/Variables.h"
 
#include "Foundation/Collections/Pool.h"
#include "Foundation/Logging/Logger.h"
#include "Foundation/Platform/FileSystemWatcher.h"
#include "Foundation/Platform/IO.h"
 
#include <queue>
#include <unordered_set>
#include <unordered_map>
#include <utility>
 
namespace
{
  Cogs::Logging::Log logger = Cogs::Logging::getLogger("ResourceManagerBase");
}
 
namespace Cogs::Core
{
  struct ResourcePointerMap
  {
    static constexpr uint32_t NoSlot = static_cast<uint32_t>(-1);
    static constexpr uint32_t NotFree = static_cast<uint32_t>(-2);
 
    struct PointerEntry {
      uint32_t next = NoSlot;
      ResourceBase* resource = nullptr;
    };
 
    uint32_t insert(ResourceBase * ptr)
    {
      uint32_t slot;
      if (free != NoSlot) {
        // Get from free list
        slot = free;
        free = slots[free].next;
      }
      else {
        slot = static_cast<uint32_t>(slots.size());
        slots.emplace_back();
      }
 
      numUsed++;
      slots[slot].next = NotFree;
      slots[slot].resource = ptr;
      return slot;
    }
 
    void remove(uint32_t slot)
    {
      assert(slot != NoSlot && "Invalid slot index Using Freed resource.");
      assert(slot < slots.size() && "Slot index out of range.");
      if (slots[slot].resource == nullptr) {
        assert(false && "Slot index freed.");
        return;
      }
 
      // Add to free-list. Mark freed.
      slots[slot].next = free;
      slots[slot].resource = nullptr;
      free = slot;
      numUsed--;
    }
 
    size_t size() const
    {
      return numUsed;
    }
 
    std::span<const PointerEntry> view() const
    {
      return slots;
    }
 
  private:
    std::vector<PointerEntry> slots;
    uint32_t free = NoSlot;
    size_t numUsed = 0u;
  };
 
  struct ResourceManagerBaseStorage
  {
    ResourceManagerBaseStorage() : resourceInfos(MemBlockType::ResourceManager) {}
 
    Collections::Pool<ResourceInfo> resourceInfos;                    
 
    std::queue<ResourceLoadInfoBase *> loading;                       
    std::queue<ResourceLoadInfoBase *> queued;                        
 
    std::unordered_set<ResourceHandleBase> updatedResources;          
    std::vector<ResourceBase *> orphanedResources;                    
    ResourcePointerMap allocatedResources;                            
 
    std::unordered_map<ResourceId, ResourceHandleBase> resourcesById; 
 
    std::unordered_map<size_t, ResourceBase *> resourcesBySource;     
 
    Mutex updatedMutex;
    Mutex resourceMutex;
    Mutex loadingMutex;
 
    Mutex orphanedMutex;
 
    Atomic<ResourceId> nextResourceId = 1;
  };
}
 
Cogs::Core::ResourceManagerBase::ResourceManagerBase(Context * context) :
  storage(std::make_unique<ResourceManagerBaseStorage>()),
  context(context)
{
 
}
 
Cogs::Core::ResourceManagerBase::~ResourceManagerBase()
{
  assert(storage->updatedResources.size() == 0 && "Updated resources must be cleared before manager is destructed.");
}
 
void Cogs::Core::ResourceManagerBase::initialize()
{
  //NOTE: This section is primarily used to guard against some edge case behavior seen in example
  // code where the engine is reinitialized multiple times in a short timeframe, with potential
  // background processes being run by extensions etc.
  LockGuard lock(storage->updatedMutex);
  LockGuard rlock(storage->resourceMutex);
 
  // Ensure resources allocated and changed previous to the initialize call are put in the
  // update queue.
  for (const auto & entry : storage->allocatedResources.view()) {
    ResourceBase* a = entry.resource;
    if (a != nullptr && a->hasChanged() && !a->isProxy() && a->referenceCount()) {
      storage->updatedResources.insert(generateHandle(a));
    }
  }
}
 
void Cogs::Core::ResourceManagerBase::clear()
{
  storage->nextResourceId = 1;
 
  clearUpdated();
}
 
void Cogs::Core::ResourceManagerBase::reportLeaks(std::string_view typeName)
{
  if (context && context->variables->get("resources.reportLeaks", false)) {
    LockGuard lock(storage->resourceMutex);
 
    for (const auto & entry : storage->allocatedResources.view()) {
      const ResourceBase* resource = entry.resource;
      if (resource && resource->referenceCount()) {
        LOG_WARNING(logger, "%.*s still alive with reference count: %d.", StringViewFormat(typeName), resource->referenceCount());
      }
    }
  }
}
 
std::string Cogs::Core::ResourceManagerBase::getReport()
{
  size_t infos, ids, sources, allocated;
  {
    LockGuard guard(storage->resourceMutex);
    infos = storage->resourceInfos.size();
    ids = storage->resourcesById.size();
    sources = storage->resourcesById.size();
    allocated = storage->allocatedResources.size();
  }
  size_t loading, queued;
  {
    LockGuard guard(storage->loadingMutex);
    loading = storage->loading.size();
    queued = storage->queued.size();
  }
  size_t updated;
  {
    LockGuard guard(storage->updatedMutex);
    updated = storage->updatedResources.size();
  }
  size_t orphaned;
  {
    LockGuard guard(storage->orphanedMutex);
    orphaned = storage->orphanedResources.size();
  }
 
  char buf[1024];
  int rv = std::snprintf(buf, sizeof(buf), "alloc=%3zu, ids=%3zu, sources=%3zu infos=%3zu, load=%3zu, queue=%3zu, update=%3zu, orphan=%3zu", allocated, ids, sources, infos, loading, queued, updated, orphaned);
  if (0 <= rv && static_cast<size_t>(rv) < sizeof(buf)) {
    return std::string(buf);
  }
  else {
    return std::string();
  }
}
 
bool Cogs::Core::ResourceManagerBase::checkPreloaded(ResourceLoadInfoBase * loadInfo)
{
  if (!context->resourceStore->hasResource(loadInfo->resourcePath)) {
    if (!loadInfo->preloading) {
      context->resourceStore->preloadResources({ loadInfo->resourcePath });
      loadInfo->preloading = true;
    }
 
    // Re-queue until all dependencies available.
    queueResource(loadInfo);
 
    return false;
  }
 
  return true;
}
 
void Cogs::Core::ResourceManagerBase::clearUpdated()
{
  {
    LockGuard lock(storage->updatedMutex);
 
    storage->updatedResources.clear();
  }
 
  LockGuard lock(storage->resourceMutex);
 
  storage->resourcesById.clear();
}
 
int Cogs::Core::ResourceManagerBase::getUpdateQuota() const
{
  return context->variables->get("resources.globalFrameUpdateQuota", 0);
}
 
Cogs::Core::ResourceHandleBase Cogs::Core::ResourceManagerBase::releaseInternal(ResourceId resourceId)
{
  LockGuard lock(storage->resourceMutex);
 
  auto found = storage->resourcesById.find(resourceId);
 
  if (found == storage->resourcesById.end()) {
    LOG_ERROR(logger, "Could not release resource %d.", resourceId);
 
    return ResourceHandleBase::NoHandle;
  }
 
  // Remove from handle map. Lookup wil fail.
  // Actual delete will be handled when no active refs left.
  ResourceHandleBase h = found->second;
  storage->resourcesById.erase(found);
  return h;
}
 
Cogs::Core::ResourceHandleBase Cogs::Core::ResourceManagerBase::loadResourceInternal(ResourceLoadInfoBase * loadInfo)
{
  if (loadInfo->isReload()) {
    if (!loadInfo->handle) {
      LOG_ERROR(logger, "Cannot reload invalid (empty) resource.");
      return loadInfo->handle;
    }
 
    loadInfo->handle->unsetFlag(ResourceFlags::Queued);
    loadInfo->handle->setLoading();
 
    {
      LockGuard loadingLock(storage->loadingMutex);
 
      storage->loading.push(loadInfo);
    }
 
    return loadInfo->handle;
  }
 
  // Resolve the resource path to a unique path/protocol combination.
  auto [path, protocol] = context->resourceStore->getResourceLocation(loadInfo->resourcePath);
  loadInfo->protocol = protocol;
 
  LockGuard resourceLock(storage->resourceMutex);
 
  size_t code = Cogs::hash(path);
 
  if (!path.empty()) {
    if (shouldMergeBySource() && (loadInfo->loadFlags & ResourceLoadFlags::ForceUnique) == 0) {
      // Check if resource with the given source path has already been loaded.
      auto found = storage->resourcesBySource.find(code);
 
      if (found != storage->resourcesBySource.end()) {
        if (found->second->referenceCount() > 0) {
          destroyLoadInfoInternal(loadInfo);
          return found->second;
        }
      }
    }
 
    // Use the fixed-up path as resource source/path.
    loadInfo->resourcePath = path;
 
    if (loadInfo->resourceName.empty() && !loadInfo->noDefaultName()) {
      loadInfo->resourceName = IO::fileName(loadInfo->resourcePath);
    }
 
    if (shouldAutoReload()) {
      loadInfo->loadFlags |= ResourceLoadFlags::AutoReload;
    }
  }
 
  if (!loadInfo->handle) {
    loadInfo->handle = getOrCreateInternal(loadInfo->resourceId);
  }
 
  auto resource = loadInfo->handle.get();
 
  assert(!resource->isDeleted() && "Resource pending deletion not valid destination.");
 
  resource->setName(loadInfo->resourceName);
  if ((loadInfo->loadFlags & ResourceLoadFlags::DoNotStoreSource) == 0) {
    resource->setSource(loadInfo->resourcePath);
  }
 
  if (loadInfo->resourcePath.size()) {
    storage->resourcesBySource[code] = resource;
  }
 
  resource->setFlags(loadInfo->resourceFlags);
  resource->unsetFlag(ResourceFlags::Queued);
  resource->setLoading();
 
  if (loadInfo->shouldAutoReload() && loadInfo->protocol == ResourceProtocol::File) {
    auto callback = [this, resource](FileSystemWatcher::Event /*e*/) mutable {
      handleReload(ResourceHandleBase(resource));
    };
 
    if (context->watcher->watchFile(path, callback)) {
      resource->setFlag(ResourceFlags::Watched);
    }
  }
 
  {
    LockGuard loadingLock(storage->loadingMutex);
 
    storage->loading.push(loadInfo);
  }
 
  return loadInfo->handle;
}
 
void Cogs::Core::ResourceManagerBase::processLoading()
{
  UniqueLock loadingLock(storage->loadingMutex);
 
  const size_t maxLoaded = 1000;
 
  size_t loaded = 0;
 
  while (storage->loading.size() && loaded++ < maxLoaded) {
    auto loadInfo = std::move(storage->loading.front());
    storage->loading.pop();
 
    loadInfo->handle->setLoading();
 
    // Must unlock to observe locking order
    loadingLock.unlock();
    handleLoadInternal(loadInfo);
    loadingLock.lock();
  }
 
  while (storage->queued.size()) {
    storage->loading.push(std::move(storage->queued.front()));
    storage->queued.pop();
  }
 
  loadingLock.unlock();
  postProcessLoading();
}
 
void Cogs::Core::ResourceManagerBase::setProcessed(ResourceLoadInfoBase * loadInfo, bool swapping)
{
  auto * resource = loadInfo->handle.get();
  if (resource) {
    resource->unsetFlag(ResourceFlags::Loading);
 
    if (resource->hasFailedLoad()) {
      handleFailedLoadInternal(loadInfo);
      context->engine->invokeResourceLoadCallback((int)resource->getType(), loadInfo->resourceId, 1);
    }
    else {
      if (!swapping) {
        resource->setFlag(ResourceFlags::Loaded);
        resource->setChanged();
      }
      context->engine->invokeResourceLoadCallback((int)resource->getType(), loadInfo->resourceId, 0);
    }
  }
  context->engine->setDirty();
  destroyLoadInfoInternal(loadInfo);
}
 
void Cogs::Core::ResourceManagerBase::activateResources()
{
  std::vector<ResourceHandleBase> updated;
 
  const int quota = getUpdateQuota();
 
  {
    LockGuard updatedLock(storage->updatedMutex);
 
    if (quota>0 && static_cast<size_t>(quota) < storage->updatedResources.size()) {
      while (storage->updatedResources.size() && updated.size() < static_cast<size_t>(quota)) {
        auto it = storage->updatedResources.begin();
        updated.emplace_back(*it);
        storage->updatedResources.erase(it);
      }
    } else {
      updated.assign(storage->updatedResources.begin(), storage->updatedResources.end());
 
      storage->updatedResources.clear();
    }
  }
 
  for (auto & r : updated) {
    auto resource = r.get();
 
    if (r->hasChanged()) {
      const auto result = handleActivationInternal(r, resource);
      resource->unsetFlag(ResourceFlags::Changed);
 
      switch (result) {
      case Cogs::Core::ActivationResult::Failure:
        resource->setFlag(ResourceFlags::FailedActivation);
        break;
      case Cogs::Core::ActivationResult::Postponed:
        // Setting the state to Changed here will automatically re-queue the resource in
        // the updatedResources queue.
        resource->setChanged();
        break;
      case Cogs::Core::ActivationResult::Success:
      default:
        resource->setResident();
 
        if (resource->isDependency()) {
          // Run an additional frame to let code waiting for resource activation detect the
          // active state.
          context->engine->setDirty();
        }
 
        break;
      }
    }
  }
}
 
void Cogs::Core::ResourceManagerBase::fillDeletionQueue(std::vector<ResourceBase *> & deletion)
{
  const int quota = getUpdateQuota();
 
  {
    LockGuard orphanedLock(storage->orphanedMutex);
 
    if (quota>0 && static_cast<size_t>(quota) < storage->orphanedResources.size()) {
      deletion.assign(storage->orphanedResources.begin(), storage->orphanedResources.begin() + quota);
      storage->orphanedResources.erase(storage->orphanedResources.begin(), storage->orphanedResources.begin() + quota);
    } else {
      deletion.assign(std::begin(storage->orphanedResources), std::end(storage->orphanedResources));
      storage->orphanedResources.clear();
    }
  }
}
 
void Cogs::Core::ResourceManagerBase::processDeletion()
{
  std::vector<ResourceBase *> deletion;
 
  fillDeletionQueue(deletion);
 
  {
    LockGuard resourceLock(storage->resourceMutex);
 
    for (auto & resource : deletion) {
#ifdef _DEBUG
      assert(resource->referenceCount() == 0 && "Owned resource may not be destroyed.");
#else
      if (resource->referenceCount() > 0) {
        LOG_WARNING(logger, "Cannot destroy resource with reference count %u.", resource->referenceCount());
        continue;
      }
#endif
      destroy(resource);
    }
  }
}
 
void Cogs::Core::ResourceManagerBase::safeDestroy(ResourceBase * resource)
{
  LockGuard lock(storage->resourceMutex);
 
  destroy(resource);
}
 
void Cogs::Core::ResourceManagerBase::destroy(ResourceBase * resource)
{
  // If the resource has been resurrected, bail out before destruction.
  if (resource->referenceCount()) {
    LOG_WARNING(logger, "Resurrecting resource.");
    return;
  }
 
  storage->allocatedResources.remove(resource->getSlot());
  resource->setSlot(ResourcePointerMap::NoSlot);
 
  if (!resource->isProxy()) {
    handleDeletionInternal(resource);
  }
 
  if (!resource->getSource().empty()) {
    storage->resourcesBySource.erase(Cogs::hash(resource->getSource()));
  }
 
  if (resource->isSet(ResourceFlags::Watched)) {
    context->watcher->unwatchFile(resource->getSource());
  }
 
  resource->setOwner(nullptr);
  resource->unsetFlag((ResourceFlags)~0);
  resource->setFlags(ResourceFlags::Deleted);
 
  if (resource->info) {
    storage->resourceInfos.destroy(resource->info);
    resource->info = nullptr;
  }
 
  destroyInternal(resource);
}
 
void Cogs::Core::ResourceManagerBase::destroyLocked(ResourceBase * resource)
{
  LockGuard lock(storage->resourceMutex);
 
  destroyInternalLocked(resource);
}
 
ResourceId Cogs::Core::ResourceManagerBase::getNextResourceId()
{
  return storage->nextResourceId++;
}
 
void Cogs::Core::ResourceManagerBase::resourceChanged(ResourceBase * resource)
{
  if (resource->isProxy()) return;
 
  LockGuard updatedLock(storage->updatedMutex);
 
  if (resource->referenceCount() == 0) {
    return;
  }
 
  storage->updatedResources.insert(generateHandle(resource));
}
 
void Cogs::Core::ResourceManagerBase::resourceDeleted(ResourceBase * resource)
{
  LockGuard orphanedLock(storage->orphanedMutex);
 
  // Resource is free and marked Orphaned.
  // Still stored in resource lookup tables and must be checked if Orphaned before used.
  storage->orphanedResources.push_back(resource);
}
 
void Cogs::Core::ResourceManagerBase::setResourceId(ResourceBase * resource, ResourceId id)
{
  LockGuard resourceLock(storage->resourceMutex);
 
  resource->setId(id);
 
  storage->resourcesById[id] = resource;
}
 
void Cogs::Core::ResourceManagerBase::queueResource(ResourceLoadInfoBase * loadInfo)
{
  loadInfo->handle->setFlag(ResourceFlags::Queued);
 
  storage->queued.push(loadInfo);
}
 
Cogs::Core::ResourceHandleBase Cogs::Core::ResourceManagerBase::getResourceHandle(ResourceId id) const
{
  LockGuard lock(storage->resourceMutex);
 
  auto it = storage->resourcesById.find(id);
  if (id == NoResourceId || it == storage->resourcesById.end()) {
    LOG_WARNING(logger, "Could not fetch handle for invalid id: %d", id);
    return ResourceHandleBase::NoHandle;
  }
 
  return it->second;
}
 
bool Cogs::Core::ResourceManagerBase::isValidResource(ResourceId id) const
{
  LockGuard lock(storage->resourceMutex);
  return storage->resourcesById.contains(id);
}
 
bool Cogs::Core::ResourceManagerBase::shouldAutoReload() const
{
  return context->variables->get("resources.autoReload", false);
}
 
std::vector<Cogs::Core::ResourceBase *> Cogs::Core::ResourceManagerBase::getAllocatedResources() const
{
  LockGuard lock(storage->resourceMutex);
  auto      v = storage->allocatedResources.view();
  std::vector<Cogs::Core::ResourceBase*> resources;
  resources.reserve(v.size());
  for (const auto entry : v) {
    if (entry.resource) {
      resources.emplace_back(entry.resource);
    }
  }
 
  return resources;
}
 
Cogs::Core::ResourceHandleBase Cogs::Core::ResourceManagerBase::getOrCreate(ResourceId resourceId)
{
  LockGuard resourceLock(storage->resourceMutex);
 
  return getOrCreateInternal(resourceId);
}
 
Cogs::Core::ResourceHandleBase Cogs::Core::ResourceManagerBase::getOrCreateInternal(ResourceId resourceId)
{
  if (resourceId != NoResourceId) {
    auto existingIt = storage->resourcesById.find(resourceId);
 
    if (existingIt != storage->resourcesById.end()) {
      return generateHandle(existingIt->second.get());
    } else {
      auto handle = createResourceInternal();
 
      handle.get()->setId(resourceId);
      storage->resourcesById[resourceId] = handle;
 
      return handle;
    }
  } else {
    return createResourceInternal();
  }
}
 
Cogs::Core::ResourceHandleBase Cogs::Core::ResourceManagerBase::getByName(std::string_view name) const
{
  LockGuard resourceLock(storage->resourceMutex);
 
  for (const auto & r : storage->allocatedResources.view()) {
    if (r.resource && name == r.resource->getName()) {
      if (!r.resource->isOrphaned()) {
        return r.resource;
      }
      else {
        LOG_TRACE(logger, "getByName: Ignoring orphaned resource: %.*s .", StringViewFormat(name));
      }
    }
  }
 
  return ResourceHandleBase::NoHandle;
}
 
Cogs::Core::ResourceHandleBase Cogs::Core::ResourceManagerBase::getBySource(std::string_view source) const
{
  LockGuard resourceLock(storage->resourceMutex);
 
  auto foundIt = storage->resourcesBySource.find(Cogs::hash(source));
 
  if (foundIt != storage->resourcesBySource.end()) {
    if (!foundIt->second->isOrphaned()) {
      return foundIt->second;
    }
  }
  return ResourceHandleBase::NoHandle;
}
 
Cogs::Core::ResourceHandleBase Cogs::Core::ResourceManagerBase::createResource()
{
  LockGuard resourceLock(storage->resourceMutex);
 
  return createResourceInternal();
}
 
Cogs::Core::ResourceHandleBase Cogs::Core::ResourceManagerBase::createResourceInternal()
{
  auto resource = createInternal();
 
  resource->setInfo(storage->resourceInfos.create());
 
  resource->setType(resourceType);
  resource->setOwner(this);
  resource->setInitialized();
 
  // Track allocated resources to be able to check e.g if any are left when this instance is destructed.
  resource->setSlot(storage->allocatedResources.insert(resource));
 
  return ResourceHandleBase(resource);
}
 
Cogs::Core::ResourceHandleBase Cogs::Core::ResourceManagerBase::generateHandle(ResourceBase * resource)
{
  return ResourceHandleBase(resource);
}
 
size_t Cogs::Core::ResourceManagerBase::updatedResourceCount()
{
  LockGuard lock(storage->updatedMutex);
  return storage->updatedResources.size();
}
 
size_t Cogs::Core::ResourceManagerBase::orphanedResourceCount()
{
  LockGuard lock(storage->orphanedMutex);
  return storage->orphanedResources.size();
}
 
size_t Cogs::Core::ResourceManagerBase::allocatedResourceCount()
{
  LockGuard lock(storage->resourceMutex);
  return storage->allocatedResources.size();
}