birkeh/o3de
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

Preapre codebase for FileRequest compiletime improvements (#6192)

Browse Source 
* Preapre codebase for FileRequest compiletime improvements

This is preparing grounds for the next PR that will contain the 'meat'
of the changes.

Signed-off-by: nemerle <96597+nemerle@users.noreply.github.com>

* Remove spurious newline.

Signed-off-by: nemerle <96597+nemerle@users.noreply.github.com>
This commit is contained in:
Artur K
2021-12-10 14:08:24 +01:00
committed by GitHub
parent 633a9e939a
commit 53b88d4752
47 changed files with 3739 additions and 3777 deletions
Download Patch File Download Diff File Expand all files Collapse all files
Code/Editor/GameEngine.cpp
+2 -2
View File
@@ -442,7 +442,7 @@ AZ::Outcome<void, AZStd::string> CGameEngine::Init(
REGISTER_COMMAND("quit", CGameEngine::HandleQuitRequest, VF_RESTRICTEDMODE, "Quit/Shutdown the engine");
EBUS_EVENT(CrySystemEventBus, OnCryEditorInitialized);
return AZ::Success();
}
@@ -465,7 +465,7 @@ void CGameEngine::SetLevelPath(const QString& path)
const char* oldExtension = EditorUtils::LevelFile::GetOldCryFileExtension();
const char* defaultExtension = EditorUtils::LevelFile::GetDefaultFileExtension();
// Store off if
// Store off if
if (QFileInfo(path + oldExtension).exists())
{
m_levelExtension = oldExtension;
Code/Framework/AzCore/AzCore/Asset/AssetCommon.h
+14 -14
View File
@@ -168,7 +168,7 @@ namespace AZ
virtual bool IsRegisterReadonlyAndShareable() { return true; }
/**
* Override this function to control automatic reload behavior.
* Override this function to control automatic reload behavior.
* By default, the asset will reload automatically.
* Return false to disable automatic reload. Potential use cases include:
* 1, If an asset is dependent on a parent asset(i.e.both assets need to be reloaded as a group) the parent asset can explicitly reload the child.
@@ -200,10 +200,10 @@ namespace AZ
AssetHandler* m_registeredHandler{ nullptr };
// This is used to identify a unique asset and should only be set by the asset manager
// This is used to identify a unique asset and should only be set by the asset manager
// and therefore does not need to be atomic.
// All shared copy of an asset should have the same identifier and therefore
// should not be modified while making copy of an existing asset.
// should not be modified while making copy of an existing asset.
int m_creationToken = s_defaultCreationToken;
// General purpose flags that should only be accessed within the asset mutex
AZStd::bitset<32> m_flags;
@@ -430,7 +430,7 @@ namespace AZ
*/
void UpgradeAssetInfo();
/**
/**
* for debugging purposes - creates a string that represents the assets id, subid, hint, and name.
* You should use this function for any time you want to show the full details of an asset in a log message
* as it will always produce a consistent output string. By convention, don't surround the output of this call
@@ -586,26 +586,26 @@ namespace AZ
/// Called when an asset is loaded, patched and ready to be used.
virtual void OnAssetReady(Asset<AssetData> asset) { (void)asset; }
/// Called when an asset has been moved (usually due to de-fragmentation/compaction), if possible the only data pointer is provided otherwise NULL.
virtual void OnAssetMoved(Asset<AssetData> asset, void* oldDataPointer) { (void)asset; (void)oldDataPointer; }
/// Called before an asset reload has started.
virtual void OnAssetPreReload(Asset<AssetData> asset) { (void)asset; }
/// Called when an asset has been reloaded (usually in tool mode and loose more). It should not be called in final build.
virtual void OnAssetReloaded(Asset<AssetData> asset) { (void)asset; }
/// Called when an asset failed to reload.
virtual void OnAssetReloadError(Asset<AssetData> asset) { (void)asset; }
/// Called when an asset has been saved. In general most assets can't be saved (in a game) so make sure you check the flag.
virtual void OnAssetSaved(Asset<AssetData> asset, bool isSuccessful) { (void)asset; (void)isSuccessful; }
/// Called when an asset is unloaded.
virtual void OnAssetUnloaded(const AssetId assetId, const AssetType assetType) { (void)assetId; (void)assetType; }
/**
/**
* Called when an error happened with an asset. When this message is received the asset should be considered broken by default.
* Note that this can happen when the asset errors during load, but also happens when the asset is missing (not in catalog etc.)
* in the case of an asset that is completely missing, the Asset<T> passed in here will have no hint or other information about
@@ -1094,7 +1094,7 @@ namespace AZ
// if we are a different asset (or being swapped with a empty) then we just swap as usual.
AZStd::swap(m_assetHint, rhs.m_assetHint);
}
}
//=========================================================================
@@ -1218,7 +1218,7 @@ namespace AZ
/// Indiscriminately skips all asset references.
bool AssetFilterNoAssetLoading(const AssetFilterInfo& filterInfo);
// Shared ProductDependency concepts between AP and LY
// Shared ProductDependency concepts between AP and LY
namespace ProductDependencyInfo
{
//! Corresponds to all ProductDependencyFlags, not just LoadBehaviors
Code/Framework/AzCore/AzCore/Asset/AssetDataStream.cpp
+1 -1
View File
@@ -83,7 +83,7 @@ namespace AZ::Data
AZ_PROFILE_SCOPE(AzCore, "AZ::Data::LoadAssetDataStreamCallback %s",
m_filePath.c_str());
// Get the results
// Get the results
auto streamer = AZ::Interface<AZ::IO::IStreamer>::Get();
AZ::u64 bytesRead = 0;
streamer->GetReadRequestResult(fileHandle, m_buffer, bytesRead,
Code/Framework/AzCore/AzCore/Asset/AssetDataStream.h
+1 -1
View File
@@ -100,7 +100,7 @@ namespace AZ::Data
//! The path and file name of the asset being loaded
AZStd::string m_filePath;
//! The offset into the file to start loading at.
//! The offset into the file to start loading at.
size_t m_fileOffset{ 0 };
//! The amount of data that's expected to be loaded.
Code/Framework/AzCore/AzCore/IO/Streamer/BlockCache.h
+137 -137
View File
@@ -19,144 +19,144 @@
#include <AzCore/std/containers/deque.h>
#include <AzCore/std/smart_ptr/unique_ptr.h>
namespace AZ::IO
{
struct BlockCacheConfig final :
public IStreamerStackConfig
{
AZ_RTTI(AZ::IO::BlockCacheConfig, "{70120525-88A4-40B6-A75B-BAA7E8FD77F3}", IStreamerStackConfig);
AZ_CLASS_ALLOCATOR(BlockCacheConfig, AZ::SystemAllocator, 0);
~BlockCacheConfig() override = default;
AZStd::shared_ptr<StreamStackEntry> AddStreamStackEntry(
const HardwareInformation& hardware, AZStd::shared_ptr<StreamStackEntry> parent) override;
static void Reflect(AZ::ReflectContext* context);
//! Dynamic options for the blocks size.
//! It's possible to set static sizes or use the names from this enum to have AZ::IO::Streamer automatically fill in the sizes.
//! Fixed sizes are set through the Settings Registry with "BlockSize": 524288, while dynamic values are set like
//! "BlockSize": "MemoryAlignment". In the latter case AZ::IO::Streamer will use the available hardware information and fill
//! in the actual value.
enum BlockSize : u32
{
MaxTransfer = AZStd::numeric_limits<u32>::max(), //!< The largest possible block size.
MemoryAlignment = MaxTransfer - 1, //!< The size of the minimal memory requirement of the storage device.
SizeAlignment = MemoryAlignment - 1 //!< The minimal read size required by the storage device.
};
//! The overall size of the cache in megabytes.
u32 m_cacheSizeMib{ 8 };
//! The size of the individual blocks inside the cache.
BlockSize m_blockSize{ BlockSize::MemoryAlignment };
};
class BlockCache
: public StreamStackEntry
{
public:
BlockCache(u64 cacheSize, u32 blockSize, u32 alignment, bool onlyEpilogWrites);
BlockCache(BlockCache&& rhs) = delete;
BlockCache(const BlockCache& rhs) = delete;
~BlockCache() override;
BlockCache& operator=(BlockCache&& rhs) = delete;
BlockCache& operator=(const BlockCache& rhs) = delete;
void QueueRequest(FileRequest* request) override;
bool ExecuteRequests() override;
void UpdateStatus(Status& status) const override;
void UpdateCompletionEstimates(AZStd::chrono::system_clock::time_point now, AZStd::vector<FileRequest*>& internalPending,
StreamerContext::PreparedQueue::iterator pendingBegin, StreamerContext::PreparedQueue::iterator pendingEnd) override;
void AddDelayedRequests(AZStd::vector<FileRequest*>& internalPending);
void UpdatePendingRequestEstimations();
void FlushCache(const RequestPath& filePath);
void FlushEntireCache();
void CollectStatistics(AZStd::vector<Statistic>& statistics) const override;
double CalculateHitRatePercentage() const;
double CalculateCacheableRatePercentage() const;
s32 CalculateAvailableRequestSlots() const;
protected:
static constexpr u32 s_fileNotCached = static_cast<u32>(-1);
enum class CacheResult
{
ReadFromCache, //!< Data was found in the cache and reused.
CacheMiss, //!< Data wasn't found in the cache and no sub request was queued.
Queued, //!< A sub request was created or appended and queued for processing on the next entry in the streamer stack.
Delayed //!< There's no more room to queue a new request, so delay the request until a slot becomes available.
};
struct Section
{
u8* m_output{ nullptr }; //!< The buffer to write the data to.
FileRequest* m_parent{ nullptr }; //!< If set, the file request that is split up by this section.
FileRequest* m_wait{ nullptr }; //!< If set, this contains a "wait"-operation that blocks an operation chain from continuing until this section has been loaded.
u64 m_readOffset{ 0 }; //!< Offset into the file to start reading from.
u64 m_readSize{ 0 }; //!< Number of bytes to read from file.
u64 m_blockOffset{ 0 }; //!< Offset into the cache block to start copying from.
u64 m_copySize{ 0 }; //!< Number of bytes to copy from cache.
u32 m_cacheBlockIndex{ s_fileNotCached }; //!< If assigned, the index of the cache block assigned to this section.
bool m_used{ false }; //!< Whether or not this section is used in further processing.
// Add the provided section in front of this one.
void Prefix(const Section& section);
};
using TimePoint = AZStd::chrono::system_clock::time_point;
void ReadFile(FileRequest* request, FileRequest::ReadData& data);
void ContinueReadFile(FileRequest* request, u64 fileLength);
CacheResult ReadFromCache(FileRequest* request, Section& section, const RequestPath& filePath);
CacheResult ReadFromCache(FileRequest* request, Section& section, u32 cacheBlock);
CacheResult ServiceFromCache(FileRequest* request, Section& section, const RequestPath& filePath, bool sharedRead);
void CompleteRead(FileRequest& request);
bool SplitRequest(Section& prolog, Section& main, Section& epilog, const RequestPath& filePath, u64 fileLength,
u64 offset, u64 size, u8* buffer) const;
u8* GetCacheBlockData(u32 index);
void TouchBlock(u32 index);
AZ::u32 RecycleOldestBlock(const RequestPath& filePath, u64 offset);
u32 FindInCache(const RequestPath& filePath, u64 offset) const;
bool IsCacheBlockInFlight(u32 index) const;
void ResetCacheEntry(u32 index);
void ResetCache();
//! Map of the file requests that are being processed and the sections of the parent requests they'll complete.
AZStd::unordered_multimap<FileRequest*, Section> m_pendingRequests;
//! List of file sections that were delayed because the cache was full.
AZStd::deque<Section> m_delayedSections;
AZ::Statistics::RunningStatistic m_hitRateStat;
AZ::Statistics::RunningStatistic m_cacheableStat;
u8* m_cache;
u64 m_cacheSize;
u32 m_blockSize;
u32 m_alignment;
u32 m_numBlocks;
s32 m_numInFlightRequests{ 0 };
//! The file path associated with a cache block.
AZStd::unique_ptr<RequestPath[]> m_cachedPaths; // Array of m_numBlocks size.
//! The offset into the file the cache blocks starts at.
AZStd::unique_ptr<u64[]> m_cachedOffsets; // Array of m_numBlocks size.
//! The last time the cache block was read from.
AZStd::unique_ptr<TimePoint[]> m_blockLastTouched; // Array of m_numBlocks size.
//! The file request that's currently read data into the cache block. If null, the block has been read.
AZStd::unique_ptr<FileRequest*[]> m_inFlightRequests; // Array of m_numbBlocks size.
//! The number of requests waiting for meta data to be retrieved.
s32 m_numMetaDataRetrievalInProgress{ 0 };
//! Whether or not only the epilog ever writes to the cache.
bool m_onlyEpilogWrites;
};
} // namespace AZ::IO
namespace AZ
{
namespace IO
{
struct BlockCacheConfig final :
public IStreamerStackConfig
{
AZ_RTTI(AZ::IO::BlockCacheConfig, "{70120525-88A4-40B6-A75B-BAA7E8FD77F3}", IStreamerStackConfig);
AZ_CLASS_ALLOCATOR(BlockCacheConfig, AZ::SystemAllocator, 0);
~BlockCacheConfig() override = default;
AZStd::shared_ptr<StreamStackEntry> AddStreamStackEntry(
const HardwareInformation& hardware, AZStd::shared_ptr<StreamStackEntry> parent) override;
static void Reflect(AZ::ReflectContext* context);
//! Dynamic options for the blocks size.
//! It's possible to set static sizes or use the names from this enum to have AZ::IO::Streamer automatically fill in the sizes.
//! Fixed sizes are set through the Settings Registry with "BlockSize": 524288, while dynamic values are set like
//! "BlockSize": "MemoryAlignment". In the latter case AZ::IO::Streamer will use the available hardware information and fill
//! in the actual value.
enum BlockSize : u32
{
MaxTransfer = AZStd::numeric_limits<u32>::max(), //!< The largest possible block size.
MemoryAlignment = MaxTransfer - 1, //!< The size of the minimal memory requirement of the storage device.
SizeAlignment = MemoryAlignment - 1 //!< The minimal read size required by the storage device.
};
//! The overall size of the cache in megabytes.
u32 m_cacheSizeMib{ 8 };
//! The size of the individual blocks inside the cache.
BlockSize m_blockSize{ BlockSize::MemoryAlignment };
};
class BlockCache
: public StreamStackEntry
{
public:
BlockCache(u64 cacheSize, u32 blockSize, u32 alignment, bool onlyEpilogWrites);
BlockCache(BlockCache&& rhs) = delete;
BlockCache(const BlockCache& rhs) = delete;
~BlockCache() override;
BlockCache& operator=(BlockCache&& rhs) = delete;
BlockCache& operator=(const BlockCache& rhs) = delete;
void QueueRequest(FileRequest* request) override;
bool ExecuteRequests() override;
void UpdateStatus(Status& status) const override;
void UpdateCompletionEstimates(AZStd::chrono::system_clock::time_point now, AZStd::vector<FileRequest*>& internalPending,
StreamerContext::PreparedQueue::iterator pendingBegin, StreamerContext::PreparedQueue::iterator pendingEnd) override;
void AddDelayedRequests(AZStd::vector<FileRequest*>& internalPending);
void UpdatePendingRequestEstimations();
void FlushCache(const RequestPath& filePath);
void FlushEntireCache();
void CollectStatistics(AZStd::vector<Statistic>& statistics) const override;
double CalculateHitRatePercentage() const;
double CalculateCacheableRatePercentage() const;
s32 CalculateAvailableRequestSlots() const;
protected:
static constexpr u32 s_fileNotCached = static_cast<u32>(-1);
enum class CacheResult
{
ReadFromCache, //!< Data was found in the cache and reused.
CacheMiss, //!< Data wasn't found in the cache and no sub request was queued.
Queued, //!< A sub request was created or appended and queued for processing on the next entry in the streamer stack.
Delayed //!< There's no more room to queue a new request, so delay the request until a slot becomes available.
};
struct Section
{
u8* m_output{ nullptr }; //!< The buffer to write the data to.
FileRequest* m_parent{ nullptr }; //!< If set, the file request that is split up by this section.
FileRequest* m_wait{ nullptr }; //!< If set, this contains a "wait"-operation that blocks an operation chain from continuing until this section has been loaded.
u64 m_readOffset{ 0 }; //!< Offset into the file to start reading from.
u64 m_readSize{ 0 }; //!< Number of bytes to read from file.
u64 m_blockOffset{ 0 }; //!< Offset into the cache block to start copying from.
u64 m_copySize{ 0 }; //!< Number of bytes to copy from cache.
u32 m_cacheBlockIndex{ s_fileNotCached }; //!< If assigned, the index of the cache block assigned to this section.
bool m_used{ false }; //!< Whether or not this section is used in further processing.
// Add the provided section in front of this one.
void Prefix(const Section& section);
};
using TimePoint = AZStd::chrono::system_clock::time_point;
void ReadFile(FileRequest* request, FileRequest::ReadData& data);
void ContinueReadFile(FileRequest* request, u64 fileLength);
CacheResult ReadFromCache(FileRequest* request, Section& section, const RequestPath& filePath);
CacheResult ReadFromCache(FileRequest* request, Section& section, u32 cacheBlock);
CacheResult ServiceFromCache(FileRequest* request, Section& section, const RequestPath& filePath, bool sharedRead);
void CompleteRead(FileRequest& request);
bool SplitRequest(Section& prolog, Section& main, Section& epilog, const RequestPath& filePath, u64 fileLength,
u64 offset, u64 size, u8* buffer) const;
u8* GetCacheBlockData(u32 index);
void TouchBlock(u32 index);
AZ::u32 RecycleOldestBlock(const RequestPath& filePath, u64 offset);
u32 FindInCache(const RequestPath& filePath, u64 offset) const;
bool IsCacheBlockInFlight(u32 index) const;
void ResetCacheEntry(u32 index);
void ResetCache();
//! Map of the file requests that are being processed and the sections of the parent requests they'll complete.
AZStd::unordered_multimap<FileRequest*, Section> m_pendingRequests;
//! List of file sections that were delayed because the cache was full.
AZStd::deque<Section> m_delayedSections;
AZ::Statistics::RunningStatistic m_hitRateStat;
AZ::Statistics::RunningStatistic m_cacheableStat;
u8* m_cache;
u64 m_cacheSize;
u32 m_blockSize;
u32 m_alignment;
u32 m_numBlocks;
s32 m_numInFlightRequests{ 0 };
//! The file path associated with a cache block.
AZStd::unique_ptr<RequestPath[]> m_cachedPaths; // Array of m_numBlocks size.
//! The offset into the file the cache blocks starts at.
AZStd::unique_ptr<u64[]> m_cachedOffsets; // Array of m_numBlocks size.
//! The last time the cache block was read from.
AZStd::unique_ptr<TimePoint[]> m_blockLastTouched; // Array of m_numBlocks size.
//! The file request that's currently read data into the cache block. If null, the block has been read.
AZStd::unique_ptr<FileRequest*[]> m_inFlightRequests; // Array of m_numbBlocks size.
//! The number of requests waiting for meta data to be retrieved.
s32 m_numMetaDataRetrievalInProgress{ 0 };
//! Whether or not only the epilog ever writes to the cache.
bool m_onlyEpilogWrites;
};
} // namespace IO
AZ_TYPE_INFO_SPECIALIZE(AZ::IO::BlockCacheConfig::BlockSize, "{5D4D597D-4605-462D-A27D-8046115C5381}");
} // namespace AZ
Code/Framework/AzCore/AzCore/IO/Streamer/DedicatedCache.h
+54 -57
View File
@@ -18,77 +18,74 @@
#include <AzCore/std/containers/vector.h>
#include <AzCore/std/smart_ptr/unique_ptr.h>
namespace AZ
namespace AZ::IO
{
namespace IO
struct DedicatedCacheConfig final :
public IStreamerStackConfig
{
struct DedicatedCacheConfig final :
public IStreamerStackConfig
{
AZ_RTTI(AZ::IO::DedicatedCacheConfig, "{DF0F6029-02B0-464C-9846-524654335BCC}", IStreamerStackConfig);
AZ_CLASS_ALLOCATOR(DedicatedCacheConfig, AZ::SystemAllocator, 0);
AZ_RTTI(AZ::IO::DedicatedCacheConfig, "{DF0F6029-02B0-464C-9846-524654335BCC}", IStreamerStackConfig);
AZ_CLASS_ALLOCATOR(DedicatedCacheConfig, AZ::SystemAllocator, 0);
~DedicatedCacheConfig() override = default;
AZStd::shared_ptr<StreamStackEntry> AddStreamStackEntry(
const HardwareInformation& hardware, AZStd::shared_ptr<StreamStackEntry> parent) override;
static void Reflect(AZ::ReflectContext* context);
~DedicatedCacheConfig() override = default;
AZStd::shared_ptr<StreamStackEntry> AddStreamStackEntry(
const HardwareInformation& hardware, AZStd::shared_ptr<StreamStackEntry> parent) override;
static void Reflect(AZ::ReflectContext* context);
//! The size of the individual blocks inside the cache.
BlockCacheConfig::BlockSize m_blockSize{ BlockCacheConfig::BlockSize::MemoryAlignment };
//! The overall size of the cache in megabytes.
u32 m_cacheSizeMib{ 8 };
//! If true, only the epilog is written otherwise the prolog and epilog are written. In either case both prolog and epilog are read.
//! For uses of the cache that read mostly sequentially this flag should be set to true. If reads are more random than it's better
//! to set this flag to false.
bool m_writeOnlyEpilog{ true };
};
//! The size of the individual blocks inside the cache.
BlockCacheConfig::BlockSize m_blockSize{ BlockCacheConfig::BlockSize::MemoryAlignment };
//! The overall size of the cache in megabytes.
u32 m_cacheSizeMib{ 8 };
//! If true, only the epilog is written otherwise the prolog and epilog are written. In either case both prolog and epilog are read.
//! For uses of the cache that read mostly sequentially this flag should be set to true. If reads are more random than it's better
//! to set this flag to false.
bool m_writeOnlyEpilog{ true };
};
class DedicatedCache
: public StreamStackEntry
{
public:
DedicatedCache(u64 cacheSize, u32 blockSize, u32 alignment, bool onlyEpilogWrites);
void SetNext(AZStd::shared_ptr<StreamStackEntry> next) override;
void SetContext(StreamerContext& context) override;
class DedicatedCache
: public StreamStackEntry
{
public:
DedicatedCache(u64 cacheSize, u32 blockSize, u32 alignment, bool onlyEpilogWrites);
void PrepareRequest(FileRequest* request) override;
void QueueRequest(FileRequest* request) override;
bool ExecuteRequests() override;
void SetNext(AZStd::shared_ptr<StreamStackEntry> next) override;
void SetContext(StreamerContext& context) override;
void UpdateStatus(Status& status) const override;
void PrepareRequest(FileRequest* request) override;
void QueueRequest(FileRequest* request) override;
bool ExecuteRequests() override;
void UpdateCompletionEstimates(AZStd::chrono::system_clock::time_point now, AZStd::vector<FileRequest*>& internalPending,
StreamerContext::PreparedQueue::iterator pendingBegin, StreamerContext::PreparedQueue::iterator pendingEnd) override;
void UpdateStatus(Status& status) const override;
void CollectStatistics(AZStd::vector<Statistic>& statistics) const override;
void UpdateCompletionEstimates(AZStd::chrono::system_clock::time_point now, AZStd::vector<FileRequest*>& internalPending,
StreamerContext::PreparedQueue::iterator pendingBegin, StreamerContext::PreparedQueue::iterator pendingEnd) override;
private:
void CreateDedicatedCache(FileRequest* request, FileRequest::CreateDedicatedCacheData& data);
void DestroyDedicatedCache(FileRequest* request, FileRequest::DestroyDedicatedCacheData& data);
void CollectStatistics(AZStd::vector<Statistic>& statistics) const override;
void ReadFile(FileRequest* request, FileRequest::ReadData& data);
size_t FindCache(const RequestPath& filename, FileRange range);
size_t FindCache(const RequestPath& filename, u64 offset);
private:
void CreateDedicatedCache(FileRequest* request, FileRequest::CreateDedicatedCacheData& data);
void DestroyDedicatedCache(FileRequest* request, FileRequest::DestroyDedicatedCacheData& data);
void FlushCache(const RequestPath& filePath);
void FlushEntireCache();
void ReadFile(FileRequest* request, FileRequest::ReadData& data);
size_t FindCache(const RequestPath& filename, FileRange range);
size_t FindCache(const RequestPath& filename, u64 offset);
AZStd::vector<RequestPath> m_cachedFileNames;
AZStd::vector<FileRange> m_cachedFileRanges;
AZStd::vector<AZStd::unique_ptr<BlockCache>> m_cachedFileCaches;
AZStd::vector<size_t> m_cachedFileRefCounts;
void FlushCache(const RequestPath& filePath);
void FlushEntireCache();
AZ::Statistics::RunningStatistic m_usagePercentageStat;
AZStd::vector<RequestPath> m_cachedFileNames;
AZStd::vector<FileRange> m_cachedFileRanges;
AZStd::vector<AZStd::unique_ptr<BlockCache>> m_cachedFileCaches;
AZStd::vector<size_t> m_cachedFileRefCounts;
AZ::Statistics::RunningStatistic m_usagePercentageStat;
#if AZ_STREAMER_ADD_EXTRA_PROFILING_INFO
AZ::Statistics::RunningStatistic m_overallHitRateStat;
AZ::Statistics::RunningStatistic m_overallCacheableRateStat;
AZ::Statistics::RunningStatistic m_overallHitRateStat;
AZ::Statistics::RunningStatistic m_overallCacheableRateStat;
#endif
u64 m_cacheSize;
u32 m_alignment;
u32 m_blockSize;
bool m_onlyEpilogWrites;
};
} // namespace IO
} // namespace AZ
u64 m_cacheSize;
u32 m_alignment;
u32 m_blockSize;
bool m_onlyEpilogWrites;
};
} // namespace AZ::IO
Code/Framework/AzCore/AzCore/IO/Streamer/FileRequest.h
+370 -372
View File
@@ -21,403 +21,401 @@
#include <AzCore/std/smart_ptr/shared_ptr.h>
#include <AzCore/std/string/string_view.h>
namespace AZ
namespace AZ::IO
{
namespace IO
class StreamStackEntry;
class ExternalFileRequest;
using FileRequestPtr = AZStd::intrusive_ptr<ExternalFileRequest>;
class FileRequest final
{
class StreamStackEntry;
class ExternalFileRequest;
public:
inline constexpr static AZStd::chrono::system_clock::time_point s_noDeadlineTime = AZStd::chrono::system_clock::time_point::max();
using FileRequestPtr = AZStd::intrusive_ptr<ExternalFileRequest>;
class FileRequest final
friend class StreamerContext;
friend class ExternalFileRequest;
//! Stores a reference to the external request so it stays alive while the request is being processed.
//! This is needed because Streamer supports fire-and-forget requests since completion can be handled by
//! registering a callback.
struct ExternalRequestData
{
public:
inline constexpr static AZStd::chrono::system_clock::time_point s_noDeadlineTime = AZStd::chrono::system_clock::time_point::max();
inline constexpr static IStreamerTypes::Priority s_orderPriority = IStreamerTypes::s_priorityMedium;
inline constexpr static bool s_failWhenUnhandled = true;
friend class StreamerContext;
friend class ExternalFileRequest;
explicit ExternalRequestData(FileRequestPtr&& request);
//! Stores a reference to the external request so it stays alive while the request is being processed.
//! This is needed because Streamer supports fire-and-forget requests since completion can be handled by
//! registering a callback.
struct ExternalRequestData
{
inline constexpr static IStreamerTypes::Priority s_orderPriority = IStreamerTypes::s_priorityMedium;
inline constexpr static bool s_failWhenUnhandled = true;
FileRequestPtr m_request; //!< The request that was send to Streamer.
};
explicit ExternalRequestData(FileRequestPtr&& request);
FileRequestPtr m_request; //!< The request that was send to Streamer.
};
//! Stores an instance of a RequestPath. To reduce copying instances of a RequestPath functions that
//! need a path take them by reference to the original request. In some cases a path originates from
//! within in the stack and temporary storage is needed. This struct allows for that temporary storage
//! so it can be safely referenced later.
struct RequestPathStoreData
{
inline constexpr static IStreamerTypes::Priority s_orderPriority = IStreamerTypes::s_priorityMedium;
inline constexpr static bool s_failWhenUnhandled = true;
//! Stores an instance of a RequestPath. To reduce copying instances of a RequestPath functions that
//! need a path take them by reference to the original request. In some cases a path originates from
//! within in the stack and temporary storage is needed. This struct allows for that temporary storage
//! so it can be safely referenced later.
struct RequestPathStoreData
{
inline constexpr static IStreamerTypes::Priority s_orderPriority = IStreamerTypes::s_priorityMedium;
inline constexpr static bool s_failWhenUnhandled = true;
explicit RequestPathStoreData(RequestPath path);
explicit RequestPathStoreData(RequestPath path);
RequestPath m_path;
};
RequestPath m_path;
};
//! Request to read data. This is an untranslated request and holds a relative path. The Scheduler
//! will translate this to the appropriate ReadData or CompressedReadData.
struct ReadRequestData
{
inline constexpr static IStreamerTypes::Priority s_orderPriority = IStreamerTypes::s_priorityMedium;
inline constexpr static bool s_failWhenUnhandled = true;
//! Request to read data. This is an untranslated request and holds a relative path. The Scheduler
//! will translate this to the appropriate ReadData or CompressedReadData.
struct ReadRequestData
{
inline constexpr static IStreamerTypes::Priority s_orderPriority = IStreamerTypes::s_priorityMedium;
inline constexpr static bool s_failWhenUnhandled = true;
ReadRequestData(RequestPath path, void* output, u64 outputSize, u64 offset, u64 size,
AZStd::chrono::system_clock::time_point deadline, IStreamerTypes::Priority priority);
ReadRequestData(RequestPath path, IStreamerTypes::RequestMemoryAllocator* allocator, u64 offset, u64 size,
AZStd::chrono::system_clock::time_point deadline, IStreamerTypes::Priority priority);
~ReadRequestData();
RequestPath m_path; //!< Relative path to the target file.
IStreamerTypes::RequestMemoryAllocator* m_allocator; //!< Allocator used to manage the memory for this request.
AZStd::chrono::system_clock::time_point m_deadline; //!< Time by which this request should have been completed.
void* m_output; //!< The memory address assigned (during processing) to store the read data to.
u64 m_outputSize; //!< The memory size of the addressed used to store the read data.
u64 m_offset; //!< The offset in bytes into the file.
u64 m_size; //!< The number of bytes to read from the file.
IStreamerTypes::Priority m_priority; //!< Priority used for ordering requests. This is used when requests have the same deadline.
IStreamerTypes::MemoryType m_memoryType; //!< The type of memory provided by the allocator if used.
};
//! Request to read data. This is a translated request and holds an absolute path and has been
//! resolved to the archive file if needed.
struct ReadData
{
inline constexpr static IStreamerTypes::Priority s_orderPriority = IStreamerTypes::s_priorityMedium;
inline constexpr static bool s_failWhenUnhandled = true;
ReadData(void* output, u64 outputSize, const RequestPath& path, u64 offset, u64 size, bool sharedRead);
const RequestPath& m_path; //!< The path to the file that contains the requested data.
void* m_output; //!< Target output to write the read data to.
u64 m_outputSize; //!< Size of memory m_output points to. This needs to be at least as big as m_size, but can be bigger.
u64 m_offset; //!< The offset in bytes into the file.
u64 m_size; //!< The number of bytes to read from the file.
bool m_sharedRead; //!< True if other code will be reading from the file or the stack entry can exclusively lock.
};
//! Request to read and decompress data.
struct CompressedReadData
{
inline constexpr static IStreamerTypes::Priority s_orderPriority = IStreamerTypes::s_priorityMedium;
inline constexpr static bool s_failWhenUnhandled = true;
CompressedReadData(CompressionInfo&& compressionInfo, void* output, u64 readOffset, u64 readSize);
CompressionInfo m_compressionInfo;
void* m_output; //!< Target output to write the read data to.
u64 m_readOffset; //!< The offset into the decompressed to start copying from.
u64 m_readSize; //!< Number of bytes to read from the decompressed file.
};
//! Holds the progress of an operation chain until this request is explicitly completed.
struct WaitData
{
inline constexpr static IStreamerTypes::Priority s_orderPriority = IStreamerTypes::s_priorityMedium;
inline constexpr static bool s_failWhenUnhandled = true;
};
//! Checks to see if any node in the stack can find a file at the provided path.
struct FileExistsCheckData
{
inline constexpr static IStreamerTypes::Priority s_orderPriority = IStreamerTypes::s_priorityHigh;
inline constexpr static bool s_failWhenUnhandled = false;
explicit FileExistsCheckData(const RequestPath& path);
const RequestPath& m_path;
bool m_found{ false };
};
//! Searches for a file in the stack and retrieves the meta data. This may be slower than a file exists
//! check.
struct FileMetaDataRetrievalData
{
inline constexpr static IStreamerTypes::Priority s_orderPriority = IStreamerTypes::s_priorityHigh;
inline constexpr static bool s_failWhenUnhandled = false;
explicit FileMetaDataRetrievalData(const RequestPath& path);
const RequestPath& m_path;
u64 m_fileSize{ 0 };
bool m_found{ false };
};
//! Cancels a request in the stream stack, if possible.
struct CancelData
{
inline constexpr static IStreamerTypes::Priority s_orderPriority = IStreamerTypes::s_priorityHighest;
inline constexpr static bool s_failWhenUnhandled = false;
explicit CancelData(FileRequestPtr target);
FileRequestPtr m_target; //!< The request that will be canceled.
};
//! Updates the priority and deadline of a request that has not been queued yet.
struct RescheduleData
{
inline constexpr static IStreamerTypes::Priority s_orderPriority = IStreamerTypes::s_priorityHigh;
inline constexpr static bool s_failWhenUnhandled = false;
RescheduleData(FileRequestPtr target, AZStd::chrono::system_clock::time_point newDeadline, IStreamerTypes::Priority newPriority);
FileRequestPtr m_target; //!< The request that will be rescheduled.
AZStd::chrono::system_clock::time_point m_newDeadline; //!< The new deadline for the request.
IStreamerTypes::Priority m_newPriority; //!< The new priority for the request.
};
//! Flushes all references to the provided file in the streaming stack.
struct FlushData
{
inline constexpr static IStreamerTypes::Priority s_orderPriority = IStreamerTypes::s_priorityHigh;
inline constexpr static bool s_failWhenUnhandled = false;
explicit FlushData(RequestPath path);
RequestPath m_path;
};
//! Flushes all caches in the streaming stack.
struct FlushAllData
{
inline constexpr static IStreamerTypes::Priority s_orderPriority = IStreamerTypes::s_priorityHigh;
inline constexpr static bool s_failWhenUnhandled = false;
};
//! Creates a cache dedicated to a single file. This is best used for files where blocks are read from
//! periodically such as audio banks of video files.
struct CreateDedicatedCacheData
{
inline constexpr static IStreamerTypes::Priority s_orderPriority = IStreamerTypes::s_priorityHigh;
inline constexpr static bool s_failWhenUnhandled = false;
CreateDedicatedCacheData(RequestPath path, const FileRange& range);
RequestPath m_path;
FileRange m_range;
};
//! Destroys a cache dedicated to a single file that was previously created by CreateDedicatedCache
struct DestroyDedicatedCacheData
{
inline constexpr static IStreamerTypes::Priority s_orderPriority = IStreamerTypes::s_priorityHigh;
inline constexpr static bool s_failWhenUnhandled = false;
DestroyDedicatedCacheData(RequestPath path, const FileRange& range);
RequestPath m_path;
FileRange m_range;
};
struct ReportData
{
inline constexpr static IStreamerTypes::Priority s_orderPriority = IStreamerTypes::s_priorityLow;
inline constexpr static bool s_failWhenUnhandled = false;
enum class ReportType
{
FileLocks
};
explicit ReportData(ReportType reportType);
ReportType m_reportType;
};
//! Data for a custom command. This can be used by nodes added extensions that need data that can't be stored
//! in the already provided data.
struct CustomData
{
inline constexpr static IStreamerTypes::Priority s_orderPriority = IStreamerTypes::s_priorityMedium;
CustomData(AZStd::any data, bool failWhenUnhandled);
AZStd::any m_data; //!< The data for the custom request.
bool m_failWhenUnhandled; //!< Whether or not the request is marked as failed or success when no node process it.
};
using CommandVariant = AZStd::variant<AZStd::monostate, ExternalRequestData, RequestPathStoreData, ReadRequestData, ReadData,
CompressedReadData, WaitData, FileExistsCheckData, FileMetaDataRetrievalData, CancelData, RescheduleData, FlushData,
FlushAllData, CreateDedicatedCacheData, DestroyDedicatedCacheData, ReportData, CustomData>;
using OnCompletionCallback = AZStd::function<void(FileRequest& request)>;
AZ_CLASS_ALLOCATOR(FileRequest, SystemAllocator, 0);
enum class Usage : u8
{
Internal,
External
};
void CreateRequestLink(FileRequestPtr&& request);
void CreateRequestPathStore(FileRequest* parent, RequestPath path);
void CreateReadRequest(RequestPath path, void* output, u64 outputSize, u64 offset, u64 size,
ReadRequestData(RequestPath path, void* output, u64 outputSize, u64 offset, u64 size,
AZStd::chrono::system_clock::time_point deadline, IStreamerTypes::Priority priority);
void CreateReadRequest(RequestPath path, IStreamerTypes::RequestMemoryAllocator* allocator, u64 offset, u64 size,
ReadRequestData(RequestPath path, IStreamerTypes::RequestMemoryAllocator* allocator, u64 offset, u64 size,
AZStd::chrono::system_clock::time_point deadline, IStreamerTypes::Priority priority);
void CreateRead(FileRequest* parent, void* output, u64 outputSize, const RequestPath& path, u64 offset, u64 size, bool sharedRead = false);
void CreateCompressedRead(FileRequest* parent, const CompressionInfo& compressionInfo, void* output,
u64 readOffset, u64 readSize);
void CreateCompressedRead(FileRequest* parent, CompressionInfo&& compressionInfo, void* output,
u64 readOffset, u64 readSize);
void CreateWait(FileRequest* parent);
void CreateFileExistsCheck(const RequestPath& path);
void CreateFileMetaDataRetrieval(const RequestPath& path);
void CreateCancel(FileRequestPtr target);
void CreateReschedule(FileRequestPtr target, AZStd::chrono::system_clock::time_point newDeadline, IStreamerTypes::Priority newPriority);
void CreateFlush(RequestPath path);
void CreateFlushAll();
void CreateDedicatedCacheCreation(RequestPath path, const FileRange& range = {}, FileRequest* parent = nullptr);
void CreateDedicatedCacheDestruction(RequestPath path, const FileRange& range = {}, FileRequest* parent = nullptr);
void CreateReport(ReportData::ReportType reportType);
void CreateCustom(AZStd::any data, bool failWhenUnhandled = true, FileRequest* parent = nullptr);
~ReadRequestData();
void SetCompletionCallback(OnCompletionCallback callback);
CommandVariant& GetCommand();
const CommandVariant& GetCommand() const;
IStreamerTypes::RequestStatus GetStatus() const;
void SetStatus(IStreamerTypes::RequestStatus newStatus);
FileRequest* GetParent();
const FileRequest* GetParent() const;
size_t GetNumDependencies() const;
static constexpr size_t GetMaxNumDependencies();
//! Whether or not this request should fail if no node in the chain has picked up the request.
bool FailsWhenUnhandled() const;
//! Checks the chain of request for the provided command. Returns the command if found, otherwise null.
template<typename T> T* GetCommandFromChain();
//! Checks the chain of request for the provided command. Returns the command if found, otherwise null.
template<typename T> const T* GetCommandFromChain() const;
//! Determines if this request is contributing to the external request.
bool WorksOn(FileRequestPtr& request) const;
//! Returns the id that's assigned to the request when it was added to the pending queue.
//! The id will always increment so a smaller id means it was originally queued earlier.
size_t GetPendingId() const;
//! Set the estimated completion time for this request and it's immediate parent. The general approach
//! to getting the final estimation is to bubble up the estimation, with ever entry in the stack adding
//! it's own additional delay.
void SetEstimatedCompletion(AZStd::chrono::system_clock::time_point time);
AZStd::chrono::system_clock::time_point GetEstimatedCompletion() const;
private:
explicit FileRequest(Usage usage = Usage::Internal);
~FileRequest();
void Reset();
void SetOptionalParent(FileRequest* parent);
inline static void OnCompletionPlaceholder(const FileRequest& /*request*/) {}
//! Command and parameters for the request.
CommandVariant m_command;
//! Status of the request.
AZStd::atomic<IStreamerTypes::RequestStatus> m_status{ IStreamerTypes::RequestStatus::Pending };
//! Called once the request has completed. This will always be called from the Streamer thread
//! and thread safety is the responsibility of called function. When assigning a lambda avoid
//! capturing a FileRequestPtr by value as this will cause a circular reference which causes
//! the FileRequestPtr to never be released and causes a memory leak. This call will
//! block the main Streamer thread until it returns so callbacks should be kept short. If
//! a longer running task is needed consider using a job to do the work.
OnCompletionCallback m_onCompletion;
//! Estimated time this request will complete. This is an estimation and depends on many
//! factors which can cause it to change drastically from moment to moment.
AZStd::chrono::system_clock::time_point m_estimatedCompletion;
//! The file request that has a dependency on this one. This can be null if there are no
//! other request depending on this one to complete.
FileRequest* m_parent{ nullptr };
//! Id assigned when the request is added to the pending queue.
size_t m_pendingId{ 0 };
//! The number of dependent file request that need to complete before this one is done.
u16 m_dependencies{ 0 };
//! Internal request. If this is true the request is created inside the streaming stack and never
//! leaves it. If true it will automatically be maintained by the scheduler, if false than it's
//! up to the owner to recycle this request.
Usage m_usage{ Usage::Internal };
//! Whether or not this request is currently in a recycle bin. This allows detecting double deletes.
bool m_inRecycleBin{ false };
RequestPath m_path; //!< Relative path to the target file.
IStreamerTypes::RequestMemoryAllocator* m_allocator; //!< Allocator used to manage the memory for this request.
AZStd::chrono::system_clock::time_point m_deadline; //!< Time by which this request should have been completed.
void* m_output; //!< The memory address assigned (during processing) to store the read data to.
u64 m_outputSize; //!< The memory size of the addressed used to store the read data.
u64 m_offset; //!< The offset in bytes into the file.
u64 m_size; //!< The number of bytes to read from the file.
IStreamerTypes::Priority m_priority; //!< Priority used for ordering requests. This is used when requests have the same deadline.
IStreamerTypes::MemoryType m_memoryType; //!< The type of memory provided by the allocator if used.
};
class StreamerContext;
class FileRequestHandle;
//! ExternalFileRequest is a wrapper around the FileRequest so it's safe to use outside the
//! Streaming Stack. The main differences are that ExternalFileRequest is used in a thread-safe
//! context and it doesn't get automatically destroyed upon completion. Instead intrusive_ptr is
//! used to handle clean up.
class ExternalFileRequest final
//! Request to read data. This is a translated request and holds an absolute path and has been
//! resolved to the archive file if needed.
struct ReadData
{
friend struct AZStd::IntrusivePtrCountPolicy<ExternalFileRequest>;
friend class FileRequestHandle;
friend class FileRequest;
friend class Streamer;
friend class StreamerContext;
friend class Scheduler;
friend class Device;
friend bool operator==(const FileRequestHandle& lhs, const FileRequestPtr& rhs);
inline constexpr static IStreamerTypes::Priority s_orderPriority = IStreamerTypes::s_priorityMedium;
inline constexpr static bool s_failWhenUnhandled = true;
public:
AZ_CLASS_ALLOCATOR(ExternalFileRequest, SystemAllocator, 0);
ReadData(void* output, u64 outputSize, const RequestPath& path, u64 offset, u64 size, bool sharedRead);
explicit ExternalFileRequest(StreamerContext* owner);
private:
void add_ref();
void release();
FileRequest m_request;
AZStd::atomic_uint64_t m_refCount{ 0 };
StreamerContext* m_owner;
const RequestPath& m_path; //!< The path to the file that contains the requested data.
void* m_output; //!< Target output to write the read data to.
u64 m_outputSize; //!< Size of memory m_output points to. This needs to be at least as big as m_size, but can be bigger.
u64 m_offset; //!< The offset in bytes into the file.
u64 m_size; //!< The number of bytes to read from the file.
bool m_sharedRead; //!< True if other code will be reading from the file or the stack entry can exclusively lock.
};
class FileRequestHandle
//! Request to read and decompress data.
struct CompressedReadData
{
public:
friend class Streamer;
friend bool operator==(const FileRequestHandle& lhs, const FileRequestPtr& rhs);
inline constexpr static IStreamerTypes::Priority s_orderPriority = IStreamerTypes::s_priorityMedium;
inline constexpr static bool s_failWhenUnhandled = true;
// Intentional cast operator.
FileRequestHandle(FileRequest& request)
: m_request(&request)
{}
CompressedReadData(CompressionInfo&& compressionInfo, void* output, u64 readOffset, u64 readSize);
// Intentional cast operator.
FileRequestHandle(const FileRequestPtr& request)
: m_request(request ? &request->m_request : nullptr)
{}
private:
FileRequest* m_request;
CompressionInfo m_compressionInfo;
void* m_output; //!< Target output to write the read data to.
u64 m_readOffset; //!< The offset into the decompressed to start copying from.
u64 m_readSize; //!< Number of bytes to read from the decompressed file.
};
bool operator==(const FileRequestHandle& lhs, const FileRequestPtr& rhs);
bool operator==(const FileRequestPtr& lhs, const FileRequestHandle& rhs);
bool operator!=(const FileRequestHandle& lhs, const FileRequestPtr& rhs);
bool operator!=(const FileRequestPtr& lhs, const FileRequestHandle& rhs);
} // namespace IO
} // namespace AZ
//! Holds the progress of an operation chain until this request is explicitly completed.
struct WaitData
{
inline constexpr static IStreamerTypes::Priority s_orderPriority = IStreamerTypes::s_priorityMedium;
inline constexpr static bool s_failWhenUnhandled = true;
};
//! Checks to see if any node in the stack can find a file at the provided path.
struct FileExistsCheckData
{
inline constexpr static IStreamerTypes::Priority s_orderPriority = IStreamerTypes::s_priorityHigh;
inline constexpr static bool s_failWhenUnhandled = false;
explicit FileExistsCheckData(const RequestPath& path);
const RequestPath& m_path;
bool m_found{ false };
};
//! Searches for a file in the stack and retrieves the meta data. This may be slower than a file exists
//! check.
struct FileMetaDataRetrievalData
{
inline constexpr static IStreamerTypes::Priority s_orderPriority = IStreamerTypes::s_priorityHigh;
inline constexpr static bool s_failWhenUnhandled = false;
explicit FileMetaDataRetrievalData(const RequestPath& path);
const RequestPath& m_path;
u64 m_fileSize{ 0 };
bool m_found{ false };
};
//! Cancels a request in the stream stack, if possible.
struct CancelData
{
inline constexpr static IStreamerTypes::Priority s_orderPriority = IStreamerTypes::s_priorityHighest;
inline constexpr static bool s_failWhenUnhandled = false;
explicit CancelData(FileRequestPtr target);
FileRequestPtr m_target; //!< The request that will be canceled.
};
//! Updates the priority and deadline of a request that has not been queued yet.
struct RescheduleData
{
inline constexpr static IStreamerTypes::Priority s_orderPriority = IStreamerTypes::s_priorityHigh;
inline constexpr static bool s_failWhenUnhandled = false;
RescheduleData(FileRequestPtr target, AZStd::chrono::system_clock::time_point newDeadline, IStreamerTypes::Priority newPriority);
FileRequestPtr m_target; //!< The request that will be rescheduled.
AZStd::chrono::system_clock::time_point m_newDeadline; //!< The new deadline for the request.
IStreamerTypes::Priority m_newPriority; //!< The new priority for the request.
};
//! Flushes all references to the provided file in the streaming stack.
struct FlushData
{
inline constexpr static IStreamerTypes::Priority s_orderPriority = IStreamerTypes::s_priorityHigh;
inline constexpr static bool s_failWhenUnhandled = false;
explicit FlushData(RequestPath path);
RequestPath m_path;
};
//! Flushes all caches in the streaming stack.
struct FlushAllData
{
inline constexpr static IStreamerTypes::Priority s_orderPriority = IStreamerTypes::s_priorityHigh;
inline constexpr static bool s_failWhenUnhandled = false;
};
//! Creates a cache dedicated to a single file. This is best used for files where blocks are read from
//! periodically such as audio banks of video files.
struct CreateDedicatedCacheData
{
inline constexpr static IStreamerTypes::Priority s_orderPriority = IStreamerTypes::s_priorityHigh;
inline constexpr static bool s_failWhenUnhandled = false;
CreateDedicatedCacheData(RequestPath path, const FileRange& range);
RequestPath m_path;
FileRange m_range;
};
//! Destroys a cache dedicated to a single file that was previously created by CreateDedicatedCache
struct DestroyDedicatedCacheData
{
inline constexpr static IStreamerTypes::Priority s_orderPriority = IStreamerTypes::s_priorityHigh;
inline constexpr static bool s_failWhenUnhandled = false;
DestroyDedicatedCacheData(RequestPath path, const FileRange& range);
RequestPath m_path;
FileRange m_range;
};
struct ReportData
{
inline constexpr static IStreamerTypes::Priority s_orderPriority = IStreamerTypes::s_priorityLow;
inline constexpr static bool s_failWhenUnhandled = false;
enum class ReportType
{
FileLocks
};
explicit ReportData(ReportType reportType);
ReportType m_reportType;
};
//! Data for a custom command. This can be used by nodes added extensions that need data that can't be stored
//! in the already provided data.
struct CustomData
{
inline constexpr static IStreamerTypes::Priority s_orderPriority = IStreamerTypes::s_priorityMedium;
CustomData(AZStd::any data, bool failWhenUnhandled);
AZStd::any m_data; //!< The data for the custom request.
bool m_failWhenUnhandled; //!< Whether or not the request is marked as failed or success when no node process it.
};
using CommandVariant = AZStd::variant<AZStd::monostate, ExternalRequestData, RequestPathStoreData, ReadRequestData, ReadData,
CompressedReadData, WaitData, FileExistsCheckData, FileMetaDataRetrievalData, CancelData, RescheduleData, FlushData,
FlushAllData, CreateDedicatedCacheData, DestroyDedicatedCacheData, ReportData, CustomData>;
using OnCompletionCallback = AZStd::function<void(FileRequest& request)>;
AZ_CLASS_ALLOCATOR(FileRequest, SystemAllocator, 0);
enum class Usage : u8
{
Internal,
External
};
void CreateRequestLink(FileRequestPtr&& request);
void CreateRequestPathStore(FileRequest* parent, RequestPath path);
void CreateReadRequest(RequestPath path, void* output, u64 outputSize, u64 offset, u64 size,
AZStd::chrono::system_clock::time_point deadline, IStreamerTypes::Priority priority);
void CreateReadRequest(RequestPath path, IStreamerTypes::RequestMemoryAllocator* allocator, u64 offset, u64 size,
AZStd::chrono::system_clock::time_point deadline, IStreamerTypes::Priority priority);
void CreateRead(FileRequest* parent, void* output, u64 outputSize, const RequestPath& path, u64 offset, u64 size, bool sharedRead = false);
void CreateCompressedRead(FileRequest* parent, const CompressionInfo& compressionInfo, void* output,
u64 readOffset, u64 readSize);
void CreateCompressedRead(FileRequest* parent, CompressionInfo&& compressionInfo, void* output,
u64 readOffset, u64 readSize);
void CreateWait(FileRequest* parent);
void CreateFileExistsCheck(const RequestPath& path);
void CreateFileMetaDataRetrieval(const RequestPath& path);
void CreateCancel(FileRequestPtr target);
void CreateReschedule(FileRequestPtr target, AZStd::chrono::system_clock::time_point newDeadline, IStreamerTypes::Priority newPriority);
void CreateFlush(RequestPath path);
void CreateFlushAll();
void CreateDedicatedCacheCreation(RequestPath path, const FileRange& range = {}, FileRequest* parent = nullptr);
void CreateDedicatedCacheDestruction(RequestPath path, const FileRange& range = {}, FileRequest* parent = nullptr);
void CreateReport(ReportData::ReportType reportType);
void CreateCustom(AZStd::any data, bool failWhenUnhandled = true, FileRequest* parent = nullptr);
void SetCompletionCallback(OnCompletionCallback callback);
CommandVariant& GetCommand();
const CommandVariant& GetCommand() const;
IStreamerTypes::RequestStatus GetStatus() const;
void SetStatus(IStreamerTypes::RequestStatus newStatus);
FileRequest* GetParent();
const FileRequest* GetParent() const;
size_t GetNumDependencies() const;
static constexpr size_t GetMaxNumDependencies();
//! Whether or not this request should fail if no node in the chain has picked up the request.
bool FailsWhenUnhandled() const;
//! Checks the chain of request for the provided command. Returns the command if found, otherwise null.
template<typename T> T* GetCommandFromChain();
//! Checks the chain of request for the provided command. Returns the command if found, otherwise null.
template<typename T> const T* GetCommandFromChain() const;
//! Determines if this request is contributing to the external request.
bool WorksOn(FileRequestPtr& request) const;
//! Returns the id that's assigned to the request when it was added to the pending queue.
//! The id will always increment so a smaller id means it was originally queued earlier.
size_t GetPendingId() const;
//! Set the estimated completion time for this request and it's immediate parent. The general approach
//! to getting the final estimation is to bubble up the estimation, with ever entry in the stack adding
//! it's own additional delay.
void SetEstimatedCompletion(AZStd::chrono::system_clock::time_point time);
AZStd::chrono::system_clock::time_point GetEstimatedCompletion() const;
private:
explicit FileRequest(Usage usage = Usage::Internal);
~FileRequest();
void Reset();
void SetOptionalParent(FileRequest* parent);
inline static void OnCompletionPlaceholder(const FileRequest& /*request*/) {}
//! Command and parameters for the request.
CommandVariant m_command;
//! Status of the request.
AZStd::atomic<IStreamerTypes::RequestStatus> m_status{ IStreamerTypes::RequestStatus::Pending };
//! Called once the request has completed. This will always be called from the Streamer thread
//! and thread safety is the responsibility of called function. When assigning a lambda avoid
//! capturing a FileRequestPtr by value as this will cause a circular reference which causes
//! the FileRequestPtr to never be released and causes a memory leak. This call will
//! block the main Streamer thread until it returns so callbacks should be kept short. If
//! a longer running task is needed consider using a job to do the work.
OnCompletionCallback m_onCompletion;
//! Estimated time this request will complete. This is an estimation and depends on many
//! factors which can cause it to change drastically from moment to moment.
AZStd::chrono::system_clock::time_point m_estimatedCompletion;
//! The file request that has a dependency on this one. This can be null if there are no
//! other request depending on this one to complete.
FileRequest* m_parent{ nullptr };
//! Id assigned when the request is added to the pending queue.
size_t m_pendingId{ 0 };
//! The number of dependent file request that need to complete before this one is done.
u16 m_dependencies{ 0 };
//! Internal request. If this is true the request is created inside the streaming stack and never
//! leaves it. If true it will automatically be maintained by the scheduler, if false than it's
//! up to the owner to recycle this request.
Usage m_usage{ Usage::Internal };
//! Whether or not this request is currently in a recycle bin. This allows detecting double deletes.
bool m_inRecycleBin{ false };
};
class StreamerContext;
class FileRequestHandle;
//! ExternalFileRequest is a wrapper around the FileRequest so it's safe to use outside the
//! Streaming Stack. The main differences are that ExternalFileRequest is used in a thread-safe
//! context and it doesn't get automatically destroyed upon completion. Instead intrusive_ptr is
//! used to handle clean up.
class ExternalFileRequest final
{
friend struct AZStd::IntrusivePtrCountPolicy<ExternalFileRequest>;
friend class FileRequestHandle;
friend class FileRequest;
friend class Streamer;
friend class StreamerContext;
friend class Scheduler;
friend class Device;
friend bool operator==(const FileRequestHandle& lhs, const FileRequestPtr& rhs);
public:
AZ_CLASS_ALLOCATOR(ExternalFileRequest, SystemAllocator, 0);
explicit ExternalFileRequest(StreamerContext* owner);
private:
void add_ref();
void release();
FileRequest m_request;
AZStd::atomic_uint64_t m_refCount{ 0 };
StreamerContext* m_owner;
};
class FileRequestHandle
{
public:
friend class Streamer;
friend bool operator==(const FileRequestHandle& lhs, const FileRequestPtr& rhs);
// Intentional cast operator.
FileRequestHandle(FileRequest& request)
: m_request(&request)
{}
// Intentional cast operator.
FileRequestHandle(const FileRequestPtr& request)
: m_request(request ? &request->m_request : nullptr)
{}
private:
FileRequest* m_request;
};
bool operator==(const FileRequestHandle& lhs, const FileRequestPtr& rhs);
bool operator==(const FileRequestPtr& lhs, const FileRequestHandle& rhs);
bool operator!=(const FileRequestHandle& lhs, const FileRequestPtr& rhs);
bool operator!=(const FileRequestPtr& lhs, const FileRequestHandle& rhs);
} // namespace AZ::IO
#include <AzCore/IO/Streamer/FileRequest.inl>
Code/Framework/AzCore/AzCore/IO/Streamer/FullFileDecompressor.h
+94 -97
View File
@@ -19,118 +19,115 @@
#include <AzCore/std/smart_ptr/unique_ptr.h>
#include <AzCore/Statistics/RunningStatistic.h>
namespace AZ
namespace AZ::IO
{
namespace IO
struct FullFileDecompressorConfig final :
public IStreamerStackConfig
{
struct FullFileDecompressorConfig final :
public IStreamerStackConfig
AZ_RTTI(AZ::IO::FullFileDecompressorConfig, "{C96B7EC1-8C73-4493-A7CB-66F5D550FC3A}", IStreamerStackConfig);
AZ_CLASS_ALLOCATOR(FullFileDecompressorConfig, AZ::SystemAllocator, 0);
~FullFileDecompressorConfig() override = default;
AZStd::shared_ptr<StreamStackEntry> AddStreamStackEntry(
const HardwareInformation& hardware, AZStd::shared_ptr<StreamStackEntry> parent) override;
static void Reflect(AZ::ReflectContext* context);
//! Maximum number of reads that are kept in flight.
u32 m_maxNumReads{ 2 };
//! Maximum number of decompression jobs that can run simultaneously.
u32 m_maxNumJobs{ 2 };
};
//! Entry in the streaming stack that decompresses files from an archive that are stored
//! as single files and without equally distributed seek points.
//! Because the target archive has compressed the entire file, it needs to be decompressed
//! completely, so even if the file is partially read, it needs to be fully loaded. This
//! also means that there's no upper limit to the memory so every decompression job will
//! need to allocate memory as a temporary buffer (in-place decompression is not supported).
//! Finally, the lack of an upper limit also means that the duration of the decompression job
//! can vary largely so a dedicated job system is used to decompress on to avoid blocking
//! the main job system from working.
class FullFileDecompressor
: public StreamStackEntry
{
public:
FullFileDecompressor(u32 maxNumReads, u32 maxNumJobs, u32 alignment);
~FullFileDecompressor() override = default;
void PrepareRequest(FileRequest* request) override;
void QueueRequest(FileRequest* request) override;
bool ExecuteRequests() override;
void UpdateStatus(Status& status) const override;
void UpdateCompletionEstimates(AZStd::chrono::system_clock::time_point now, AZStd::vector<FileRequest*>& internalPending,
StreamerContext::PreparedQueue::iterator pendingBegin, StreamerContext::PreparedQueue::iterator pendingEnd) override;
void CollectStatistics(AZStd::vector<Statistic>& statistics) const override;
private:
using Buffer = u8*;
enum class ReadBufferStatus : uint8_t
{
AZ_RTTI(AZ::IO::FullFileDecompressorConfig, "{C96B7EC1-8C73-4493-A7CB-66F5D550FC3A}", IStreamerStackConfig);
AZ_CLASS_ALLOCATOR(FullFileDecompressorConfig, AZ::SystemAllocator, 0);
~FullFileDecompressorConfig() override = default;
AZStd::shared_ptr<StreamStackEntry> AddStreamStackEntry(
const HardwareInformation& hardware, AZStd::shared_ptr<StreamStackEntry> parent) override;
static void Reflect(AZ::ReflectContext* context);
//! Maximum number of reads that are kept in flight.
u32 m_maxNumReads{ 2 };
//! Maximum number of decompression jobs that can run simultaneously.
u32 m_maxNumJobs{ 2 };
Unused,
ReadInFlight,
PendingDecompression
};
//! Entry in the streaming stack that decompresses files from an archive that are stored
//! as single files and without equally distributed seek points.
//! Because the target archive has compressed the entire file, it needs to be decompressed
//! completely, so even if the file is partially read, it needs to be fully loaded. This
//! also means that there's no upper limit to the memory so every decompression job will
//! need to allocate memory as a temporary buffer (in-place decompression is not supported).
//! Finally, the lack of an upper limit also means that the duration of the decompression job
//! can vary largely so a dedicated job system is used to decompress on to avoid blocking
//! the main job system from working.
class FullFileDecompressor
: public StreamStackEntry
struct DecompressionInformation
{
public:
FullFileDecompressor(u32 maxNumReads, u32 maxNumJobs, u32 alignment);
~FullFileDecompressor() override = default;
bool IsProcessing() const;
void PrepareRequest(FileRequest* request) override;
void QueueRequest(FileRequest* request) override;
bool ExecuteRequests() override;
void UpdateStatus(Status& status) const override;
void UpdateCompletionEstimates(AZStd::chrono::system_clock::time_point now, AZStd::vector<FileRequest*>& internalPending,
StreamerContext::PreparedQueue::iterator pendingBegin, StreamerContext::PreparedQueue::iterator pendingEnd) override;
void CollectStatistics(AZStd::vector<Statistic>& statistics) const override;
AZStd::chrono::high_resolution_clock::time_point m_queueStartTime;
AZStd::chrono::high_resolution_clock::time_point m_jobStartTime;
Buffer m_compressedData{ nullptr };
FileRequest* m_waitRequest{ nullptr };
u32 m_alignmentOffset{ 0 };
};
private:
using Buffer = u8*;
bool IsIdle() const;
enum class ReadBufferStatus : uint8_t
{
Unused,
ReadInFlight,
PendingDecompression
};
void PrepareReadRequest(FileRequest* request, FileRequest::ReadRequestData& data);
void PrepareDedicatedCache(FileRequest* request, const RequestPath& path);
void FileExistsCheck(FileRequest* checkRequest);
struct DecompressionInformation
{
bool IsProcessing() const;
void EstimateCompressedReadRequest(FileRequest* request, AZStd::chrono::microseconds& cumulativeDelay,
AZStd::chrono::microseconds decompressionDelay, double totalDecompressionDurationUs, double totalBytesDecompressed) const;
AZStd::chrono::high_resolution_clock::time_point m_queueStartTime;
AZStd::chrono::high_resolution_clock::time_point m_jobStartTime;
Buffer m_compressedData{ nullptr };
FileRequest* m_waitRequest{ nullptr };
u32 m_alignmentOffset{ 0 };
};
void StartArchiveRead(FileRequest* compressedReadRequest);
void FinishArchiveRead(FileRequest* readRequest, u32 readSlot);
bool StartDecompressions();
void FinishDecompression(FileRequest* waitRequest, u32 jobSlot);
bool IsIdle() const;
static void FullDecompression(StreamerContext* context, DecompressionInformation& info);
static void PartialDecompression(StreamerContext* context, DecompressionInformation& info);
void PrepareReadRequest(FileRequest* request, FileRequest::ReadRequestData& data);
void PrepareDedicatedCache(FileRequest* request, const RequestPath& path);
void FileExistsCheck(FileRequest* checkRequest);
AZStd::deque<FileRequest*> m_pendingReads;
AZStd::deque<FileRequest*> m_pendingFileExistChecks;
void EstimateCompressedReadRequest(FileRequest* request, AZStd::chrono::microseconds& cumulativeDelay,
AZStd::chrono::microseconds decompressionDelay, double totalDecompressionDurationUs, double totalBytesDecompressed) const;
void StartArchiveRead(FileRequest* compressedReadRequest);
void FinishArchiveRead(FileRequest* readRequest, u32 readSlot);
bool StartDecompressions();
void FinishDecompression(FileRequest* waitRequest, u32 jobSlot);
static void FullDecompression(StreamerContext* context, DecompressionInformation& info);
static void PartialDecompression(StreamerContext* context, DecompressionInformation& info);
AZStd::deque<FileRequest*> m_pendingReads;
AZStd::deque<FileRequest*> m_pendingFileExistChecks;
AverageWindow<size_t, double, s_statisticsWindowSize> m_decompressionJobDelayMicroSec;
AverageWindow<size_t, double, s_statisticsWindowSize> m_decompressionDurationMicroSec;
AverageWindow<size_t, double, s_statisticsWindowSize> m_bytesDecompressed;
AverageWindow<size_t, double, s_statisticsWindowSize> m_decompressionJobDelayMicroSec;
AverageWindow<size_t, double, s_statisticsWindowSize> m_decompressionDurationMicroSec;
AverageWindow<size_t, double, s_statisticsWindowSize> m_bytesDecompressed;
#if AZ_STREAMER_ADD_EXTRA_PROFILING_INFO
AZ::Statistics::RunningStatistic m_decompressionBoundStat;
AZ::Statistics::RunningStatistic m_readBoundStat;
AZ::Statistics::RunningStatistic m_decompressionBoundStat;
AZ::Statistics::RunningStatistic m_readBoundStat;
#endif
AZStd::unique_ptr<Buffer[]> m_readBuffers;
// Nullptr if not reading, the read request if reading the file and the wait request for decompression when waiting on decompression.
AZStd::unique_ptr<FileRequest*[]> m_readRequests;
AZStd::unique_ptr<ReadBufferStatus[]> m_readBufferStatus;
AZStd::unique_ptr<DecompressionInformation[]> m_processingJobs;
AZStd::unique_ptr<JobManager> m_decompressionJobManager;
AZStd::unique_ptr<JobContext> m_decompressionjobContext;
AZStd::unique_ptr<Buffer[]> m_readBuffers;
// Nullptr if not reading, the read request if reading the file and the wait request for decompression when waiting on decompression.
AZStd::unique_ptr<FileRequest*[]> m_readRequests;
AZStd::unique_ptr<ReadBufferStatus[]> m_readBufferStatus;
size_t m_memoryUsage{ 0 }; //!< Amount of memory used for buffers by the decompressor.
u32 m_maxNumReads{ 2 };
u32 m_numInFlightReads{ 0 };
u32 m_numPendingDecompression{ 0 };
u32 m_maxNumJobs{ 1 };
u32 m_numRunningJobs{ 0 };
u32 m_alignment{ 0 };
};
} // namespace IO
} // namespace AZ
AZStd::unique_ptr<DecompressionInformation[]> m_processingJobs;
AZStd::unique_ptr<JobManager> m_decompressionJobManager;
AZStd::unique_ptr<JobContext> m_decompressionjobContext;
size_t m_memoryUsage{ 0 }; //!< Amount of memory used for buffers by the decompressor.
u32 m_maxNumReads{ 2 };
u32 m_numInFlightReads{ 0 };
u32 m_numPendingDecompression{ 0 };
u32 m_maxNumJobs{ 1 };
u32 m_numRunningJobs{ 0 };
u32 m_alignment{ 0 };
};
} // namespace AZ::IO
Code/Framework/AzCore/AzCore/IO/Streamer/Scheduler.cpp
+3 -3
View File
@@ -227,7 +227,7 @@ namespace AZ::IO
{
auto parentReadRequest = next->GetCommandFromChain<FileRequest::ReadRequestData>();
AZ_Assert(parentReadRequest != nullptr, "The issued read request can't be found for the (compressed) read command.");
size_t size = parentReadRequest->m_size;
if (parentReadRequest->m_output == nullptr)
{
@@ -266,7 +266,7 @@ namespace AZ::IO
m_processingStartTime = AZStd::chrono::system_clock::now();
}
#endif
if constexpr (AZStd::is_same_v<Command, FileRequest::ReadData>)
{
m_threadData.m_lastFilePath = args.m_path;
@@ -411,7 +411,7 @@ namespace AZ::IO
++pendingIt;
}
}
m_threadData.m_streamStack->QueueRequest(request);
}
Code/Framework/AzCore/AzCore/IO/Streamer/Scheduler.h
+2 -2
View File
@@ -23,7 +23,7 @@
namespace AZ::IO
{
class FileRequest;
class Scheduler final
{
public:
@@ -63,7 +63,7 @@ namespace AZ::IO
void Thread_ProcessTillIdle();
void Thread_ProcessCancelRequest(FileRequest* request, FileRequest::CancelData& data);
void Thread_ProcessRescheduleRequest(FileRequest* request, FileRequest::RescheduleData& data);
enum class Order
{
FirstRequest, //< The first request is the most important to process next.
Code/Framework/AzCore/AzCore/IO/Streamer/StorageDrive.cpp
+394 -397
View File
@@ -16,454 +16,451 @@
#include <AzCore/std/smart_ptr/make_shared.h>
#include <AzCore/std/typetraits/decay.h>
namespace AZ
namespace AZ::IO
{
namespace IO
AZStd::shared_ptr<StreamStackEntry> StorageDriveConfig::AddStreamStackEntry(
[[maybe_unused]] const HardwareInformation& hardware, [[maybe_unused]] AZStd::shared_ptr<StreamStackEntry> parent)
{
AZStd::shared_ptr<StreamStackEntry> StorageDriveConfig::AddStreamStackEntry(
[[maybe_unused]] const HardwareInformation& hardware, [[maybe_unused]] AZStd::shared_ptr<StreamStackEntry> parent)
{
return AZStd::make_shared<StorageDrive>(m_maxFileHandles);
}
return AZStd::make_shared<StorageDrive>(m_maxFileHandles);
}
void StorageDriveConfig::Reflect(AZ::ReflectContext* context)
void StorageDriveConfig::Reflect(AZ::ReflectContext* context)
{
if (auto serializeContext = azrtti_cast<AZ::SerializeContext*>(context); serializeContext != nullptr)
{
if (auto serializeContext = azrtti_cast<AZ::SerializeContext*>(context); serializeContext != nullptr)
serializeContext->Class<StorageDriveConfig, IStreamerStackConfig>()
->Version(1)
->Field("MaxFileHandles", &StorageDriveConfig::m_maxFileHandles);
}
}
const AZStd::chrono::microseconds StorageDrive::s_averageSeekTime =
AZStd::chrono::milliseconds(9) + // Common average seek time for desktop hdd drives.
AZStd::chrono::milliseconds(3); // Rotational latency for a 7200RPM disk
StorageDrive::StorageDrive(u32 maxFileHandles)
: StreamStackEntry("Storage drive (generic)")
{
m_fileLastUsed.resize(maxFileHandles, AZStd::chrono::system_clock::time_point::min());
m_filePaths.resize(maxFileHandles);
m_fileHandles.resize(maxFileHandles);
// Add initial dummy values to the stats to avoid division by zero later on and avoid needing branches.
m_readSizeAverage.PushEntry(1);
m_readTimeAverage.PushEntry(AZStd::chrono::microseconds(1));
}
void StorageDrive::SetNext(AZStd::shared_ptr<StreamStackEntry> /*next*/)
{
AZ_Assert(false, "StorageDrive isn't allowed to have a node to forward requests to.");
}
void StorageDrive::PrepareRequest(FileRequest* request)
{
AZ_PROFILE_FUNCTION(AzCore);
AZ_Assert(request, "PrepareRequest was provided a null request.");
if (AZStd::holds_alternative<FileRequest::ReadRequestData>(request->GetCommand()))
{
auto& readRequest = AZStd::get<FileRequest::ReadRequestData>(request->GetCommand());
FileRequest* read = m_context->GetNewInternalRequest();
read->CreateRead(request, readRequest.m_output, readRequest.m_outputSize, readRequest.m_path,
readRequest.m_offset, readRequest.m_size);
m_context->PushPreparedRequest(read);
return;
}
StreamStackEntry::PrepareRequest(request);
}
void StorageDrive::QueueRequest(FileRequest* request)
{
AZ_Assert(request, "QueueRequest was provided a null request.");
AZStd::visit([this, request](auto&& args)
{
using Command = AZStd::decay_t<decltype(args)>;
if constexpr (AZStd::is_same_v<Command, FileRequest::ReadData> ||
AZStd::is_same_v<Command, FileRequest::FileExistsCheckData> ||
AZStd::is_same_v<Command, FileRequest::FileMetaDataRetrievalData>)
{
serializeContext->Class<StorageDriveConfig, IStreamerStackConfig>()
->Version(1)
->Field("MaxFileHandles", &StorageDriveConfig::m_maxFileHandles);
}
}
const AZStd::chrono::microseconds StorageDrive::s_averageSeekTime =
AZStd::chrono::milliseconds(9) + // Common average seek time for desktop hdd drives.
AZStd::chrono::milliseconds(3); // Rotational latency for a 7200RPM disk
StorageDrive::StorageDrive(u32 maxFileHandles)
: StreamStackEntry("Storage drive (generic)")
{
m_fileLastUsed.resize(maxFileHandles, AZStd::chrono::system_clock::time_point::min());
m_filePaths.resize(maxFileHandles);
m_fileHandles.resize(maxFileHandles);
// Add initial dummy values to the stats to avoid division by zero later on and avoid needing branches.
m_readSizeAverage.PushEntry(1);
m_readTimeAverage.PushEntry(AZStd::chrono::microseconds(1));
}
void StorageDrive::SetNext(AZStd::shared_ptr<StreamStackEntry> /*next*/)
{
AZ_Assert(false, "StorageDrive isn't allowed to have a node to forward requests to.");
}
void StorageDrive::PrepareRequest(FileRequest* request)
{
AZ_PROFILE_FUNCTION(AzCore);
AZ_Assert(request, "PrepareRequest was provided a null request.");
if (AZStd::holds_alternative<FileRequest::ReadRequestData>(request->GetCommand()))
{
auto& readRequest = AZStd::get<FileRequest::ReadRequestData>(request->GetCommand());
FileRequest* read = m_context->GetNewInternalRequest();
read->CreateRead(request, readRequest.m_output, readRequest.m_outputSize, readRequest.m_path,
readRequest.m_offset, readRequest.m_size);
m_context->PushPreparedRequest(read);
m_pendingRequests.push_back(request);
return;
}
StreamStackEntry::PrepareRequest(request);
}
else if constexpr (AZStd::is_same_v<Command, FileRequest::CancelData>)
{
CancelRequest(request, args.m_target);
return;
}
else
{
if constexpr (AZStd::is_same_v<Command, FileRequest::FlushData>)
{
FlushCache(args.m_path);
}
else if constexpr (AZStd::is_same_v<Command, FileRequest::FlushAllData>)
{
FlushEntireCache();
}
else if constexpr (AZStd::is_same_v<Command, FileRequest::ReportData>)
{
Report(args);
}
StreamStackEntry::QueueRequest(request);
}
}, request->GetCommand());
}
void StorageDrive::QueueRequest(FileRequest* request)
bool StorageDrive::ExecuteRequests()
{
if (!m_pendingRequests.empty())
{
AZ_Assert(request, "QueueRequest was provided a null request.");
FileRequest* request = m_pendingRequests.front();
AZStd::visit([this, request](auto&& args)
{
using Command = AZStd::decay_t<decltype(args)>;
if constexpr (AZStd::is_same_v<Command, FileRequest::ReadData> ||
AZStd::is_same_v<Command, FileRequest::FileExistsCheckData> ||
AZStd::is_same_v<Command, FileRequest::FileMetaDataRetrievalData>)
{
m_pendingRequests.push_back(request);
return;
}
else if constexpr (AZStd::is_same_v<Command, FileRequest::CancelData>)
{
CancelRequest(request, args.m_target);
return;
}
else
{
if constexpr (AZStd::is_same_v<Command, FileRequest::FlushData>)
{
FlushCache(args.m_path);
}
else if constexpr (AZStd::is_same_v<Command, FileRequest::FlushAllData>)
{
FlushEntireCache();
}
else if constexpr (AZStd::is_same_v<Command, FileRequest::ReportData>)
{
Report(args);
}
StreamStackEntry::QueueRequest(request);
}
}, request->GetCommand());
}
bool StorageDrive::ExecuteRequests()
{
if (!m_pendingRequests.empty())
{
FileRequest* request = m_pendingRequests.front();
AZStd::visit([this, request](auto&& args)
{
using Command = AZStd::decay_t<decltype(args)>;
if constexpr (AZStd::is_same_v<Command, FileRequest::ReadData>)
{
ReadFile(request);
}
else if constexpr (AZStd::is_same_v<Command, FileRequest::FileExistsCheckData>)
{
FileExistsRequest(request);
}
else if constexpr (AZStd::is_same_v<Command, FileRequest::FileMetaDataRetrievalData>)
{
FileMetaDataRetrievalRequest(request);
}
}, request->GetCommand());
m_pendingRequests.pop_front();
return true;
}
else
{
return false;
}
}
void StorageDrive::UpdateStatus(Status& status) const
{
// Only participate if there are actually any reads done.
if (m_fileOpenCloseTimeAverage.GetNumRecorded() > 0)
{
s32 availableSlots = s_maxRequests - aznumeric_cast<s32>(m_pendingRequests.size());
StreamStackEntry::UpdateStatus(status);
status.m_numAvailableSlots = AZStd::min(status.m_numAvailableSlots, availableSlots);
status.m_isIdle = status.m_isIdle && m_pendingRequests.empty();
}
else
{
status.m_numAvailableSlots = AZStd::min(status.m_numAvailableSlots, s_maxRequests);
}
}
void StorageDrive::UpdateCompletionEstimates(AZStd::chrono::system_clock::time_point now,
AZStd::vector<FileRequest*>& internalPending, StreamerContext::PreparedQueue::iterator pendingBegin,
StreamerContext::PreparedQueue::iterator pendingEnd)
{
StreamStackEntry::UpdateCompletionEstimates(now, internalPending, pendingBegin, pendingEnd);
const RequestPath* activeFile = nullptr;
if (m_activeCacheSlot != s_fileNotFound)
{
activeFile = &m_filePaths[m_activeCacheSlot];
}
u64 activeOffset = m_activeOffset;
// Estimate requests in this stack entry.
for (FileRequest* request : m_pendingRequests)
{
EstimateCompletionTimeForRequest(request, now, activeFile, activeOffset);
}
// Estimate internally pending requests. Because this call will go from the top of the stack to the bottom,
// but estimation is calculated from the bottom to the top, this list should be processed in reverse order.
for (auto requestIt = internalPending.rbegin(); requestIt != internalPending.rend(); ++requestIt)
{
EstimateCompletionTimeForRequest(*requestIt, now, activeFile, activeOffset);
}
// Estimate pending requests that have not been queued yet.
for (auto requestIt = pendingBegin; requestIt != pendingEnd; ++requestIt)
{
EstimateCompletionTimeForRequest(*requestIt, now, activeFile, activeOffset);
}
}
void StorageDrive::EstimateCompletionTimeForRequest(FileRequest* request, AZStd::chrono::system_clock::time_point& startTime,
const RequestPath*& activeFile, u64& activeOffset) const
{
u64 readSize = 0;
u64 offset = 0;
const RequestPath* targetFile = nullptr;
AZStd::visit([&](auto&& args)
{
using Command = AZStd::decay_t<decltype(args)>;
if constexpr (AZStd::is_same_v<Command, FileRequest::ReadData>)
{
targetFile = &args.m_path;
readSize = args.m_size;
offset = args.m_offset;
}
else if constexpr (AZStd::is_same_v<Command, FileRequest::CompressedReadData>)
{
targetFile = &args.m_compressionInfo.m_archiveFilename;
readSize = args.m_compressionInfo.m_compressedSize;
offset = args.m_compressionInfo.m_offset;
ReadFile(request);
}
else if constexpr (AZStd::is_same_v<Command, FileRequest::FileExistsCheckData>)
{
readSize = 0;
AZStd::chrono::microseconds averageTime = m_getFileExistsTimeAverage.CalculateAverage();
startTime += averageTime;
FileExistsRequest(request);
}
else if constexpr (AZStd::is_same_v<Command, FileRequest::FileMetaDataRetrievalData>)
{
readSize = 0;
AZStd::chrono::microseconds averageTime = m_getFileMetaDataTimeAverage.CalculateAverage();
startTime += averageTime;
FileMetaDataRetrievalRequest(request);
}
}, request->GetCommand());
m_pendingRequests.pop_front();
return true;
}
else
{
return false;
}
}
if (readSize > 0)
{
if (activeFile && activeFile != targetFile)
{
if (FindFileInCache(*targetFile) == s_fileNotFound)
{
AZStd::chrono::microseconds fileOpenCloseTimeAverage = m_fileOpenCloseTimeAverage.CalculateAverage();
startTime += fileOpenCloseTimeAverage;
}
startTime += s_averageSeekTime;
activeOffset = std::numeric_limits<u64>::max();
}
else if (activeOffset != offset)
{
startTime += s_averageSeekTime;
}
void StorageDrive::UpdateStatus(Status& status) const
{
// Only participate if there are actually any reads done.
if (m_fileOpenCloseTimeAverage.GetNumRecorded() > 0)
{
s32 availableSlots = s_maxRequests - aznumeric_cast<s32>(m_pendingRequests.size());
StreamStackEntry::UpdateStatus(status);
status.m_numAvailableSlots = AZStd::min(status.m_numAvailableSlots, availableSlots);
status.m_isIdle = status.m_isIdle && m_pendingRequests.empty();
}
else
{
status.m_numAvailableSlots = AZStd::min(status.m_numAvailableSlots, s_maxRequests);
}
}
u64 totalBytesRead = m_readSizeAverage.GetTotal();
double totalReadTimeUSec = aznumeric_caster(m_readTimeAverage.GetTotal().count());
startTime += AZStd::chrono::microseconds(aznumeric_cast<u64>((readSize * totalReadTimeUSec) / totalBytesRead));
activeOffset = offset + readSize;
}
request->SetEstimatedCompletion(startTime);
void StorageDrive::UpdateCompletionEstimates(AZStd::chrono::system_clock::time_point now,
AZStd::vector<FileRequest*>& internalPending, StreamerContext::PreparedQueue::iterator pendingBegin,
StreamerContext::PreparedQueue::iterator pendingEnd)
{
StreamStackEntry::UpdateCompletionEstimates(now, internalPending, pendingBegin, pendingEnd);
const RequestPath* activeFile = nullptr;
if (m_activeCacheSlot != s_fileNotFound)
{
activeFile = &m_filePaths[m_activeCacheSlot];
}
u64 activeOffset = m_activeOffset;
// Estimate requests in this stack entry.
for (FileRequest* request : m_pendingRequests)
{
EstimateCompletionTimeForRequest(request, now, activeFile, activeOffset);
}
void StorageDrive::ReadFile(FileRequest* request)
// Estimate internally pending requests. Because this call will go from the top of the stack to the bottom,
// but estimation is calculated from the bottom to the top, this list should be processed in reverse order.
for (auto requestIt = internalPending.rbegin(); requestIt != internalPending.rend(); ++requestIt)
{
AZ_PROFILE_FUNCTION(AzCore);
auto data = AZStd::get_if<FileRequest::ReadData>(&request->GetCommand());
AZ_Assert(data, "FileRequest queued on StorageDrive to be read didn't contain read data.");
SystemFile* file = nullptr;
// If the file is already open, use that file handle and update it's last touched time.
size_t cacheIndex = FindFileInCache(data->m_path);
if (cacheIndex != s_fileNotFound)
{
file = m_fileHandles[cacheIndex].get();
m_fileLastUsed[cacheIndex] = AZStd::chrono::high_resolution_clock::now();
}
// If the file is not open, eject the entry from the cache that hasn't been used for the longest time
// and open the file for reading.
if (!file)
{
AZStd::chrono::system_clock::time_point oldest = m_fileLastUsed[0];
cacheIndex = 0;
size_t numFiles = m_filePaths.size();
for (size_t i = 1; i < numFiles; ++i)
{
if (m_fileLastUsed[i] < oldest)
{
oldest = m_fileLastUsed[i];
cacheIndex = i;
}
}
TIMED_AVERAGE_WINDOW_SCOPE(m_fileOpenCloseTimeAverage);
AZStd::unique_ptr<SystemFile> newFile = AZStd::make_unique<SystemFile>();
bool isOpen = newFile->Open(data->m_path.GetAbsolutePath(), SystemFile::OpenMode::SF_OPEN_READ_ONLY);
if (!isOpen)
{
request->SetStatus(IStreamerTypes::RequestStatus::Failed);
m_context->MarkRequestAsCompleted(request);
return;
}
file = newFile.get();
m_fileLastUsed[cacheIndex] = AZStd::chrono::high_resolution_clock::now();
m_fileHandles[cacheIndex] = AZStd::move(newFile);
m_filePaths[cacheIndex] = data->m_path;
}
AZ_Assert(file, "While searching for file '%s' StorageDevice::ReadFile failed to detect a problem.", data->m_path.GetRelativePath());
u64 bytesRead = 0;
{
TIMED_AVERAGE_WINDOW_SCOPE(m_readTimeAverage);
if (file->Tell() != data->m_offset)
{
file->Seek(data->m_offset, SystemFile::SeekMode::SF_SEEK_BEGIN);
}
bytesRead = file->Read(data->m_size, data->m_output);
}
m_readSizeAverage.PushEntry(bytesRead);
m_activeCacheSlot = cacheIndex;
m_activeOffset = data->m_offset + bytesRead;
request->SetStatus(bytesRead == data->m_size ? IStreamerTypes::RequestStatus::Completed : IStreamerTypes::RequestStatus::Failed);
m_context->MarkRequestAsCompleted(request);
EstimateCompletionTimeForRequest(*requestIt, now, activeFile, activeOffset);
}
void StorageDrive::CancelRequest(FileRequest* cancelRequest, FileRequestPtr& target)
// Estimate pending requests that have not been queued yet.
for (auto requestIt = pendingBegin; requestIt != pendingEnd; ++requestIt)
{
for (auto it = m_pendingRequests.begin(); it != m_pendingRequests.end();)
EstimateCompletionTimeForRequest(*requestIt, now, activeFile, activeOffset);
}
}
void StorageDrive::EstimateCompletionTimeForRequest(FileRequest* request, AZStd::chrono::system_clock::time_point& startTime,
const RequestPath*& activeFile, u64& activeOffset) const
{
u64 readSize = 0;
u64 offset = 0;
const RequestPath* targetFile = nullptr;
AZStd::visit([&](auto&& args)
{
using Command = AZStd::decay_t<decltype(args)>;
if constexpr (AZStd::is_same_v<Command, FileRequest::ReadData>)
{
if ((*it)->WorksOn(target))
{
(*it)->SetStatus(IStreamerTypes::RequestStatus::Canceled);
m_context->MarkRequestAsCompleted(*it);
it = m_pendingRequests.erase(it);
}
else
{
++it;
}
targetFile = &args.m_path;
readSize = args.m_size;
offset = args.m_offset;
}
cancelRequest->SetStatus(IStreamerTypes::RequestStatus::Completed);
m_context->MarkRequestAsCompleted(cancelRequest);
else if constexpr (AZStd::is_same_v<Command, FileRequest::CompressedReadData>)
{
targetFile = &args.m_compressionInfo.m_archiveFilename;
readSize = args.m_compressionInfo.m_compressedSize;
offset = args.m_compressionInfo.m_offset;
}
else if constexpr (AZStd::is_same_v<Command, FileRequest::FileExistsCheckData>)
{
readSize = 0;
AZStd::chrono::microseconds averageTime = m_getFileExistsTimeAverage.CalculateAverage();
startTime += averageTime;
}
else if constexpr (AZStd::is_same_v<Command, FileRequest::FileMetaDataRetrievalData>)
{
readSize = 0;
AZStd::chrono::microseconds averageTime = m_getFileMetaDataTimeAverage.CalculateAverage();
startTime += averageTime;
}
}, request->GetCommand());
if (readSize > 0)
{
if (activeFile && activeFile != targetFile)
{
if (FindFileInCache(*targetFile) == s_fileNotFound)
{
AZStd::chrono::microseconds fileOpenCloseTimeAverage = m_fileOpenCloseTimeAverage.CalculateAverage();
startTime += fileOpenCloseTimeAverage;
}
startTime += s_averageSeekTime;
activeOffset = std::numeric_limits<u64>::max();
}
else if (activeOffset != offset)
{
startTime += s_averageSeekTime;
}
u64 totalBytesRead = m_readSizeAverage.GetTotal();
double totalReadTimeUSec = aznumeric_caster(m_readTimeAverage.GetTotal().count());
startTime += AZStd::chrono::microseconds(aznumeric_cast<u64>((readSize * totalReadTimeUSec) / totalBytesRead));
activeOffset = offset + readSize;
}
request->SetEstimatedCompletion(startTime);
}
void StorageDrive::ReadFile(FileRequest* request)
{
AZ_PROFILE_FUNCTION(AzCore);
auto data = AZStd::get_if<FileRequest::ReadData>(&request->GetCommand());
AZ_Assert(data, "FileRequest queued on StorageDrive to be read didn't contain read data.");
SystemFile* file = nullptr;
// If the file is already open, use that file handle and update it's last touched time.
size_t cacheIndex = FindFileInCache(data->m_path);
if (cacheIndex != s_fileNotFound)
{
file = m_fileHandles[cacheIndex].get();
m_fileLastUsed[cacheIndex] = AZStd::chrono::high_resolution_clock::now();
}
void StorageDrive::FileExistsRequest(FileRequest* request)
// If the file is not open, eject the entry from the cache that hasn't been used for the longest time
// and open the file for reading.
if (!file)
{
AZ_PROFILE_FUNCTION(AzCore);
TIMED_AVERAGE_WINDOW_SCOPE(m_getFileExistsTimeAverage);
auto& fileExists = AZStd::get<FileRequest::FileExistsCheckData>(request->GetCommand());
size_t cacheIndex = FindFileInCache(fileExists.m_path);
if (cacheIndex != s_fileNotFound)
AZStd::chrono::system_clock::time_point oldest = m_fileLastUsed[0];
cacheIndex = 0;
size_t numFiles = m_filePaths.size();
for (size_t i = 1; i < numFiles; ++i)
{
fileExists.m_found = true;
if (m_fileLastUsed[i] < oldest)
{
oldest = m_fileLastUsed[i];
cacheIndex = i;
}
}
TIMED_AVERAGE_WINDOW_SCOPE(m_fileOpenCloseTimeAverage);
AZStd::unique_ptr<SystemFile> newFile = AZStd::make_unique<SystemFile>();
bool isOpen = newFile->Open(data->m_path.GetAbsolutePath(), SystemFile::OpenMode::SF_OPEN_READ_ONLY);
if (!isOpen)
{
request->SetStatus(IStreamerTypes::RequestStatus::Failed);
m_context->MarkRequestAsCompleted(request);
return;
}
file = newFile.get();
m_fileLastUsed[cacheIndex] = AZStd::chrono::high_resolution_clock::now();
m_fileHandles[cacheIndex] = AZStd::move(newFile);
m_filePaths[cacheIndex] = data->m_path;
}
AZ_Assert(file, "While searching for file '%s' StorageDevice::ReadFile failed to detect a problem.", data->m_path.GetRelativePath());
u64 bytesRead = 0;
{
TIMED_AVERAGE_WINDOW_SCOPE(m_readTimeAverage);
if (file->Tell() != data->m_offset)
{
file->Seek(data->m_offset, SystemFile::SeekMode::SF_SEEK_BEGIN);
}
bytesRead = file->Read(data->m_size, data->m_output);
}
m_readSizeAverage.PushEntry(bytesRead);
m_activeCacheSlot = cacheIndex;
m_activeOffset = data->m_offset + bytesRead;
request->SetStatus(bytesRead == data->m_size ? IStreamerTypes::RequestStatus::Completed : IStreamerTypes::RequestStatus::Failed);
m_context->MarkRequestAsCompleted(request);
}
void StorageDrive::CancelRequest(FileRequest* cancelRequest, FileRequestPtr& target)
{
for (auto it = m_pendingRequests.begin(); it != m_pendingRequests.end();)
{
if ((*it)->WorksOn(target))
{
(*it)->SetStatus(IStreamerTypes::RequestStatus::Canceled);
m_context->MarkRequestAsCompleted(*it);
it = m_pendingRequests.erase(it);
}
else
{
fileExists.m_found = SystemFile::Exists(fileExists.m_path.GetAbsolutePath());
++it;
}
m_context->MarkRequestAsCompleted(request);
}
cancelRequest->SetStatus(IStreamerTypes::RequestStatus::Completed);
m_context->MarkRequestAsCompleted(cancelRequest);
}
void StorageDrive::FileMetaDataRetrievalRequest(FileRequest* request)
void StorageDrive::FileExistsRequest(FileRequest* request)
{
AZ_PROFILE_FUNCTION(AzCore);
TIMED_AVERAGE_WINDOW_SCOPE(m_getFileExistsTimeAverage);
auto& fileExists = AZStd::get<FileRequest::FileExistsCheckData>(request->GetCommand());
size_t cacheIndex = FindFileInCache(fileExists.m_path);
if (cacheIndex != s_fileNotFound)
{
AZ_PROFILE_FUNCTION(AzCore);
TIMED_AVERAGE_WINDOW_SCOPE(m_getFileMetaDataTimeAverage);
fileExists.m_found = true;
}
else
{
fileExists.m_found = SystemFile::Exists(fileExists.m_path.GetAbsolutePath());
}
m_context->MarkRequestAsCompleted(request);
}
auto& command = AZStd::get<FileRequest::FileMetaDataRetrievalData>(request->GetCommand());
// If the file is already open, use the file handle which usually is cheaper than asking for the file by name.
size_t cacheIndex = FindFileInCache(command.m_path);
if (cacheIndex != s_fileNotFound)
void StorageDrive::FileMetaDataRetrievalRequest(FileRequest* request)
{
AZ_PROFILE_FUNCTION(AzCore);
TIMED_AVERAGE_WINDOW_SCOPE(m_getFileMetaDataTimeAverage);
auto& command = AZStd::get<FileRequest::FileMetaDataRetrievalData>(request->GetCommand());
// If the file is already open, use the file handle which usually is cheaper than asking for the file by name.
size_t cacheIndex = FindFileInCache(command.m_path);
if (cacheIndex != s_fileNotFound)
{
AZ_Assert(m_fileHandles[cacheIndex],
"File path '%s' doesn't have an associated file handle.", m_filePaths[cacheIndex].GetRelativePath());
command.m_fileSize = m_fileHandles[cacheIndex]->Length();
command.m_found = true;
request->SetStatus(IStreamerTypes::RequestStatus::Completed);
}
else
{
// The file is not open yet, so try to get the file size by name.
u64 size = SystemFile::Length(command.m_path.GetAbsolutePath());
if (size != 0) // SystemFile::Length doesn't allow telling a zero-sized file apart from a invalid path.
{
AZ_Assert(m_fileHandles[cacheIndex],
"File path '%s' doesn't have an associated file handle.", m_filePaths[cacheIndex].GetRelativePath());
command.m_fileSize = m_fileHandles[cacheIndex]->Length();
command.m_fileSize = size;
command.m_found = true;
request->SetStatus(IStreamerTypes::RequestStatus::Completed);
}
else
{
// The file is not open yet, so try to get the file size by name.
u64 size = SystemFile::Length(command.m_path.GetAbsolutePath());
if (size != 0) // SystemFile::Length doesn't allow telling a zero-sized file apart from a invalid path.
request->SetStatus(IStreamerTypes::RequestStatus::Failed);
}
}
m_context->MarkRequestAsCompleted(request);
}
void StorageDrive::FlushCache(const RequestPath& filePath)
{
size_t cacheIndex = FindFileInCache(filePath);
if (cacheIndex != s_fileNotFound)
{
m_fileLastUsed[cacheIndex] = AZStd::chrono::system_clock::time_point();
m_fileHandles[cacheIndex].reset();
m_filePaths[cacheIndex].Clear();
}
}
void StorageDrive::FlushEntireCache()
{
size_t numFiles = m_filePaths.size();
for (size_t i = 0; i < numFiles; ++i)
{
m_fileLastUsed[i] = AZStd::chrono::system_clock::time_point();
m_fileHandles[i].reset();
m_filePaths[i].Clear();
}
}
size_t StorageDrive::FindFileInCache(const RequestPath& filePath) const
{
size_t numFiles = m_filePaths.size();
for (size_t i = 0; i < numFiles; ++i)
{
if (m_filePaths[i] == filePath)
{
return i;
}
}
return s_fileNotFound;
}
void StorageDrive::CollectStatistics(AZStd::vector<Statistic>& statistics) const
{
constexpr double bytesToMB = (1024.0 * 1024.0);
using DoubleSeconds = AZStd::chrono::duration<double>;
double totalBytesReadMB = m_readSizeAverage.GetTotal() / bytesToMB;
double totalReadTimeSec = AZStd::chrono::duration_cast<DoubleSeconds>(m_readTimeAverage.GetTotal()).count();
if (m_readSizeAverage.GetTotal() > 1) // A default value is always added.
{
statistics.push_back(Statistic::CreateFloat(m_name, "Read Speed (avg. mbps)", totalBytesReadMB / totalReadTimeSec));
}
if (m_fileOpenCloseTimeAverage.GetNumRecorded() > 0)
{
statistics.push_back(Statistic::CreateInteger(m_name, "File Open & Close (avg. us)", m_fileOpenCloseTimeAverage.CalculateAverage().count()));
statistics.push_back(Statistic::CreateInteger(m_name, "Get file exists (avg. us)", m_getFileExistsTimeAverage.CalculateAverage().count()));
statistics.push_back(Statistic::CreateInteger(m_name, "Get file meta data (avg. us)", m_getFileMetaDataTimeAverage.CalculateAverage().count()));
statistics.push_back(Statistic::CreateInteger(m_name, "Available slots", s64{ s_maxRequests } - m_pendingRequests.size()));
}
}
void StorageDrive::Report(const FileRequest::ReportData& data) const
{
switch (data.m_reportType)
{
case FileRequest::ReportData::ReportType::FileLocks:
for (u32 i = 0; i < m_fileHandles.size(); ++i)
{
if (m_fileHandles[i] != nullptr)
{
command.m_fileSize = size;
command.m_found = true;
request->SetStatus(IStreamerTypes::RequestStatus::Completed);
}
else
{
request->SetStatus(IStreamerTypes::RequestStatus::Failed);
AZ_Printf("Streamer", "File lock in %s : '%s'.\n", m_name.c_str(), m_filePaths[i].GetRelativePath());
}
}
m_context->MarkRequestAsCompleted(request);
break;
default:
break;
}
void StorageDrive::FlushCache(const RequestPath& filePath)
{
size_t cacheIndex = FindFileInCache(filePath);
if (cacheIndex != s_fileNotFound)
{
m_fileLastUsed[cacheIndex] = AZStd::chrono::system_clock::time_point();
m_fileHandles[cacheIndex].reset();
m_filePaths[cacheIndex].Clear();
}
}
void StorageDrive::FlushEntireCache()
{
size_t numFiles = m_filePaths.size();
for (size_t i = 0; i < numFiles; ++i)
{
m_fileLastUsed[i] = AZStd::chrono::system_clock::time_point();
m_fileHandles[i].reset();
m_filePaths[i].Clear();
}
}
size_t StorageDrive::FindFileInCache(const RequestPath& filePath) const
{
size_t numFiles = m_filePaths.size();
for (size_t i = 0; i < numFiles; ++i)
{
if (m_filePaths[i] == filePath)
{
return i;
}
}
return s_fileNotFound;
}
void StorageDrive::CollectStatistics(AZStd::vector<Statistic>& statistics) const
{
constexpr double bytesToMB = (1024.0 * 1024.0);
using DoubleSeconds = AZStd::chrono::duration<double>;
double totalBytesReadMB = m_readSizeAverage.GetTotal() / bytesToMB;
double totalReadTimeSec = AZStd::chrono::duration_cast<DoubleSeconds>(m_readTimeAverage.GetTotal()).count();
if (m_readSizeAverage.GetTotal() > 1) // A default value is always added.
{
statistics.push_back(Statistic::CreateFloat(m_name, "Read Speed (avg. mbps)", totalBytesReadMB / totalReadTimeSec));
}
if (m_fileOpenCloseTimeAverage.GetNumRecorded() > 0)
{
statistics.push_back(Statistic::CreateInteger(m_name, "File Open & Close (avg. us)", m_fileOpenCloseTimeAverage.CalculateAverage().count()));
statistics.push_back(Statistic::CreateInteger(m_name, "Get file exists (avg. us)", m_getFileExistsTimeAverage.CalculateAverage().count()));
statistics.push_back(Statistic::CreateInteger(m_name, "Get file meta data (avg. us)", m_getFileMetaDataTimeAverage.CalculateAverage().count()));
statistics.push_back(Statistic::CreateInteger(m_name, "Available slots", s64{ s_maxRequests } - m_pendingRequests.size()));
}
}
void StorageDrive::Report(const FileRequest::ReportData& data) const
{
switch (data.m_reportType)
{
case FileRequest::ReportData::ReportType::FileLocks:
for (u32 i = 0; i < m_fileHandles.size(); ++i)
{
if (m_fileHandles[i] != nullptr)
{
AZ_Printf("Streamer", "File lock in %s : '%s'.\n", m_name.c_str(), m_filePaths[i].GetRelativePath());
}
}
break;
default:
break;
}
}
} // namespace IO
} // namespace AZ
}
} // namespace AZ::IO
Code/Framework/AzCore/AzCore/IO/Streamer/StorageDrive.h
+63 -66
View File
@@ -16,85 +16,82 @@
#include <AzCore/std/containers/vector.h>
#include <AzCore/std/chrono/clocks.h>
namespace AZ
namespace AZ::IO
{
namespace IO
struct StorageDriveConfig final :
public IStreamerStackConfig
{
struct StorageDriveConfig final :
public IStreamerStackConfig
{
AZ_RTTI(AZ::IO::StorageDriveConfig, "{3D568902-6C09-4E9E-A4DB-8B561481D298}", IStreamerStackConfig);
AZ_CLASS_ALLOCATOR(StorageDriveConfig, AZ::SystemAllocator, 0);
AZ_RTTI(AZ::IO::StorageDriveConfig, "{3D568902-6C09-4E9E-A4DB-8B561481D298}", IStreamerStackConfig);
AZ_CLASS_ALLOCATOR(StorageDriveConfig, AZ::SystemAllocator, 0);
~StorageDriveConfig() override = default;
AZStd::shared_ptr<StreamStackEntry> AddStreamStackEntry(
const HardwareInformation& hardware, AZStd::shared_ptr<StreamStackEntry> parent) override;
static void Reflect(AZ::ReflectContext* context);
~StorageDriveConfig() override = default;
AZStd::shared_ptr<StreamStackEntry> AddStreamStackEntry(
const HardwareInformation& hardware, AZStd::shared_ptr<StreamStackEntry> parent) override;
static void Reflect(AZ::ReflectContext* context);
u32 m_maxFileHandles{1024};
};
u32 m_maxFileHandles{1024};
};
//! Platform agnostic version of a storage drive, such as hdd, ssd, dvd, etc.
//! This stream stack entry is responsible for accessing a storage drive to
//! retrieve file information and data.
//! This entry is designed as a catch-all for any reads that weren't handled
//! by platform specific implementations or the virtual file system. It should
//! by the last entry in the stack as it will not forward calls to the next entry.
class StorageDrive
: public StreamStackEntry
{
public:
explicit StorageDrive(u32 maxFileHandles);
~StorageDrive() override = default;
//! Platform agnostic version of a storage drive, such as hdd, ssd, dvd, etc.
//! This stream stack entry is responsible for accessing a storage drive to
//! retrieve file information and data.
//! This entry is designed as a catch-all for any reads that weren't handled
//! by platform specific implementations or the virtual file system. It should
//! by the last entry in the stack as it will not forward calls to the next entry.
class StorageDrive
: public StreamStackEntry
{
public:
explicit StorageDrive(u32 maxFileHandles);
~StorageDrive() override = default;
void SetNext(AZStd::shared_ptr<StreamStackEntry> next) override;
void SetNext(AZStd::shared_ptr<StreamStackEntry> next) override;
void PrepareRequest(FileRequest* request) override;
void QueueRequest(FileRequest* request) override;
bool ExecuteRequests() override;
void PrepareRequest(FileRequest* request) override;
void QueueRequest(FileRequest* request) override;
bool ExecuteRequests() override;
void UpdateStatus(Status& status) const override;
void UpdateCompletionEstimates(AZStd::chrono::system_clock::time_point now, AZStd::vector<FileRequest*>& internalPending,
StreamerContext::PreparedQueue::iterator pendingBegin, StreamerContext::PreparedQueue::iterator pendingEnd) override;
void UpdateStatus(Status& status) const override;
void UpdateCompletionEstimates(AZStd::chrono::system_clock::time_point now, AZStd::vector<FileRequest*>& internalPending,
StreamerContext::PreparedQueue::iterator pendingBegin, StreamerContext::PreparedQueue::iterator pendingEnd) override;
void CollectStatistics(AZStd::vector<Statistic>& statistics) const override;
void CollectStatistics(AZStd::vector<Statistic>& statistics) const override;
protected:
static const AZStd::chrono::microseconds s_averageSeekTime;
static constexpr s32 s_maxRequests = 1;
protected:
static const AZStd::chrono::microseconds s_averageSeekTime;
static constexpr s32 s_maxRequests = 1;
size_t FindFileInCache(const RequestPath& filePath) const;
void ReadFile(FileRequest* request);
void CancelRequest(FileRequest* cancelRequest, FileRequestPtr& target);
void FileExistsRequest(FileRequest* request);
void FileMetaDataRetrievalRequest(FileRequest* request);
void FlushCache(const RequestPath& filePath);
void FlushEntireCache();
size_t FindFileInCache(const RequestPath& filePath) const;
void ReadFile(FileRequest* request);
void CancelRequest(FileRequest* cancelRequest, FileRequestPtr& target);
void FileExistsRequest(FileRequest* request);
void FileMetaDataRetrievalRequest(FileRequest* request);
void FlushCache(const RequestPath& filePath);
void FlushEntireCache();
void EstimateCompletionTimeForRequest(FileRequest* request, AZStd::chrono::system_clock::time_point& startTime,
const RequestPath*& activeFile, u64& activeOffset) const;
void EstimateCompletionTimeForRequest(FileRequest* request, AZStd::chrono::system_clock::time_point& startTime,
const RequestPath*& activeFile, u64& activeOffset) const;
void Report(const FileRequest::ReportData& data) const;
void Report(const FileRequest::ReportData& data) const;
TimedAverageWindow<s_statisticsWindowSize> m_fileOpenCloseTimeAverage;
TimedAverageWindow<s_statisticsWindowSize> m_getFileExistsTimeAverage;
TimedAverageWindow<s_statisticsWindowSize> m_getFileMetaDataTimeAverage;
TimedAverageWindow<s_statisticsWindowSize> m_readTimeAverage;
AverageWindow<u64, float, s_statisticsWindowSize> m_readSizeAverage;
//! File requests that are queued for processing.
AZStd::deque<FileRequest*> m_pendingRequests;
TimedAverageWindow<s_statisticsWindowSize> m_fileOpenCloseTimeAverage;
TimedAverageWindow<s_statisticsWindowSize> m_getFileExistsTimeAverage;
TimedAverageWindow<s_statisticsWindowSize> m_getFileMetaDataTimeAverage;
TimedAverageWindow<s_statisticsWindowSize> m_readTimeAverage;
AverageWindow<u64, float, s_statisticsWindowSize> m_readSizeAverage;
//! File requests that are queued for processing.
AZStd::deque<FileRequest*> m_pendingRequests;
//! The last time a file handle was used to access a file. The handle is stored in m_fileHandles.
AZStd::vector<AZStd::chrono::system_clock::time_point> m_fileLastUsed;
//! The file path to the file handle. The handle is stored in m_fileHandles.
AZStd::vector<RequestPath> m_filePaths;
//! A list of file handles that's being cached in case they're needed again in the future.
AZStd::vector<AZStd::unique_ptr<SystemFile>> m_fileHandles;
//! The last time a file handle was used to access a file. The handle is stored in m_fileHandles.
AZStd::vector<AZStd::chrono::system_clock::time_point> m_fileLastUsed;
//! The file path to the file handle. The handle is stored in m_fileHandles.
AZStd::vector<RequestPath> m_filePaths;
//! A list of file handles that's being cached in case they're needed again in the future.
AZStd::vector<AZStd::unique_ptr<SystemFile>> m_fileHandles;
//! The offset into the file that's cached by the active cache slot.
u64 m_activeOffset = 0;
//! The index into m_fileHandles for the file that's currently being read.
size_t m_activeCacheSlot = s_fileNotFound;
};
} // namespace IO
} // namespace AZ
//! The offset into the file that's cached by the active cache slot.
u64 m_activeOffset = 0;
//! The index into m_fileHandles for the file that's currently being read.
size_t m_activeCacheSlot = s_fileNotFound;
};
} // namespace AZ::IO
Code/Framework/AzCore/AzCore/IO/Streamer/Streamer.cpp
+2 -2
View File
@@ -219,7 +219,7 @@ namespace AZ::IO
{
AZ_Assert(HasRequestCompleted(request), "Claiming memory from a read request that's still in progress. "
"This can lead to crashing if data is still being streamed to the request's buffer.");
// The caller has claimed the buffer and is now responsible for clearing it.
// The caller has claimed the buffer and is now responsible for clearing it.
readRequest->m_allocator->UnlockAllocator();
readRequest->m_allocator = nullptr;
}
@@ -293,7 +293,7 @@ namespace AZ::IO
request->m_request.CreateReport(reportType);
return request;
}
Streamer::Streamer(const AZStd::thread_desc& threadDesc, AZStd::unique_ptr<Scheduler> streamStack)
: m_streamStack(AZStd::move(streamStack))
{
Code/Framework/AzCore/AzCore/IO/Streamer/StreamerContext.h
+90 -93
View File
@@ -17,116 +17,113 @@
#include <AzCore/std/containers/queue.h>
#include <AzCore/Statistics/RunningStatistic.h>
namespace AZ
namespace AZ::IO
{
namespace IO
class StreamerContext
{
class StreamerContext
{
public:
using PreparedQueue = AZStd::deque<FileRequest*>;
public:
using PreparedQueue = AZStd::deque<FileRequest*>;
~StreamerContext();
~StreamerContext();
//! Gets a new file request, either by creating a new instance or
//! picking one from the recycle bin. This version should only be used
//! by nodes on the streaming stack as it's not thread safe, but faster.
//! The scheduler will automatically recycle these requests.
FileRequest* GetNewInternalRequest();
//! Gets a new file request, either by creating a new instance or
//! picking one from the recycle bin. This version is for use by
//! any system outside the stream stack and is thread safe. Once the
//! reference count in the request hits zero it will automatically be recycled.
FileRequestPtr GetNewExternalRequest();
//! Gets a batch of new file requests, either by creating new instances or
//! picking from the recycle bin. This version is for use by
//! any system outside the stream stack and is thread safe. The owner
//! needs to manually recycle these requests once they're done. Requests
//! with a reference count of zero will automatically be recycled.
//! If multiple requests need to be create this is preferable as it only locks the
//! recycle bin once.
void GetNewExternalRequestBatch(AZStd::vector<FileRequestPtr>& requests, size_t count);
//! Gets a new file request, either by creating a new instance or
//! picking one from the recycle bin. This version should only be used
//! by nodes on the streaming stack as it's not thread safe, but faster.
//! The scheduler will automatically recycle these requests.
FileRequest* GetNewInternalRequest();
//! Gets a new file request, either by creating a new instance or
//! picking one from the recycle bin. This version is for use by
//! any system outside the stream stack and is thread safe. Once the
//! reference count in the request hits zero it will automatically be recycled.
FileRequestPtr GetNewExternalRequest();
//! Gets a batch of new file requests, either by creating new instances or
//! picking from the recycle bin. This version is for use by
//! any system outside the stream stack and is thread safe. The owner
//! needs to manually recycle these requests once they're done. Requests
//! with a reference count of zero will automatically be recycled.
//! If multiple requests need to be create this is preferable as it only locks the
//! recycle bin once.
void GetNewExternalRequestBatch(AZStd::vector<FileRequestPtr>& requests, size_t count);
//! Gets the number of prepared requests. Prepared requests are requests
//! that are ready to be queued up for further processing.
size_t GetNumPreparedRequests() const;
//! Gets the next prepared request that should be queued. Prepared requests
//! are requests that are ready to be queued up for further processing.
FileRequest* PopPreparedRequest();
//! Adds a prepared request for later queuing and processing.
void PushPreparedRequest(FileRequest* request);
//! Gets the prepared requests that are queued to be processed.
PreparedQueue& GetPreparedRequests();
//! Gets the prepared requests that are queued to be processed.
const PreparedQueue& GetPreparedRequests() const;
//! Gets the number of prepared requests. Prepared requests are requests
//! that are ready to be queued up for further processing.
size_t GetNumPreparedRequests() const;
//! Gets the next prepared request that should be queued. Prepared requests
//! are requests that are ready to be queued up for further processing.
FileRequest* PopPreparedRequest();
//! Adds a prepared request for later queuing and processing.
void PushPreparedRequest(FileRequest* request);
//! Gets the prepared requests that are queued to be processed.
PreparedQueue& GetPreparedRequests();
//! Gets the prepared requests that are queued to be processed.
const PreparedQueue& GetPreparedRequests() const;
//! Marks a request as completed so the main thread in Streamer can close it out.
//! This can be safely called from multiple threads.
void MarkRequestAsCompleted(FileRequest* request);
//! Rejects a request by removing it from the chain and recycling it.
//! Only requests without children can be rejected. If the rejected request has a parent it might need to be processed
//! further.
//! @param request The request to remove and recycle.
//! @return The parent request of the rejected request or null if there was no parent.
FileRequest* RejectRequest(FileRequest* request);
//! Adds an old request to the recycle bin so it can be reused later.
void RecycleRequest(FileRequest* request);
//! Adds an old external request to the recycle bin so it can be reused later.
void RecycleRequest(ExternalFileRequest* request);
//! Marks a request as completed so the main thread in Streamer can close it out.
//! This can be safely called from multiple threads.
void MarkRequestAsCompleted(FileRequest* request);
//! Rejects a request by removing it from the chain and recycling it.
//! Only requests without children can be rejected. If the rejected request has a parent it might need to be processed
//! further.
//! @param request The request to remove and recycle.
//! @return The parent request of the rejected request or null if there was no parent.
FileRequest* RejectRequest(FileRequest* request);
//! Adds an old request to the recycle bin so it can be reused later.
void RecycleRequest(FileRequest* request);
//! Adds an old external request to the recycle bin so it can be reused later.
void RecycleRequest(ExternalFileRequest* request);
//! Does the FinalizeRequest callback where appropriate and does some bookkeeping to finalize requests.
//! @return True if any requests were finalized, otherwise false.
bool FinalizeCompletedRequests();
//! Does the FinalizeRequest callback where appropriate and does some bookkeeping to finalize requests.
//! @return True if any requests were finalized, otherwise false.
bool FinalizeCompletedRequests();
//! Causes the main thread for streamer to wake up and process any pending requests. If the thread
//! is already awake, nothing happens.
void WakeUpSchedulingThread();
//! If there's no pending messages this will cause the main thread for streamer to go to sleep.
void SuspendSchedulingThread();
//! Returns the native primitive(s) used to suspend and wake up the scheduling thread and possibly other threads.
AZ::Platform::StreamerContextThreadSync& GetStreamerThreadSynchronizer();
//! Causes the main thread for streamer to wake up and process any pending requests. If the thread
//! is already awake, nothing happens.
void WakeUpSchedulingThread();
//! If there's no pending messages this will cause the main thread for streamer to go to sleep.
void SuspendSchedulingThread();
//! Returns the native primitive(s) used to suspend and wake up the scheduling thread and possibly other threads.
AZ::Platform::StreamerContextThreadSync& GetStreamerThreadSynchronizer();
//! Collects statistics recorded during processing. This will only return statistics for the
//! context. Use the CollectStatistics on AZ::IO::Streamer to get all statistics.
void CollectStatistics(AZStd::vector<Statistic>& statistics);
//! Collects statistics recorded during processing. This will only return statistics for the
//! context. Use the CollectStatistics on AZ::IO::Streamer to get all statistics.
void CollectStatistics(AZStd::vector<Statistic>& statistics);
private:
//! Gets a new FileRequestPtr. This version is for internal use only and is not thread-safe.
//! This will be called by GetNewExternalRequest or GetNewExternalRequestBatch which are responsible
//! for managing the lock to the recycle bin.
FileRequestPtr GetNewExternalRequestUnguarded();
private:
//! Gets a new FileRequestPtr. This version is for internal use only and is not thread-safe.
//! This will be called by GetNewExternalRequest or GetNewExternalRequestBatch which are responsible
//! for managing the lock to the recycle bin.
FileRequestPtr GetNewExternalRequestUnguarded();
inline static constexpr size_t s_initialRecycleBinSize = 64;
inline static constexpr size_t s_initialRecycleBinSize = 64;
AZStd::mutex m_externalRecycleBinGuard;
AZStd::vector<ExternalFileRequest*> m_externalRecycleBin;
AZStd::vector<FileRequest*> m_internalRecycleBin;
// The completion is guarded so other threads can perform async IO and safely mark requests as completed.
AZStd::recursive_mutex m_completedGuard;
AZStd::queue<FileRequest*> m_completed;
AZStd::mutex m_externalRecycleBinGuard;
AZStd::vector<ExternalFileRequest*> m_externalRecycleBin;
AZStd::vector<FileRequest*> m_internalRecycleBin;
// The prepared request queue is not guarded and should only be called from the main Streamer thread.
PreparedQueue m_preparedRequests;
// The completion is guarded so other threads can perform async IO and safely mark requests as completed.
AZStd::recursive_mutex m_completedGuard;
AZStd::queue<FileRequest*> m_completed;
// The prepared request queue is not guarded and should only be called from the main Streamer thread.
PreparedQueue m_preparedRequests;
#if AZ_STREAMER_ADD_EXTRA_PROFILING_INFO
//! By how much time the prediction was off. This mostly covers the latter part of scheduling, which
//! gets more precise the closer the request gets to completion.
AZ::Statistics::RunningStatistic m_predictionAccuracyUsStat;
//! By how much time the prediction was off. This mostly covers the latter part of scheduling, which
//! gets more precise the closer the request gets to completion.
AZ::Statistics::RunningStatistic m_predictionAccuracyUsStat;
//! Tracks the percentage of requests with late predictions where the request completed earlier than expected,
//! versus the requests that completed later than predicted.
AZ::Statistics::RunningStatistic m_latePredictionsPercentageStat;
//! Tracks the percentage of requests with late predictions where the request completed earlier than expected,
//! versus the requests that completed later than predicted.
AZ::Statistics::RunningStatistic m_latePredictionsPercentageStat;
//! Percentage of requests that missed their deadline. If percentage is too high it can indicate that
//! there are too many file requests or the deadlines for requests are too tight.
AZ::Statistics::RunningStatistic m_missedDeadlinePercentageStat;
//! Percentage of requests that missed their deadline. If percentage is too high it can indicate that
//! there are too many file requests or the deadlines for requests are too tight.
AZ::Statistics::RunningStatistic m_missedDeadlinePercentageStat;
#endif // AZ_STREAMER_ADD_EXTRA_PROFILING_INFO
//! Platform-specific synchronization object used to suspend the Streamer thread and wake it up to resume procesing.
AZ::Platform::StreamerContextThreadSync m_threadSync;
//! Platform-specific synchronization object used to suspend the Streamer thread and wake it up to resume procesing.
AZ::Platform::StreamerContextThreadSync m_threadSync;
size_t m_pendingIdCounter{ 0 };
};
} // namespace IO
} // namespace AZ
size_t m_pendingIdCounter{ 0 };
};
} // namespace AZ::IO
Code/Framework/AzCore/Platform/Windows/AzCore/IO/Streamer/StorageDrive_Windows.cpp
+5 -5
View File
@@ -73,7 +73,7 @@ namespace AZ::IO
, m_constructionOptions(options)
{
AZ_Assert(!drivePaths.empty(), "StorageDrive_win requires at least one drive path to work.");
// Get drive paths
m_drivePaths.reserve(drivePaths.size());
for (AZStd::string_view drivePath : drivePaths)
@@ -583,7 +583,7 @@ namespace AZ::IO
// If any are unaligned to the sector sizes, make adjustments and allocate an aligned buffer.
const bool alignedAddr = IStreamerTypes::IsAlignedTo(data->m_output, aznumeric_caster(m_physicalSectorSize));
const bool alignedOffs = IStreamerTypes::IsAlignedTo(data->m_offset, aznumeric_caster(m_logicalSectorSize));
// Adjust the offset if it's misaligned.
// Align the offset down to next lowest sector.
// Change the size to compensate.
@@ -656,7 +656,7 @@ namespace AZ::IO
Statistic::PlotImmediate(m_name, DirectReadsName, m_directReadsPercentageStat.GetMostRecentSample());
#endif // AZ_STREAMER_ADD_EXTRA_PROFILING_INFO
}
FileReadStatus& readStatus = m_readSlots_statusInfo[readSlot];
LPOVERLAPPED overlapped = &readStatus.m_overlapped;
overlapped->Offset = aznumeric_caster(readOffs);
@@ -716,7 +716,7 @@ namespace AZ::IO
Statistic::PlotImmediate(m_name, FileSwitchesName, m_fileSwitchPercentageStat.GetMostRecentSample());
Statistic::PlotImmediate(m_name, SeeksName, m_seekPercentageStat.GetMostRecentSample());
#endif // AZ_STREAMER_ADD_EXTRA_PROFILING_INFO
m_fileCache_activeReads[fileCacheSlot]++;
m_activeCacheSlot = fileCacheSlot;
m_activeOffset = readOffs + readSize;
@@ -1007,7 +1007,7 @@ namespace AZ::IO
auto readCommand = AZStd::get_if<FileRequest::ReadData>(&fileReadInfo.m_request->GetCommand());
AZ_Assert(readCommand != nullptr, "Request stored with the overlapped I/O call did not contain a read request.");
if (fileReadInfo.m_sectorAlignedOutput && !encounteredError)
{
auto offsetAddress = reinterpret_cast<u8*>(fileReadInfo.m_sectorAlignedOutput) + fileReadInfo.m_copyBackOffset;
Code/Framework/AzCore/Platform/Windows/AzCore/IO/Streamer/StorageDrive_Windows.h
+2 -2
View File
@@ -40,7 +40,7 @@ namespace AZ::IO
//! make adjustments. For the most optimal performance align read buffers to the physicalSectorSize.
u8 m_enableUnbufferedReads : 1;
//! Globally enable file sharing. This allows files to used outside AZ::IO::Streamer, including other applications
//! while in use by AZ::IO::Streamer.
//! while in use by AZ::IO::Streamer.
u8 m_enableSharing : 1;
//! If true, only information that's explicitly requested or issues are reported. If false, status information
//! such as when drives are created and destroyed is reported as well.
@@ -99,7 +99,7 @@ namespace AZ::IO
FileRequest* m_request{ nullptr };
void* m_sectorAlignedOutput{ nullptr }; // Internally allocated buffer that is sector aligned.
size_t m_copyBackOffset{ 0 };
void AllocateAlignedBuffer(size_t size, size_t sectorSize);
void Clear();
};
Code/Framework/AzCore/Tests/Streamer/FullDecompressorTests.cpp
+4 -4
View File
@@ -89,7 +89,7 @@ namespace AZ::IO
m_context = nullptr;
AllocatorInstance<ThreadPoolAllocator>::Destroy();
AllocatorInstance<PoolAllocator>::Destroy();
AllocatorInstance<PoolAllocator>::Destroy();
UnitTest::AllocatorsFixture::TearDown();
}
@@ -123,7 +123,7 @@ namespace AZ::IO
.WillRepeatedly(Return(false));
EXPECT_CALL(*m_mock, QueueRequest(_));
EXPECT_CALL(*m_mock, UpdateStatus(_)).Times(AnyNumber());
switch (mockResult)
{
case ReadResult::Success:
@@ -267,7 +267,7 @@ namespace AZ::IO
{
allCompleted = allCompleted && request.GetStatus() == IStreamerTypes::RequestStatus::Completed;
};
FileRequest* requests[count];
AZStd::unique_ptr<u32[]> buffers[count];
for (size_t i = 0; i < count; ++i)
@@ -300,7 +300,7 @@ namespace AZ::IO
size = size >> 2;
for (u64 i = 0; i < size; ++i)
{
// Using assert here because in case of a problem EXPECT would
// Using assert here because in case of a problem EXPECT would
// cause a large amount of log noise.
ASSERT_EQ(buffer[i], offset + (i << 2));
}
Code/Framework/AzCore/Tests/Streamer/ReadSplitterTests.cpp
+2 -2
View File
@@ -359,7 +359,7 @@ namespace AZ::IO
.Times(2)
.WillRepeatedly([this](FileRequest* request) { m_context.MarkRequestAsCompleted(request); });
m_context.FinalizeCompletedRequests();
azfree(memory);
}
@@ -415,7 +415,7 @@ namespace AZ::IO
m_context.FinalizeCompletedRequests();
EXPECT_EQ(2, completedRequests);
azfree(memory1);
azfree(memory0);
}
Code/Framework/AzCore/Tests/Streamer/SchedulerTests.cpp
+6 -6
View File
@@ -30,7 +30,7 @@ namespace AZ::IO
{
using ::testing::_;
using ::testing::AnyNumber;
UnitTest::AllocatorsFixture::SetUp();
m_mock = AZStd::make_shared<StreamStackEntryMock>();
@@ -78,7 +78,7 @@ namespace AZ::IO
{
using ::testing::_;
using ::testing::AtLeast;
EXPECT_CALL(*m_mock, UpdateStatus(_)).Times(AtLeast(1));
EXPECT_CALL(*m_mock, UpdateCompletionEstimates(_, _, _, _)).Times(AtLeast(1));
EXPECT_CALL(*m_mock, PrepareRequest(_))
@@ -115,7 +115,7 @@ namespace AZ::IO
void MockAllocatorForUnclaimedMemory(IStreamerTypes::RequestMemoryAllocatorMock& mock, AZStd::binary_semaphore& sync)
{
using ::testing::_;
EXPECT_CALL(mock, LockAllocator()).Times(1);
EXPECT_CALL(mock, UnlockAllocator())
.Times(1)
@@ -256,13 +256,13 @@ namespace AZ::IO
using ::testing::_;
using ::testing::AtLeast;
using ::testing::Return;
EXPECT_CALL(*m_mock, UpdateStatus(_)).Times(AtLeast(1));
EXPECT_CALL(*m_mock, UpdateCompletionEstimates(_, _, _, _)).Times(AtLeast(1));
EXPECT_CALL(*m_mock, PrepareRequest(_)).Times(AtLeast(1));
EXPECT_CALL(*m_mock, ExecuteRequests()).Times(AtLeast(1));
EXPECT_CALL(*m_mock, QueueRequest(_)).Times(1);
AZStd::atomic_int counter = 2;
AZStd::binary_semaphore sync;
auto wait = [&sync, &counter](FileRequestHandle)
@@ -350,7 +350,7 @@ namespace AZ::IO
EXPECT_CALL(*m_mock, UpdateStatus(_)).Times(AnyNumber());
EXPECT_CALL(*m_mock, UpdateCompletionEstimates(_, _, _, _)).Times(AnyNumber());
// Pretend to be busy [Iterations] times, then set the status to idle so the Scheduler thread can exit.
EXPECT_CALL(*m_mock, ExecuteRequests())
.Times(Iterations + 1)
Code/Framework/AzCore/Tests/Streamer/StreamStackEntryConformityTests.h
+6 -6
View File
@@ -97,7 +97,7 @@ namespace AZ::IO
TYPED_TEST_P(StreamStackEntryConformityTests, SetContext_ContextIsForwardedToNext_SetContextOnMockIsCalled)
{
using ::testing::_;
auto mock = AZStd::make_shared<StreamStackEntryMock>();
auto entry = this->m_description.CreateInstance();
entry.SetNext(mock);
@@ -194,14 +194,14 @@ namespace AZ::IO
TYPED_TEST_P(StreamStackEntryConformityTests, UpdateStatus_ForwardsCallToNext_NextRecievedCall)
{
using ::testing::_;
auto mock = AZStd::make_shared<StreamStackEntryMock>();
auto entry = this->m_description.CreateInstance();
entry.SetNext(mock);
EXPECT_CALL(*mock, UpdateStatus(_))
.Times(1);
StreamStackEntry::Status status;
entry.UpdateStatus(status);
}
@@ -241,7 +241,7 @@ namespace AZ::IO
TYPED_TEST_P(StreamStackEntryConformityTests, UpdateStatus_NextHasSmallerNumSlots_ReturnsSmallestNumSlots)
{
using ::testing::_;
if (this->m_description.UsesSlots())
{
auto mock = AZStd::make_shared<StreamStackEntryMock>();
@@ -264,7 +264,7 @@ namespace AZ::IO
TYPED_TEST_P(StreamStackEntryConformityTests, UpdateStatus_NextHasLargerNumSlots_ReturnsSmallestNumSlots)
{
using ::testing::_;
if (this->m_description.UsesSlots())
{
auto mock = AZStd::make_shared<StreamStackEntryMock>();
@@ -289,7 +289,7 @@ namespace AZ::IO
TYPED_TEST_P(StreamStackEntryConformityTests, UpdateCompletionEstimates_ForwardsCallToNext_NextRecievedCall)
{
using ::testing::_;
auto mock = AZStd::make_shared<StreamStackEntryMock>();
auto entry = this->m_description.CreateInstance();
entry.SetNext(mock);
Code/Framework/AzCore/Tests/StreamerTests.cpp
+562 -565
View File
File diff suppressed because it is too large Load Diff
Code/Framework/AzFramework/AzFramework/IO/RemoteStorageDrive.cpp
+6 -6
View File
@@ -199,7 +199,7 @@ namespace AzFramework
{
activeFile = &m_filePaths[m_activeCacheSlot];
}
// Estimate requests in this stack entry.
for (FileRequest* request : m_pendingRequests)
{
@@ -279,7 +279,7 @@ namespace AzFramework
using namespace AZ::IO;
AZ_PROFILE_FUNCTION(AzCore);
auto data = AZStd::get_if<FileRequest::ReadData>(&request->GetCommand());
AZ_Assert(data, "Request doing reading in the RemoteStorageDrive didn't contain read data.");
@@ -292,7 +292,7 @@ namespace AzFramework
file = m_fileHandles[cacheIndex];
m_fileLastUsed[cacheIndex] = AZStd::chrono::high_resolution_clock::now();
}
// If the file is not open, eject the oldest entry from the cache and open the file for reading.
if (file == InvalidHandle)
{
@@ -325,7 +325,7 @@ namespace AzFramework
}
m_activeCacheSlot = cacheIndex;
AZ_Assert(file != InvalidHandle,
AZ_Assert(file != InvalidHandle,
"While searching for file '%s' RemoteStorageDevice::ReadFile encountered a problem that wasn't reported.", data->m_path.GetRelativePath());
{
TIMED_AVERAGE_WINDOW_SCOPE(m_readTimeAverage);
@@ -357,7 +357,7 @@ namespace AzFramework
}
}
m_readSizeAverage.PushEntry(data->m_size);
request->SetStatus(IStreamerTypes::RequestStatus::Completed);
m_context->MarkRequestAsCompleted(request);
}
@@ -507,7 +507,7 @@ namespace AzFramework
using namespace AZ::IO;
using DoubleSeconds = AZStd::chrono::duration<double>;
double totalBytesReadMB = m_readSizeAverage.GetTotal() / (1024.0 * 1024.0);
double totalReadTimeSec = AZStd::chrono::duration_cast<DoubleSeconds>(m_readTimeAverage.GetTotal()).count();
if (m_readSizeAverage.GetTotal() > 1) // A default is always added.
Code/Framework/AzFramework/AzFramework/IO/RemoteStorageDrive.h
+1 -1
View File
@@ -53,7 +53,7 @@ namespace AzFramework
protected:
static constexpr AZ::s32 s_maxRequests = 1;
void ReadFile(AZ::IO::FileRequest* request);
bool CancelRequest(AZ::IO::FileRequest* cancelRequest, AZ::IO::FileRequestPtr& target);
void FileExistsRequest(AZ::IO::FileRequest* request);
Code/Framework/AzFramework/AzFramework/Script/ScriptComponent.cpp
+11 -11
View File
@@ -142,7 +142,7 @@ namespace AzFramework
AZ::Outcome<void, AZStd::string> CompileScript(ScriptCompileRequest& request, AZ::ScriptContext& scriptContext)
{
AZ_TracePrintf(request.m_errorWindow.data(), "Starting script compile.\n");
AZStd::string debugName = "@";
debugName += request.m_sourceFile;
AZStd::to_lower(debugName.begin(), debugName.end());
@@ -180,14 +180,14 @@ namespace AzFramework
{
using namespace AZ::IO;
FileIOStream outputStream;
if (!outputStream.Open(request.m_destPath.c_str(), OpenMode::ModeWrite | OpenMode::ModeBinary))
{
return AZ::Failure(AZStd::string("Failed to open output file %s", request.m_destPath.data()));
}
request.m_output = &outputStream;
if (writeAssetInfo)
{
if (request.m_prewriteCallback)
@@ -292,7 +292,7 @@ namespace AzFramework
namespace Internal
{
AZStd::string PrintLuaValue(lua_State* lua, int stackIdx, int depth = 0)
{
constexpr int MaxDepth = 4;
@@ -302,7 +302,7 @@ namespace AzFramework
}
const int elementType = lua_type(lua, stackIdx);
switch (elementType)
{
case LUA_TSTRING:
@@ -347,7 +347,7 @@ namespace AzFramework
{
keyValuePairs += " ";
}
}
}
}
tableStr += keyValuePairs.length() < 1024 ? keyValuePairs : AZStd::string::format("too many keys (%i)!", keyCount);
@@ -891,18 +891,18 @@ namespace AzFramework
// This is the root table (properties) it will be used as properties for all sub tables
// ScriptComponents can share the same lua script asset, but each instance's Properties table needs to be unique.
// This way the script can change a property at runtime and not affect the other ScriptComponents which are using the same script.
// For normal properties we will create new variable instances, but NetSynched variables aren't stored in Lua, and instead
// For normal properties we will create new variable instances, but NetSynched variables aren't stored in Lua, and instead
// are retrieved using the __index and __newIndex metamethods.
// Ensure that this instance of Properties table has the proper __index and __newIndex metamethods.
lua_newtable(lua); // This new table will become the Properties instance metatable. Stack: ScriptRootTable PropertiesTable EntityTable "Properties" {} {}
lua_newtable(lua); // This new table will become the Properties instance metatable. Stack: ScriptRootTable PropertiesTable EntityTable "Properties" {} {}
lua_pushliteral(lua, "__index"); // Stack: ScriptRootTable PropertiesTable EntityTable "Properties" {} {} __index
lua_pushcclosure(lua, &Internal::Properties__Index, 0); // Stack: ScriptRootTable PropertiesTable EntityTable "Properties" {} {} __index function
lua_rawset(lua, -3); // Stack: ScriptRootTable PropertiesTable EntityTable "Properties" {} {__index=Internal::Properties__Index}
lua_rawset(lua, -3); // Stack: ScriptRootTable PropertiesTable EntityTable "Properties" {} {__index=Internal::Properties__Index}
lua_pushliteral(lua, "__newindex");
lua_pushcclosure(lua, &Internal::Properties__NewIndex, 0);
lua_rawset(lua, -3); // Stack: ScriptRootTable PropertiesTable EntityTable "Properties" {} {__index=Internal::Properties__Index __newindex=Internal::Properties__NewIndex}
lua_setmetatable(lua, -2); // Stack: ScriptRootTable PropertiesTable EntityTable "Properties" {Meta{__index=Internal::Properties__Index __newindex=Internal::Properties__NewIndex} }
lua_rawset(lua, -3); // Stack: ScriptRootTable PropertiesTable EntityTable "Properties" {} {__index=Internal::Properties__Index __newindex=Internal::Properties__NewIndex}
lua_setmetatable(lua, -2); // Stack: ScriptRootTable PropertiesTable EntityTable "Properties" {Meta{__index=Internal::Properties__Index __newindex=Internal::Properties__NewIndex} }
metatableIndex = lua_gettop(lua); // This will be the metatable for all subtables
}
Code/Framework/AzToolsFramework/AzToolsFramework/ToolsComponents/ToolsAssetCatalogComponent.h
+1 -1
View File
@@ -37,7 +37,7 @@ namespace AssetProcessor
public AZ::Data::AssetCatalog
{
public:
AZ_COMPONENT(ToolsAssetCatalogComponent, "{AE68E46B-0E21-499A-8309-41408BCBE4BF}");
ToolsAssetCatalogComponent() = default;