Reece Wilson
13480a0473
[*] IOProcessor's stop method now returns true on success instea of true on delegated stop
628 lines
16 KiB
C++
628 lines
16 KiB
C++
/***
|
|
Copyright (C) 2021 J Reece Wilson (a/k/a "Reece"). All rights reserved.
|
|
|
|
File: Async.NT.cpp
|
|
Date: 2021-9-13
|
|
Author: Reece
|
|
***/
|
|
#include <Source/RuntimeInternal.hpp>
|
|
#include "FS.hpp"
|
|
#include <Source/IO/Loop/LSHandle.hpp>
|
|
//#include "IPCHandle.hpp"
|
|
#include "Async.NT.hpp"
|
|
#include "FileAdvisory.NT.hpp"
|
|
#include <Source/IO/Loop/Loop.hpp>
|
|
#include <Source/IO/IPC/IPCPipe.NT.hpp>
|
|
|
|
namespace Aurora::IO::FS
|
|
{
|
|
struct NtAsyncFileTransactionLoopSource : AuLoop::LSHandle
|
|
{
|
|
NtAsyncFileTransactionLoopSource(AuSPtr<NtAsyncFileTransaction> that) : caller_(that), Loop::LSHandle(AuUInt(that->event_))
|
|
{}
|
|
|
|
virtual bool IsSignaled() override;
|
|
virtual bool OnTrigger(AuUInt handle) override;
|
|
virtual AuLoop::ELoopSource GetType() override;
|
|
|
|
private:
|
|
AuWPtr<NtAsyncFileTransaction> 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:
|
|
{
|
|
CreateDirectories(pathex, true);
|
|
fileHandle = CreateFileW(win32Path.c_str(), GENERIC_WRITE | GENERIC_READ, NtLockAdvisoryToShare(lock), NULL, OPEN_ALWAYS, flags, NULL);
|
|
if (fileHandle == INVALID_HANDLE_VALUE)
|
|
{
|
|
fileHandle = CreateFileW(win32Path.c_str(), GENERIC_WRITE | GENERIC_READ, NtLockAdvisoryToShare(lock), NULL, CREATE_ALWAYS, flags, NULL);
|
|
}
|
|
break;
|
|
}
|
|
case EFileOpenMode::eWrite:
|
|
{
|
|
CreateDirectories(pathex, true);
|
|
fileHandle = CreateFileW(win32Path.c_str(), 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<FileHandle> NtAsyncFileStream::GetHandle()
|
|
{
|
|
return handle_;
|
|
}
|
|
|
|
void NtAsyncFileStream::Init(const AuSPtr<FileHandle> &handle)
|
|
{
|
|
this->handle_ = handle;
|
|
}
|
|
|
|
AuSPtr<IAsyncTransaction> NtAsyncFileStream::NewTransaction()
|
|
{
|
|
auto shared = AuMakeShared<NtAsyncFileTransaction>();
|
|
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;
|
|
}
|
|
|
|
OVERLAPPED a {};
|
|
a.hEvent = CreateEventA(NULL, true, 0, NULL);
|
|
|
|
DWORD read;
|
|
if (!::ReadFile(this->handle_->handle, parameters.ptr, parameters.length, NULL, &a) &&
|
|
::GetLastError() != ERROR_IO_PENDING)
|
|
{
|
|
SysPushErrorIO();
|
|
::CloseHandle(a.hEvent);
|
|
return false;
|
|
}
|
|
|
|
::WaitForSingleObject(a.hEvent, 0);
|
|
if (!::GetOverlappedResult(this->handle_->handle, &a, &read, true))
|
|
{
|
|
::CloseHandle(a.hEvent);
|
|
return false;
|
|
}
|
|
::CloseHandle(a.hEvent);
|
|
|
|
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;
|
|
}
|
|
|
|
OVERLAPPED a {};
|
|
a.hEvent = CreateEventA(NULL, true, 0, NULL);
|
|
|
|
DWORD read;
|
|
if (!::WriteFile(this->handle_->writeHandle, parameters.ptr, parameters.length, NULL, &a) &&
|
|
::GetLastError() != ERROR_IO_PENDING)
|
|
{
|
|
SysPushErrorIO();
|
|
::CloseHandle(a.hEvent);
|
|
return false;
|
|
}
|
|
|
|
::WaitForSingleObject(a.hEvent, 0);
|
|
if (!::GetOverlappedResult(this->handle_->writeHandle, &a, &read, true))
|
|
{
|
|
::CloseHandle(a.hEvent);
|
|
return false;
|
|
}
|
|
|
|
::CloseHandle(a.hEvent);
|
|
parameters.outVariable = read;
|
|
return true;
|
|
}
|
|
|
|
bool NtAsyncFileTransaction::Init(const AuSPtr<FileHandle> &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)
|
|
{
|
|
auto er = GetLastError();
|
|
if (val)
|
|
{
|
|
return true;
|
|
}
|
|
else if (er == er == ERROR_IO_PENDING)
|
|
{
|
|
return true;
|
|
}
|
|
else if (er == ERROR_BROKEN_PIPE||
|
|
er == ERROR_HANDLE_EOF)
|
|
{
|
|
SetEvent(that->event_);
|
|
|
|
that->lastAbstractStat_ = 0;
|
|
// also required:
|
|
that->hasFailed = true; // to pass completion
|
|
that->osErrorCode = er;// to suppress actual error condition
|
|
|
|
auto pipe = that->ntIpcPipeImpl.lock();
|
|
|
|
that->DispatchCb(0);
|
|
|
|
if (pipe)
|
|
{
|
|
pipe->OnEndOfReadStream();
|
|
}
|
|
|
|
that->memoryHold_.reset();
|
|
return true;
|
|
}
|
|
else
|
|
{
|
|
that->pin_.reset();
|
|
that->Reset();
|
|
that->osErrorCode = er;
|
|
that->hasFailed = true;
|
|
SysPushErrorFIO("QoA async FIO error: {} {}", that->GetFileHandle()->path, that->osErrorCode);
|
|
return false;
|
|
}
|
|
}
|
|
|
|
static VOID WINAPI GenericCompletionRoutine(
|
|
_In_ DWORD dwErrorCode,
|
|
_In_ DWORD dwNumberOfBytesTransfered,
|
|
_Inout_ LPOVERLAPPED lpOverlapped
|
|
)
|
|
{
|
|
auto transaction = reinterpret_cast<NtAsyncFileTransaction *>(reinterpret_cast<AuUInt8*>(lpOverlapped) - offsetof(NtAsyncFileTransaction, overlap_));
|
|
auto hold = AuExchange(transaction->pin_, {});
|
|
if (dwErrorCode)
|
|
{
|
|
hold->hasFailed = true;
|
|
hold->osErrorCode = dwErrorCode;
|
|
}
|
|
else if (!hold->lastAbstractStat_)
|
|
{
|
|
return;
|
|
}
|
|
|
|
SetEvent(lpOverlapped->hEvent);
|
|
|
|
hold->lastAbstractStat_ = 0;
|
|
hold->CompleteEx(dwNumberOfBytesTransfered);
|
|
hold->memoryHold_.reset();
|
|
}
|
|
|
|
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<AuMemoryViewWrite> &memoryView)
|
|
{
|
|
if (this->isIrredeemable_)
|
|
{
|
|
SysPushErrorIO("Transaction was signaled to be destroyed to reset mid synchronizable operation. You can no longer use this stream object");
|
|
return false;
|
|
}
|
|
|
|
if (!IDontWannaUsePorts())
|
|
{
|
|
return false;
|
|
}
|
|
|
|
if (!memoryView)
|
|
{
|
|
SysPushErrorArg();
|
|
return {};
|
|
}
|
|
|
|
if (this->memoryHold_)
|
|
{
|
|
SysPushErrorIO("IO Operation in progress");
|
|
return {};
|
|
}
|
|
|
|
this->latch_ = false;
|
|
::ResetEvent(this->event_);
|
|
|
|
this->memoryHold_ = memoryView;
|
|
this->hasFailed = false;
|
|
|
|
this->lastAbstractStat_ = memoryView->length;
|
|
this->lastAbstractOffset_ = offset;
|
|
this->lastBytes = 0;
|
|
|
|
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<AuMemoryViewRead> &memoryView)
|
|
{
|
|
if (this->isIrredeemable_)
|
|
{
|
|
SysPushErrorIO("Transaction was signaled to be destroyed to reset mid synchronizable operation. You can no longer use this stream object");
|
|
return false;
|
|
}
|
|
|
|
if (!IDontWannaUsePorts())
|
|
{
|
|
return false;
|
|
}
|
|
|
|
if (!memoryView)
|
|
{
|
|
SysPushErrorArg();
|
|
return {};
|
|
}
|
|
|
|
if (this->memoryHold_)
|
|
{
|
|
SysPushErrorIO("IO Operation in progress");
|
|
return {};
|
|
}
|
|
|
|
this->latch_ = false;
|
|
::ResetEvent(this->event_);
|
|
|
|
this->hasFailed = false;
|
|
|
|
this->memoryHold_ = memoryView;
|
|
this->lastBytes = 0;
|
|
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)
|
|
{
|
|
this->lastBytes = read;
|
|
|
|
if (AuExchange(this->latch_, true))
|
|
{
|
|
return;
|
|
}
|
|
|
|
this->lastBytes = read;
|
|
|
|
if (this->sub_)
|
|
{
|
|
this->sub_->OnAsyncFileOpFinished(this->lastAbstractOffset_, read);
|
|
}
|
|
|
|
}
|
|
|
|
void NtAsyncFileTransaction::Reset()
|
|
{
|
|
if (this->lastAbstractStat_)
|
|
{
|
|
this->isIrredeemable_ = true;
|
|
this->hasFailed = true;
|
|
::CancelIoEx(this->handle_->handle, &this->overlap_);
|
|
::SetEvent(this->event_);
|
|
this->osErrorCode = ERROR_ABANDONED_WAIT_0;
|
|
}
|
|
else
|
|
{
|
|
::ResetEvent(this->event_);
|
|
this->hasFailed = false;
|
|
}
|
|
|
|
this->lastBytes = 0;
|
|
this->lastAbstractStat_ = 0;
|
|
}
|
|
|
|
bool NtAsyncFileTransaction::Failed()
|
|
{
|
|
return this->hasFailed &&
|
|
this->osErrorCode != ERROR_BROKEN_PIPE &&
|
|
this->osErrorCode != ERROR_HANDLE_EOF;
|
|
}
|
|
|
|
AuUInt NtAsyncFileTransaction::GetOSErrorCode()
|
|
{
|
|
return this->hasFailed ? this->osErrorCode : ERROR_SUCCESS;
|
|
}
|
|
|
|
bool NtAsyncFileTransaction::CompleteEx(AuUInt completeRoutine)
|
|
{
|
|
DWORD read {};
|
|
|
|
|
|
if (this->isIrredeemable_)
|
|
{
|
|
::ResetEvent(this->event_);
|
|
return true;
|
|
}
|
|
|
|
if (!completeRoutine)
|
|
{
|
|
if ((this->hasFailed) ||
|
|
(this->lastBytes) ||
|
|
::GetOverlappedResult(this->handle_->handle, &this->overlap_, &read, false))
|
|
{
|
|
if (this->lastBytes || this->hasFailed)
|
|
{
|
|
return true;
|
|
}
|
|
|
|
this->lastAbstractStat_ = 0;
|
|
bool bLatched = this->latch_;
|
|
DispatchCb(read);
|
|
return read;
|
|
}
|
|
}
|
|
else
|
|
{
|
|
if (this->osErrorCode == ERROR_BROKEN_PIPE ||
|
|
this->osErrorCode == ERROR_HANDLE_EOF)
|
|
{
|
|
auto pipe = this->ntIpcPipeImpl.lock();
|
|
|
|
DispatchCb(0);
|
|
|
|
if (pipe)
|
|
{
|
|
pipe->OnEndOfReadStream();
|
|
}
|
|
}
|
|
else
|
|
{
|
|
DispatchCb(completeRoutine);
|
|
}
|
|
|
|
return true;
|
|
}
|
|
|
|
return false;
|
|
}
|
|
|
|
bool NtAsyncFileTransaction::Complete()
|
|
{
|
|
return CompleteEx(false);
|
|
}
|
|
|
|
AuUInt32 NtAsyncFileTransaction::GetLastPacketLength()
|
|
{
|
|
return this->lastBytes;
|
|
}
|
|
|
|
void NtAsyncFileTransaction::SetCallback(const AuSPtr<IAsyncFinishedSubscriber> &sub)
|
|
{
|
|
this->sub_ = sub;
|
|
}
|
|
|
|
bool NtAsyncFileTransaction::Wait(AuUInt32 timeout)
|
|
{
|
|
if (this->latch_)
|
|
{
|
|
return true;
|
|
}
|
|
|
|
DWORD ret;
|
|
do
|
|
{
|
|
ret = ::WaitForSingleObjectEx(this->event_, timeout ? timeout : INFINITE, true);
|
|
}
|
|
while (ret == WAIT_IO_COMPLETION);
|
|
|
|
if (ret == WAIT_OBJECT_0)
|
|
{
|
|
return Complete();
|
|
}
|
|
return false;
|
|
}
|
|
|
|
HANDLE NtAsyncFileTransaction::GetHandle()
|
|
{
|
|
return this->event_;
|
|
}
|
|
|
|
AuSPtr<FileHandle> NtAsyncFileTransaction::GetFileHandle()
|
|
{
|
|
return this->handle_;
|
|
}
|
|
|
|
AuSPtr<AuLoop::ILoopSource> NtAsyncFileTransaction::NewLoopSource()
|
|
{
|
|
return AuMakeShared<NtAsyncFileTransactionLoopSource>(AuSharedFromThis());
|
|
}
|
|
|
|
AUKN_SYM IAsyncFileStream *OpenAsyncNew(const AuString &path, EFileOpenMode openMode, bool directIO, EFileAdvisoryLockLevel lock)
|
|
{
|
|
AuSPtr<FileHandle> fileHandle;
|
|
NtAsyncFileStream *stream;
|
|
|
|
if (path.empty())
|
|
{
|
|
SysPushErrorParam("Empty path");
|
|
return {};
|
|
}
|
|
|
|
if (!EFileOpenModeIsValid(openMode))
|
|
{
|
|
SysPushErrorParam("Invalid open mode");
|
|
return {};
|
|
}
|
|
|
|
fileHandle = AuMakeShared<FileHandle>();
|
|
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<NtAsyncFileStream *>(handle);
|
|
}
|
|
} |