Продолжение передачи ресурсов на клиент

Src/Client/Abstract.hpp

@@ -60,11 +60,20 @@ public:
     // states
 };
 
+struct AssetEntry {
+    EnumAssets Type;
+    ResourceId Id;
+    std::string Domain, Key;
+    Resource Res;
+};
+
 /* Интерфейс рендера текущего подключения к серверу */
 class IRenderSession {
 public:
-    // Подгрузка двоичных ресурсов
+    // Изменённые ресурсы (кроме звуков)
-    virtual void onBinaryResourceAdd(std::vector<Hash_t>) = 0;
+    virtual void onAssetsChanges(std::unordered_map<EnumAssets, std::vector<AssetEntry>> resources) = 0;
+    // Потерянные ресурсы
+    virtual void onAssetsLost(std::unordered_map<EnumAssets, std::vector<ResourceId>> resources) = 0;
 
     virtual void onContentDefinesAdd(std::unordered_map<EnumDefContent, std::vector<ResourceId>>) = 0;
     virtual void onContentDefinesLost(std::unordered_map<EnumDefContent, std::vector<ResourceId>>) = 0;
@@ -147,21 +156,6 @@ class IServerSession {
     // };
 
 public:
-    struct AssetEntry {
-        EnumAssets Type;
-        ResourceId Id;
-        std::string Domain, Key;
-        Resource Res;
-    };
-
-    static constexpr uint64_t TIME_BEFORE_UNLOAD_RESOURCE = 180;
-    struct {
-        // Оперируемые ресурсы
-        std::unordered_map<Hash_t, AssetEntry>                       Assets;
-        // Недавно использованные ресурсы, пока хранятся здесь в течении TIME_BEFORE_UNLOAD_RESOURCE секунд
-        std::unordered_map<Hash_t, std::pair<AssetEntry, uint64_t>>  NotInUseAssets;
-    } Binary;
-
     struct {
         std::unordered_map<DefVoxelId, DefVoxel_t>                DefVoxel;
         std::unordered_map<DefNodeId, DefNode_t>                  DefNode;

Src/Client/AssetsManager.cpp

@@ -301,7 +301,7 @@ void AssetsManager::readWriteThread(AsyncUseControl::Lock lock) {
                         }
                         
                         finded = true;
-                        ReadyQueue.lock()->emplace_back(rk.Hash, PathFiles / hashKey.substr(0, 2) / hashKey.substr(2));
+                        ReadyQueue.lock()->emplace_back(rk, PathFiles / hashKey.substr(0, 2) / hashKey.substr(2));
                     } else if(errc != SQLITE_DONE) {
                         sqlite3_reset(STMT_DISK_CONTAINS);
                         MAKE_ERROR("Не удалось выполнить подготовленный запрос STMT_DISK_CONTAINS: " << sqlite3_errmsg(DB));

Src/Client/AssetsManager.hpp

@@ -95,6 +95,7 @@ public:
         Hash_t Hash;
         EnumAssets Type;
         std::string Domain, Key;
+        ResourceId Id;
     };
 
 public:
@@ -115,7 +116,7 @@ public:
     }
 
     // Получить считанные данные
-    std::vector<std::pair<Hash_t, std::optional<Resource>>> pullReads() {
+    std::vector<std::pair<ResourceKey, std::optional<Resource>>> pullReads() {
         return std::move(*ReadyQueue.lock());
     }
 
@@ -181,7 +182,7 @@ private:
     // Очередь на запись ресурсов
     TOS::SpinlockObject<std::queue<Resource>> WriteQueue;
     // Очередь на выдачу результатов чтения
-    TOS::SpinlockObject<std::vector<std::pair<Hash_t, std::optional<Resource>>>> ReadyQueue;
+    TOS::SpinlockObject<std::vector<std::pair<ResourceKey, std::optional<Resource>>>> ReadyQueue;
 
     struct Changes_t {
         std::vector<fs::path> Assets;

Src/Client/ServerSession.cpp

@@ -321,6 +321,147 @@ void ServerSession::onJoystick() {
 }
 
 void ServerSession::atFreeDrawTime(GlobalTime gTime, float dTime) {
+    // Оповещение модуля рендера об изменениях ресурсов
+    std::unordered_map<EnumAssets, std::unordered_map<ResourceId, AssetEntry>> changedResources;
+    std::unordered_map<EnumAssets, std::unordered_set<ResourceId>> lostResources;
+
+    // Обработка полученных ресурсов
+    if(!AsyncContext.LoadedAssets.get_read().empty()) {
+        std::vector<AssetEntry> assets = std::move(*AsyncContext.LoadedAssets.lock());
+        // Для сохранения ресурсов в кеше
+        std::vector<Resource> resources;
+        resources.reserve(assets.size());
+
+
+        for(AssetEntry& entry : assets) {
+            resources.push_back(entry.Res);
+            
+            // Проверяем используется ли сейчас ресурс
+            auto iter = Assets.ExistBinds[(int) entry.Type].find(entry.Id);
+            if(iter == Assets.ExistBinds[(int) entry.Type].end()) {
+                // Не используется
+                Assets.NotInUse[(int) entry.Type][entry.Domain + ':' + entry.Key] = {entry, TIME_BEFORE_UNLOAD_RESOURCE+time(nullptr)};
+            } else {
+                // Используется
+                Assets.InUse[(int) entry.Type][entry.Id] = entry;
+                changedResources[entry.Type].insert({entry.Id, entry});
+            }
+        }
+
+        // Сохраняем в кеш
+        AM->pushResources(std::move(resources));
+    }
+
+    // Обработка полученных тактов
+    while(!AsyncContext.TickSequence.get_read().empty()) {
+        TickData tick;
+        
+        {
+            auto lock = AsyncContext.TickSequence.lock();
+            tick = lock->front();
+            lock->pop();
+        }
+
+        // Потерянные привязки ресурсов
+        for(int type = 0; type < (int) EnumAssets::MAX_ENUM; type++) {
+            for(ResourceId id : tick.AssetsLost[type]) {
+                Assets.ExistBinds[type].erase(id);
+                changedResources[(EnumAssets) type].erase(id);
+            }
+            // Assets.ExistBinds[type].erase(tick.AssetsLost[type].begin(), tick.AssetsLost[type].end());    
+            lostResources[(EnumAssets) type].insert_range(tick.AssetsLost[type]);
+        }
+
+        // Запрос к дисковому кешу
+        std::vector<AssetsManager::ResourceKey> needToLoad;
+        // Новые привязки ресурсов
+        for(const AssetBindEntry& bind : tick.AssetsBinds) {
+            Assets.ExistBinds[(int) bind.Type].insert(bind.Id);
+
+            // Проверить in memory кеш по домену+ключу
+            {
+                std::string dk = bind.Domain + ':' + bind.Key;
+                auto &niubdk = Assets.NotInUse[(int) bind.Type];
+                auto iter = niubdk.find(dk);
+                if(iter != niubdk.end()) {
+                    // Есть ресурс
+                    Assets.InUse[(int) bind.Type][bind.Id] = std::get<0>(iter->second);
+                    changedResources[bind.Type].insert({bind.Id, std::get<0>(iter->second)});
+                    lostResources[bind.Type].erase(bind.Id);
+                    continue;
+                }
+            }
+
+            // Под рукой нет ресурса, отправим на проверку в AssetsManager
+            needToLoad.emplace_back(bind.Hash, bind.Type, bind.Domain, bind.Key, bind.Id);
+        }
+
+        if(!needToLoad.empty())
+            AM->pushReads(std::move(needToLoad));
+    }
+
+    // Получаем ресурсы, загруженные с дискового кеша
+    {
+        std::vector<Hash_t> request;
+        std::vector<std::pair<AssetsManager::ResourceKey, std::optional<Resource>>> resources = AM->pullReads();
+        for(auto& [key, res] : resources) {
+            if(!res) {
+                // Нужно запросить ресурс с сервера
+                request.push_back(key.Hash);
+            } else {
+                auto& a = Assets.ExistBinds[(int) key.Type];
+                AssetEntry entry = {key.Type, key.Id, key.Domain, key.Key, *res};
+
+                if(a.contains(key.Id)) {
+                    // Ресурс ещё нужен
+                    Assets.InUse[(int) key.Type][key.Id] = entry;
+                    changedResources[key.Type].insert({key.Id, entry});
+                } else {
+                    // Ресурс уже не нужен
+                    Assets.NotInUse[(int) key.Type][key.Domain + ':' + key.Key] = {entry, TIME_BEFORE_UNLOAD_RESOURCE+time(nullptr)};
+                }
+            }
+        }
+
+        if(!request.empty()) {
+            assert(request.size() < (1 << 16));
+            Net::Packet p;
+            p << (uint8_t) ToServer::L1::System << (uint8_t) ToServer::L2System::ResourceRequest 
+                << uint16_t(request.size());
+    
+            for(Hash_t& hash : request)
+                p.write((const std::byte*) hash.data(), 32);
+
+            Socket->pushPacket(std::move(p));
+        }
+    }
+
+    if(RS) {
+        // Уведомляем рендер опотерянных и изменённых ресурсах
+        if(!lostResources.empty()) {
+            std::unordered_map<EnumAssets, std::vector<ResourceId>> lostResources2;
+
+            for(auto& [type, list] : lostResources)
+                lostResources2[type].append_range(list);
+            
+            lostResources.clear();
+            RS->onAssetsLost(std::move(lostResources2));
+        }
+
+        if(!changedResources.empty()) {
+            std::unordered_map<EnumAssets, std::vector<AssetEntry>> changedResources2;
+
+            for(auto& [type, list] : changedResources) {
+                auto& a = changedResources2[type];
+                for(auto& [key, val] : list)
+                    a.push_back(val);
+            }
+
+            changedResources.clear();
+            RS->onAssetsChanges(std::move(changedResources2));
+        }
+    }
+
     GTime = gTime;
 
     Pos += glm::vec3(Speed) * dTime;
@@ -539,6 +680,9 @@ coro<> ServerSession::rP_System(Net::AsyncSocket &sock) {
         co_return;
     case ToClient::L2System::UnlinkCamera:
 
+        co_return;
+    case ToClient::L2System::SyncTick:
+        AsyncContext.TickSequence.lock()->push(std::move(AsyncContext.ThisTickEntry));
         co_return;
     default:
         protocolError();
@@ -572,13 +716,12 @@ coro<> ServerSession::rP_Resource(Net::AsyncSocket &sock) {
     case ToClient::L2Resource::Lost:
     {
         uint32_t count = co_await sock.read<uint32_t>();
-        AsyncContext.ThisTickEntry.AssetsLost.reserve(AsyncContext.ThisTickEntry.AssetsLost.size()+count);
 
         for(size_t iter = 0; iter < count; iter++) {
-            // uint8_t type = co_await sock.read<uint8_t>();
+            uint8_t type = co_await sock.read<uint8_t>();
             uint32_t id = co_await sock.read<uint32_t>();
 
-            AsyncContext.ThisTickEntry.AssetsLost.push_back(id);
+            AsyncContext.ThisTickEntry.AssetsLost[(int) type].push_back(id);
         }
     }
     case ToClient::L2Resource::InitResSend:

Src/Client/ServerSession.hpp

@@ -76,6 +76,16 @@ private:
     // Обработчик кеша ресурсов сервера
     AssetsManager::Ptr AM;
 
+    static constexpr uint64_t TIME_BEFORE_UNLOAD_RESOURCE = 180;
+    struct {
+        // Существующие привязки ресурсов
+        std::unordered_set<ResourceId> ExistBinds[(int) EnumAssets::MAX_ENUM];
+        // Используемые в данных момент ресурсы (определяется по действующей привязке)
+        std::unordered_map<ResourceId, AssetEntry> InUse[(int) EnumAssets::MAX_ENUM];
+        // Недавно использованные ресурсы, пока хранятся здесь в течении TIME_BEFORE_UNLOAD_RESOURCE секунд
+        std::unordered_map<std::string, std::pair<AssetEntry, uint64_t>> NotInUse[(int) EnumAssets::MAX_ENUM];
+    } Assets;
+
     struct AssetLoading {
         EnumAssets Type;
         ResourceId Id;
@@ -95,10 +105,10 @@ private:
         std::vector<WorldId_t> LostWorld;
         // std::vector<std::pair<WorldId_t, DefWorld>>
 
-        // Потерянные из видимости ресурсы
-        std::vector<ResourceId> AssetsLost;
         // Новые привязки ресурсов
         std::vector<AssetBindEntry> AssetsBinds;
+        // Потерянные из видимости ресурсы
+        std::vector<ResourceId> AssetsLost[(int) EnumAssets::MAX_ENUM];
     };
 
     struct {
@@ -111,8 +121,6 @@ private:
         TickData ThisTickEntry;
 
     // Обменный пункт
-        // Привязки ресурсов
-        TOS::SpinlockObject<std::vector<AssetEntry>> AssetsBindings;
         // Полученные ресурсы с сервера
         TOS::SpinlockObject<std::vector<AssetEntry>> LoadedAssets;
         // Пакеты обновлений игрового мира

Src/Client/Vulkan/VulkanRenderSession.cpp

@@ -987,7 +987,11 @@ void VulkanRenderSession::init(Vulkan *instance) {
     }
 }
 
-void VulkanRenderSession::onBinaryResourceAdd(std::vector<Hash_t>) {
+void VulkanRenderSession::onAssetsChanges(std::unordered_map<EnumAssets, std::vector<AssetEntry>> resources) {
+
+}
+
+void VulkanRenderSession::onAssetsLost(std::unordered_map<EnumAssets, std::vector<ResourceId>> resources) {
 
 }
 

Src/Client/Vulkan/VulkanRenderSession.hpp

@@ -340,7 +340,8 @@ public:
         assert(serverSession);
     }
 
-    virtual void onBinaryResourceAdd(std::vector<Hash_t>) override;
+    virtual void onAssetsChanges(std::unordered_map<EnumAssets, std::vector<AssetEntry>> resources) override;
+    virtual void onAssetsLost(std::unordered_map<EnumAssets, std::vector<ResourceId>> resources) override;
     virtual void onContentDefinesAdd(std::unordered_map<EnumDefContent, std::vector<ResourceId>>) override;
     virtual void onContentDefinesLost(std::unordered_map<EnumDefContent, std::vector<ResourceId>>) override;
     virtual void onChunksChange(WorldId_t worldId, const std::unordered_set<Pos::GlobalChunk>& changeOrAddList, const std::unordered_set<Pos::GlobalRegion>& remove) override;

Src/Common/Packets.hpp

@@ -76,7 +76,8 @@ enum struct L2System : uint8_t {
     InitEnd,
     Disconnect,
     Test_CAM_PYR_POS,
-    BlockChange
+    BlockChange,
+    ResourceRequest
 };
 
 }
@@ -140,7 +141,8 @@ enum struct L2System : uint8_t {
     Init,
     Disconnect,
     LinkCameraToEntity,
-    UnlinkCamera
+    UnlinkCamera,
+    SyncTick
 };
 
 enum struct L2Resource : uint8_t {

Src/Server/RemoteClient.cpp

@@ -491,6 +491,12 @@ ResourceRequest RemoteClient::pushPreparedPackets() {
         toSend.push_back(std::move(AssetsInWork.AssetsPacket));
     }
 
+    {
+        Net::Packet p;
+        p << (uint8_t) ToClient::L1::System << (uint8_t) ToClient::L2System::SyncTick;
+        toSend.push_back(std::move(p));
+    }
+
     Socket.pushPackets(&toSend);
     toSend.clear();
 
@@ -505,22 +511,27 @@ void RemoteClient::informateAssets(const std::vector<std::tuple<EnumAssets, Reso
 
     for(auto& [type, resId, domain, key, resource] : resources) {
         auto hash = resource.hash();
+        auto lock = NetworkAndResource.lock();
 
         // Проверка запрашиваемых клиентом ресурсов
         {
-            auto iter = std::find(AssetsInWork.ClientRequested.begin(), AssetsInWork.ClientRequested.end(), hash);
+            auto iter = std::find(lock->ClientRequested.begin(), lock->ClientRequested.end(), hash);
-            if(iter != AssetsInWork.ClientRequested.end())
+            if(iter != lock->ClientRequested.end())
             {
+                lock->ClientRequested.erase(iter);
+                lock.unlock();
+
                 auto it = std::lower_bound(AssetsInWork.OnClient.begin(), AssetsInWork.OnClient.end(), hash);
 
                 if(it == AssetsInWork.OnClient.end() || *it != hash)
                     AssetsInWork.OnClient.insert(it, hash);
 
                 AssetsInWork.ToSend.emplace_back(type, domain, key, resId, resource, 0);
+
+                lock = NetworkAndResource.lock();
             }
         }
 
-        auto lock = NetworkAndResource.lock();
         // Информирование клиента о привязках ресурсов к идентификатору
         {
             // Посмотрим что известно клиенту
@@ -528,6 +539,7 @@ void RemoteClient::informateAssets(const std::vector<std::tuple<EnumAssets, Reso
                 iter != lock->ResUses.AssetsUse[(int) type].end() 
                 && std::get<Hash_t>(iter->second) != hash
             ) {
+                lock.unlock();
                 // Требуется перепривязать идентификатор к новому хешу
                 newForClient.push_back({(EnumAssets) type, resId, domain, key, hash, resource.size()});
                 std::get<Hash_t>(iter->second) = hash;
@@ -720,6 +732,23 @@ coro<> RemoteClient::rP_System(Net::AsyncSocket &sock) {
         Actions.lock()->push(action);
         co_return;
     }
+    case ToServer::L2System::ResourceRequest:
+    {
+        uint16_t count = co_await sock.read<uint32_t>();
+        std::vector<Hash_t> hashes;
+        hashes.reserve(count);
+
+        for(int iter = 0; iter < count; iter++) {
+            Hash_t hash;
+            co_await sock.read((std::byte*) hash.data(), 32);
+            hashes.push_back(hash);
+        }
+
+        auto lock = NetworkAndResource.lock();
+        lock->NextRequest.Hashes.append_range(hashes);
+        lock->ClientRequested.append_range(hashes);
+        co_return;
+    }
     default:
         protocolError();
     }
@@ -763,7 +792,7 @@ void RemoteClient::NetworkAndResource_t::decrementAssets(ResUses_t::RefAssets_t&
             << uint32_t(lost.size());
 
         for(auto& [type, id] : lost)
-            NextPacket /* << uint8_t(type)*/  << uint32_t(id);
+            NextPacket << uint8_t(type) << uint32_t(id);
     }
 }
 

Src/Server/RemoteClient.hpp

@@ -268,6 +268,9 @@ class RemoteClient {
 
         // Запрос информации об ассетах и профилях контента
         ResourceRequest NextRequest;
+        // Запрошенные клиентом ресурсы
+        /// TODO: здесь может быть засор
+        std::vector<Hash_t> ClientRequested;
 
         void incrementAssets(const ResUses_t::RefAssets_t& bin);
         void decrementAssets(ResUses_t::RefAssets_t&& bin);
@@ -307,9 +310,8 @@ class RemoteClient {
             + хеш. Если у клиента не окажется этого ресурса, он может его запросить
         */
 
-        // Ресурсы, отправленные на клиент в этой сессии и запрошенные клиентом
+        // Ресурсы, отправленные на клиент в этой сессии
-        /// TODO: ClientRequested здесь может быть засор
+        std::vector<Hash_t> OnClient;
-        std::vector<Hash_t> OnClient, ClientRequested;
         // Отправляемые на клиент ресурсы
         // Тип, домен, ключ, идентификатор, ресурс, количество отправленных байт
         std::vector<std::tuple<EnumAssets, std::string, std::string, ResourceId, Resource, size_t>> ToSend;