/*** Copyright (C) 2021 J Reece Wilson (a/k/a "Reece"). All rights reserved. File: Async.NT.cpp Date: 2021-9-13 Author: Reece ***/ #include #include "FS.hpp" #include "Async.NT.hpp" #include "FileAdvisory.NT.hpp" #include #include namespace Aurora::IO::FS { struct NtAsyncFileTransactionLoopSource : Loop::LSHandle { NtAsyncFileTransactionLoopSource(AuSPtr that) : caller_(that), Loop::LSHandle(AuUInt(that->GetFileHandle()->handle)) {} virtual bool IsSignaled() override; virtual bool OnTrigger(AuUInt handle) override; virtual Loop::ELoopSource GetType() override; private: AuWPtr caller_; }; bool NtAsyncFileTransactionLoopSource::OnTrigger(AuUInt handle) { auto lock = caller_.lock(); if (lock) { return lock->Complete(); } return true; } Loop::ELoopSource NtAsyncFileTransactionLoopSource::GetType() { return Loop::ELoopSource::eSourceAIO; } bool NtAsyncFileTransactionLoopSource::IsSignaled() { auto lock = caller_.lock(); if (lock) { return lock->Complete(); } return LSHandle::IsSignaled(); } NtAsyncFileTransaction::~NtAsyncFileTransaction() { AuWin32CloseHandle(this->event_); } FileHandle::~FileHandle() { if (this->writeHandle != this->handle) { AuWin32CloseHandle(this->writeHandle); } this->writeHandle = INVALID_HANDLE_VALUE; AuWin32CloseHandle(this->handle); } bool FileHandle::Init(const AuString &path, EFileOpenMode openMode, bool directIO, EFileAdvisoryLockLevel lock) { HANDLE fileHandle; auto pathex = NormalizePathRet(path); if (pathex.empty()) { return false; } auto win32Path = Locale::ConvertFromUTF8(pathex); if (win32Path.empty()) { return false; } auto flags = FILE_FLAG_OVERLAPPED; if (directIO) { flags |= FILE_FLAG_NO_BUFFERING; } fileHandle = INVALID_HANDLE_VALUE; switch (openMode) { case EFileOpenMode::eRead: { fileHandle = CreateFileW(win32Path.c_str(), GENERIC_READ, NtLockAdvisoryToShare(lock), NULL, OPEN_EXISTING, flags, NULL); break; } case EFileOpenMode::eReadWrite: case EFileOpenMode::eWrite: { CreateDirectories(pathex, true); fileHandle = CreateFileW(win32Path.c_str(), (openMode == EFileOpenMode::eReadWrite) ? (GENERIC_WRITE | GENERIC_READ) : GENERIC_WRITE, NtLockAdvisoryToShare(lock), NULL, CREATE_ALWAYS, flags, NULL); break; } } if (fileHandle == INVALID_HANDLE_VALUE) { SysPushErrorIO("Missing file: {}", path); return {}; } this->directIO = directIO; this->handle = fileHandle; this->writeHandle = fileHandle; this->readOnly = openMode == EFileOpenMode::eRead; return true; } void FileHandle::Init(HANDLE read, HANDLE write) { this->directIO = true; this->handle = read; this->writeHandle = write; this->readOnly = false; } AuSPtr NtAsyncFileStream::GetHandle() { return handle_; } void NtAsyncFileStream::Init(const AuSPtr &handle) { this->handle_ = handle; } AuSPtr NtAsyncFileStream::NewTransaction() { auto shared = AuMakeShared(); if (!shared) { return {}; } if (!shared->Init(this->handle_)) { return {}; } return shared; } bool NtAsyncFileStream::BlockingTruncate(AuUInt64 length) { LARGE_INTEGER i {}; i.QuadPart = length; if (!SetFilePointerEx(this->handle_->handle, i, nullptr, FILE_BEGIN)) { SysPushErrorIO(); return false; } return SetEndOfFile(this->handle_->handle); } bool NtAsyncFileStream::BlockingRead(AuUInt64 offset, const Memory::MemoryViewStreamWrite ¶meters) { LARGE_INTEGER i {}; i.QuadPart = offset; if (!SetFilePointerEx(this->handle_->handle, i, nullptr, FILE_BEGIN)) { SysPushErrorIO(); return false; } DWORD read; if (!::ReadFile(this->handle_->handle, parameters.ptr, parameters.length, &read, nullptr)) { SysPushErrorIO(); return false; } parameters.outVariable = read; return true; } bool NtAsyncFileStream::BlockingWrite(AuUInt64 offset, const Memory::MemoryViewStreamRead ¶meters) { LARGE_INTEGER i {}; i.QuadPart = offset; if (!SetFilePointerEx(this->handle_->handle, i, nullptr, FILE_BEGIN)) { SysPushErrorIO(); return false; } DWORD read; if (!::WriteFile(this->handle_->handle, parameters.ptr, parameters.length, &read, nullptr)) { SysPushErrorIO(); return false; } parameters.outVariable = read; return true; } bool NtAsyncFileTransaction::Init(const AuSPtr &handle) { this->handle_ = handle; this->overlap_.hEvent = this->event_ = CreateEventW(nullptr, true, false, nullptr); return this->overlap_.hEvent != INVALID_HANDLE_VALUE; } static bool TranslateNtStatus(NtAsyncFileTransaction *that, BOOL val) { if ((val) || (!val && (GetLastError() == ERROR_IO_PENDING))) { return true; } else { that->pin_.reset(); that->Reset(); SysPushErrorFIO("Async FIO error: {} {}", that->GetFileHandle()->path, GetLastError()); return false; } } static VOID WINAPI GenericCompletionRoutine( _In_ DWORD dwErrorCode, _In_ DWORD dwNumberOfBytesTransfered, _Inout_ LPOVERLAPPED lpOverlapped ) { auto transaction = reinterpret_cast(reinterpret_cast(lpOverlapped) - offsetof(NtAsyncFileTransaction, overlap_)); auto hold = AuExchange(transaction->pin_, {}); hold->Complete(); } bool NtAsyncFileTransaction::IDontWannaUsePorts() { // TODO: maybe we could use a semaphore counter of read attempts // to allow APC callbacks to drag behind if (AuExchange(this->pin_, AuSharedFromThis())) { while (SleepEx(100, true) == WAIT_IO_COMPLETION) { } if (AuExchange(this->pin_, AuSharedFromThis())) { SysPushErrorUnavailableError(); return {}; } } return true; } bool NtAsyncFileTransaction::StartRead(AuUInt64 offset, const AuSPtr &memoryView) { if (!IDontWannaUsePorts()) { return false; } this->latch_ = false; ::ResetEvent(this->event_); this->memoryHold_ = memoryView; this->lastAbstractStat_ = memoryView->length; this->lastAbstractOffset_ = offset; this->overlap_.Offset = offset & 0xFFFFFFFF; this->overlap_.OffsetHigh = (offset >> 32) & 0xFFFFFFFF; auto ret = ::ReadFileEx(this->handle_->handle, memoryView->ptr, memoryView->length, &overlap_, GenericCompletionRoutine); return TranslateNtStatus(this, ret); } bool NtAsyncFileTransaction::StartWrite(AuUInt64 offset, const AuSPtr &memoryView) { if (!IDontWannaUsePorts()) { return false; } this->latch_ = false; ::ResetEvent(this->event_); this->memoryHold_ = memoryView; this->lastAbstractStat_ = memoryView->length; this->lastAbstractOffset_ = offset; this->overlap_.Offset = offset & 0xFFFFFFFF; this->overlap_.OffsetHigh = (offset >> 32) & 0xFFFFFFFF; auto ret = ::WriteFileEx(this->handle_->writeHandle, memoryView->ptr, memoryView->length, &overlap_, GenericCompletionRoutine); return TranslateNtStatus(this, ret); } void NtAsyncFileTransaction::DispatchCb(AuUInt32 read) { if (AuExchange(this->latch_, true)) { return; } auto memoryHold = this->memoryHold_; this->memoryHold_.reset(); if (this->sub_) { this->sub_->OnAsyncFileOpFinished(this->lastAbstractOffset_, read); } } void NtAsyncFileTransaction::Reset() { ::ResetEvent(this->event_); this->lastAbstractStat_ = 0; // do not use latch } bool NtAsyncFileTransaction::Complete() { DWORD read; if (!this->lastAbstractStat_) { return false; } if (this->latch_) { return false; } if (GetOverlappedResult(this->handle_->handle, &this->overlap_, &read, false)) { bool bLatched = this->latch_; DispatchCb(read); return !bLatched; } return false; } AuUInt32 NtAsyncFileTransaction::GetLastPacketLength() { DWORD read {}; GetOverlappedResult(this->handle_->handle, &this->overlap_, &read, false); return read; } void NtAsyncFileTransaction::SetCallback(const AuSPtr &sub) { this->sub_ = sub; } bool NtAsyncFileTransaction::Wait(AuUInt32 timeout) { if (this->latch_) { return true; } auto ret = WaitForSingleObjectEx(this->event_, timeout ? timeout : INFINITE, true); if (ret == WAIT_OBJECT_0) { return Complete(); } return false; } HANDLE NtAsyncFileTransaction::GetHandle() { return this->event_; } AuSPtr NtAsyncFileTransaction::GetFileHandle() { return this->handle_; } AuSPtr NtAsyncFileTransaction::NewLoopSource() { return AuMakeShared(AuSharedFromThis()); } AUKN_SYM IAsyncFileStream *OpenAsyncNew(const AuString &path, EFileOpenMode openMode, bool directIO, EFileAdvisoryLockLevel lock) { AuSPtr fileHandle; NtAsyncFileStream *stream; if (path.empty()) { SysPushErrorParam("Empty path"); return {}; } if (!EFileOpenModeIsValid(openMode)) { SysPushErrorParam("Invalid open mode"); return {}; } fileHandle = AuMakeShared(); if (!fileHandle->Init(path, openMode, directIO, lock)) { return {}; } stream = _new NtAsyncFileStream(); if (!stream) { return {}; } stream->Init(fileHandle); return stream; } AUKN_SYM void OpenAsyncRelease(IAsyncFileStream *handle) { AuSafeDelete(handle); } }