AuroraSupport/AuroraRuntime
2
1
Fork 0
Code Issues 27 Pull Requests Releases Wiki Activity

[+/*] NT IPCPipe, updated interface for bidirectional comms. Introducing linux regression

Browse Source
This commit is contained in:
Reece Wilson 2022-04-17 13:40:08 +01:00
parent 30f86b81ac
commit 371006c04e
7 changed files with 515 additions and 73 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

4
Include/Aurora/IO/FS/IAsyncFileStream.hpp
View File

@ -14,5 +14,9 @@ namespace Aurora::IO::FS
struct IAsyncFileStream
{
virtual AuSPtr<IAsyncTransaction> NewTransaction() = 0;
virtual bool BlockingTruncate(AuUInt64 length) = 0;
virtual bool BlockingRead(AuUInt64 offset, const Memory::MemoryViewStreamWrite &parameters) = 0;
virtual bool BlockingWrite(AuUInt64 offset, const Memory::MemoryViewStreamRead &parameters) = 0;
};
}

50
Include/Aurora/IPC/IPCPipe.hpp
View File

@ -9,12 +9,54 @@
namespace Aurora::IPC
{
struct IPCPipe : IExportableIPC, Loop::ILoopSource
struct IPCPipe : IExportableIPC
{
virtual bool Read (const Memory::MemoryViewStreamWrite &write, bool nonblock = true) = 0;
virtual bool Write(const Memory::MemoryViewStreamRead &read, bool nonblock = true) = 0;
/**
* @brief A loopsource indicating whether or not the pipe has an established client
* @return
*/
virtual AuSPtr<Loop::ILoopSource> AsReadChannelIsOpen() = 0;
/**
* @brief A UNIX loopsource indicating when the pipe has data available.
* Wait semantics will be simlulated on Windows with a constant-step timer.
* @return
* @warning unimplemented on windows. IsSignaled()-as-has-data works nonblocking, though.
*/
virtual AuSPtr<Loop::ILoopSource> AsReadChannelHasData() = 0;
/**
* @brief Returns a shared async file interface to a session between the server
* Undefined behaviour is expected while AsReadChannelIsOpen()->IsSignaled() is false.
* @return
*/
virtual AuSPtr<IO::FS::IAsyncTransaction> NewAsyncTransaction() = 0;
/**
* @brief Reads at least write.length from the outbound stream of the opposite end
* @param nonblocking Wait for data
* @return
*/
virtual bool Read (const Memory::MemoryViewStreamWrite &write, bool nonblocking) = 0;
/**
* @brief Writes at least write.length to the stream on the opposite end
* @param nonblocking Wait for data
* @return
*/
virtual bool Write(const Memory::MemoryViewStreamRead &read) = 0;
};
/**
* @brief Creates a new exclusive IPC pipe
* @return
*/
AUKN_SYM AuSPtr<IPCPipe> NewPipe();
AUKN_SYM AuSPtr<IPCPipe> ImportPipe(const AuString &handle);
/**
* @brief Opens an exclusive IPC pipe as the client end given a ...
* @param handleString
* @return
*/
AUKN_SYM AuSPtr<IPCPipe> ImportPipe(const AuString &handleString);
}

7
Include/Aurora/Loop/ELoopSource.hpp
View File

@ -35,11 +35,9 @@ namespace Aurora::Loop
eSourceAsync,
// IPC pipe
eSourceIPCHasClient,
eSourceIPCReadPipe,
// glib oses only
eSourceGlib,
// Process and console APIs
eProcessStdIn,
eProcessStdOut,
@ -49,6 +47,9 @@ namespace Aurora::Loop
eSourceX11,
eSourceWin32,
// glib oses
eSourceGlib,
// internal
eSourceInternalReserved1,
eSourceInternalReserved2,

120
Source/IO/FS/Async.NT.cpp
View File

@ -14,63 +14,6 @@
namespace Aurora::IO::FS
{
struct FileHandle
{
~FileHandle();
bool Init(const AuString &path, EFileOpenMode openMode, bool directIO, EFileAdvisoryLockLevel lock);
HANDLE handle = INVALID_HANDLE_VALUE;
AuString path;
bool readOnly;
bool directIO;
};
struct NtAsyncFileStream : IAsyncFileStream
{
AuSPtr<IAsyncTransaction> NewTransaction() override;
void Init(const AuSPtr<FileHandle> &handle);
AuSPtr<FileHandle> GetHandle();
private:
AuSPtr<FileHandle> handle_;
};
struct NtAsyncFileTransaction : IAsyncTransaction, AuEnableSharedFromThis<NtAsyncFileTransaction>
{
~NtAsyncFileTransaction();
bool Init(const AuSPtr<FileHandle> &handle);
bool StartRead(AuUInt64 offset, const AuSPtr<AuMemoryViewWrite> &memoryView) override;
bool StartWrite(AuUInt64 offset, const AuSPtr<AuMemoryViewRead> &memoryView) override;
bool Complete() override;
AuUInt32 GetLastPacketLength() override;
void SetCallback(const AuSPtr<IAsyncFinishedSubscriber> &sub) override;
bool Wait(AuUInt32 timeout) override;
AuSPtr<Loop::ILoopSource> NewLoopSource() override;
void DispatchCb(AuUInt32 len);
HANDLE GetHandle();
AuSPtr<FileHandle> GetFileHandle();
// Required for a very evil hack
OVERLAPPED overlap_ {};
HANDLE event_ = INVALID_HANDLE_VALUE;
AuSPtr<NtAsyncFileTransaction> pin_;
AuUInt32 lastAbstractStat_ {}, lastAbstractOffset_ {};
private:
AuSPtr<void> memoryHold_;
AuSPtr<FileHandle> handle_;
bool latch_ {};
AuSPtr<IAsyncFinishedSubscriber> sub_;
};
struct NtAsyncFileTransactionLoopSource : Loop::LSHandle
{
NtAsyncFileTransactionLoopSource(AuSPtr<NtAsyncFileTransaction> that) : caller_(that), Loop::LSHandle(AuUInt(that->GetFileHandle()->handle))
@ -164,6 +107,13 @@ namespace Aurora::IO::FS
return true;
}
void FileHandle::Init(HANDLE h)
{
this->directIO = true;
this->handle = h;
this->readOnly = false;
}
AuSPtr<FileHandle> NtAsyncFileStream::GetHandle()
{
return handle_;
@ -190,6 +140,62 @@ namespace Aurora::IO::FS
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 &parameters)
{
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 &parameters)
{
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<FileHandle> &handle)
{
this->handle_ = handle;

60
Source/IO/FS/Async.NT.hpp
View File

@ -9,5 +9,65 @@
namespace Aurora::IO::FS
{
struct FileHandle
{
~FileHandle();
bool Init(const AuString &path, EFileOpenMode openMode, bool directIO, EFileAdvisoryLockLevel lock);
void Init(HANDLE h);
HANDLE handle {INVALID_HANDLE_VALUE};
AuString path;
bool readOnly;
bool directIO;
};
struct NtAsyncFileStream : IAsyncFileStream
{
AuSPtr<IAsyncTransaction> NewTransaction() override;
bool BlockingTruncate(AuUInt64 length) override;
bool BlockingRead(AuUInt64 offset, const Memory::MemoryViewStreamWrite &parameters) override;
bool BlockingWrite(AuUInt64 offset, const Memory::MemoryViewStreamRead &parameters) override;
void Init(const AuSPtr<FileHandle> &handle);
AuSPtr<FileHandle> GetHandle();
private:
AuSPtr<FileHandle> handle_;
};
struct NtAsyncFileTransaction : IAsyncTransaction, AuEnableSharedFromThis<NtAsyncFileTransaction>
{
~NtAsyncFileTransaction();
bool Init(const AuSPtr<FileHandle> &handle);
bool StartRead(AuUInt64 offset, const AuSPtr<AuMemoryViewWrite> &memoryView) override;
bool StartWrite(AuUInt64 offset, const AuSPtr<AuMemoryViewRead> &memoryView) override;
bool Complete() override;
AuUInt32 GetLastPacketLength() override;
void SetCallback(const AuSPtr<IAsyncFinishedSubscriber> &sub) override;
bool Wait(AuUInt32 timeout) override;
AuSPtr<Loop::ILoopSource> NewLoopSource() override;
void DispatchCb(AuUInt32 len);
HANDLE GetHandle();
AuSPtr<FileHandle> GetFileHandle();
// Required for a very evil hack
OVERLAPPED overlap_ {};
HANDLE event_ = INVALID_HANDLE_VALUE;
AuSPtr<NtAsyncFileTransaction> pin_;
AuUInt32 lastAbstractStat_ {}, lastAbstractOffset_ {};
private:
AuSPtr<void> memoryHold_;
AuSPtr<FileHandle> handle_;
bool latch_ {};
AuSPtr<IAsyncFinishedSubscriber> sub_;
};
}

341
Source/IPC/IPCPipe.NT.cpp
View File

@ -10,17 +10,346 @@
#include "IPCHandle.hpp"
#include "IPCPipe.NT.hpp"
#include <Source/Loop/LSHandle.hpp>
#include <Source/Loop/LSEvent.hpp>
#include <Source/IO/FS/Async.NT.hpp>
namespace Aurora::IPC
{
AUKN_SYM AuSPtr<IPCPipe> NewPipe()
//////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////
// Pipes
//////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////
struct IPCPipeImpl;
struct IPCHasConnectionEvent : Loop::LSEvent
{
SysPushErrorUnimplemented();
return {};
IPCHasConnectionEvent(AuSPtr<IPCPipeImpl> parent);
bool IsSignaled() override;
Loop::ELoopSource GetType() override;
bool OnTrigger(AuUInt handle) override;
private:
AuWPtr<IPCPipeImpl> parent_;
};
struct IPCPipeImpl : IPCPipe, Loop::LSHandle, public AuEnableSharedFromThis<IPCPipeImpl>
{
IPCPipeImpl(HANDLE clientHandle, HANDLE serverHandle, const IPCHandle &handle);
~IPCPipeImpl();
PROXY_INTERNAL_INTERFACE_(LSHandle::)
virtual AuSPtr<IO::FS::IAsyncTransaction> NewAsyncTransaction() override;
virtual AuSPtr<Loop::ILoopSource> AsReadChannelIsOpen() override;
virtual AuSPtr<Loop::ILoopSource> AsReadChannelHasData() override;
virtual bool Read(const Memory::MemoryViewStreamWrite &write, bool nonblocking) override;
virtual bool Write(const Memory::MemoryViewStreamRead &read) override;
virtual AuString ExportToString() override;
bool IsSignaled() override;
bool WaitOn(AuUInt32 timeout) override;
Loop::ELoopSource GetType() override;
HANDLE GetPipeHandle();
void TryConnect();
OVERLAPPED overlapped {};
private:
HANDLE serverHandle_ {INVALID_HANDLE_VALUE};
HANDLE clientHandle_ {INVALID_HANDLE_VALUE};
IPCHandle ipcHandle_;
AuSPtr<IO::FS::FileHandle> fsHandle_;
AuSPtr<Loop::ILSEvent> hasClient_;
AuSPtr<IO::FS::NtAsyncFileStream> fsStream_;
AuSPtr<IPCHasConnectionEvent> lshasConnection_;
bool bFirstTime {true};
};
IPCHasConnectionEvent::IPCHasConnectionEvent(AuSPtr<IPCPipeImpl> parent) : parent_(parent), LSEvent(false, false, true)
{
}
AUKN_SYM AuSPtr<IPCPipe> ImportPipe(const AuString &handle)
bool IPCHasConnectionEvent::IsSignaled()
{
SysPushErrorUnimplemented();
return {};
return OnTrigger(0);
}
Loop::ELoopSource IPCHasConnectionEvent::GetType()
{
return Loop::ELoopSource::eSourceIPCHasClient;
}
bool IPCHasConnectionEvent::OnTrigger(AuUInt handle)
{
auto parent = this->parent_.lock();
if (!parent)
{
SysPushErrorMem("IPC pipe is dead");
return false;
}
parent->TryConnect();
return WaitForSingleObject(parent->overlapped.hEvent, 0) == WAIT_OBJECT_0;
}
IPCPipeImpl::IPCPipeImpl(HANDLE clientHandle, HANDLE serverHandle, const IPCHandle &handle) :
serverHandle_(serverHandle), clientHandle_(clientHandle), ipcHandle_(handle)
{
if (serverHandle != INVALID_HANDLE_VALUE)
{
this->hasClient_ = Loop::NewLSEvent(false, false, true);
}
this->fsHandle_ = AuMakeShared<IO::FS::FileHandle>();
this->fsStream_ = AuMakeShared<IO::FS::NtAsyncFileStream>();
this->fsHandle_->Init(this->GetPipeHandle());
this->fsStream_->Init(this->fsHandle_);
TryConnect();
}
void IPCPipeImpl::TryConnect()
{
DWORD idc;
if (this->serverHandle_ == INVALID_HANDLE_VALUE)
{
return;
}
this->overlapped.hEvent = (HANDLE)AuStaticCast<Loop::LSEvent>(this->hasClient_)->GetHandle();
bool firstTime = AuExchange(bFirstTime, false);
if (firstTime ||
(WaitForSingleObject(this->overlapped.hEvent, 0) == WAIT_OBJECT_0 &&
GetOverlappedResult(this->serverHandle_, &this->overlapped, &idc, false))
)
{
ResetEvent(this->overlapped.hEvent);
if (ConnectNamedPipe(this->serverHandle_, &this->overlapped))
{
bFirstTime = true;
TryConnect();
}
else if (GetLastError() == ERROR_IO_PENDING)
{
// No-op
}
else if (GetLastError() == ERROR_PIPE_CONNECTED)
{
SetEvent(this->overlapped.hEvent);
}
}
else
{
if (WaitForSingleObject(this->overlapped.hEvent, 0) == WAIT_OBJECT_0)
{
ResetEvent(this->overlapped.hEvent);
}
}
}
IPCPipeImpl::~IPCPipeImpl()
{
}
AuSPtr<Loop::ILoopSource> IPCPipeImpl::AsReadChannelIsOpen()
{
if (this->serverHandle_ == INVALID_HANDLE_VALUE)
{
return {};
}
if (!this->lshasConnection_)
{
this->lshasConnection_ = AuMakeShared<IPCHasConnectionEvent>(AuSharedFromThis());
}
return AuStaticCast<Loop::ILSEvent>(this->lshasConnection_);
}
AuSPtr<Loop::ILoopSource> IPCPipeImpl::AsReadChannelHasData()
{
// TODO (Hack): we should at least make a shared timer
return AuUnsafeRaiiToShared(this);
}
AuSPtr<IO::FS::IAsyncTransaction> IPCPipeImpl::NewAsyncTransaction()
{
return this->fsStream_->NewTransaction();
}
bool IPCPipeImpl::Read(const Memory::MemoryViewStreamWrite &write, bool nonblocking)
{
DWORD size = write.length;
TryConnect();
auto h = this->GetPipeHandle();
if (h == INVALID_HANDLE_VALUE)
{
SysPushErrorUninitialized();
return false;
}
if (!write.ptr)
{
write.outVariable = size;
return true;
}
auto ret = ::ReadFile(h, write.ptr, size, &size, NULL);
write.outVariable = size;
return ret;
}
bool IPCPipeImpl::Write(const Memory::MemoryViewStreamRead &read)
{
auto h = this->GetPipeHandle();
if (h == INVALID_HANDLE_VALUE)
{
SysPushErrorUninitialized();
return false;
}
TryConnect();
DWORD temp;
if (!::WriteFile(h, read.ptr, read.length, &temp, nullptr))
{
SysPushErrorIO();
return false;
}
read.outVariable = temp;
return true;
}
HANDLE IPCPipeImpl::GetPipeHandle()
{
return this->clientHandle_ == INVALID_HANDLE_VALUE ? this->serverHandle_ : this->clientHandle_;
}
bool IPCPipeImpl::IsSignaled()
{
DWORD avail {};
TryConnect();
if (!PeekNamedPipe(this->GetPipeHandle(), NULL, NULL, NULL, &avail, NULL))
{
return false;
}
return avail;
}
bool IPCPipeImpl::WaitOn(AuUInt32 timeout)
{
return LSHandle::WaitOn(timeout);
}
Loop::ELoopSource IPCPipeImpl::GetType()
{
return Loop::ELoopSource::eSourceIPCReadPipe;
}
AuString IPCPipeImpl::ExportToString()
{
TryConnect();
return this->clientHandle_ == INVALID_HANDLE_VALUE ?
this->ipcHandle_.ToString() :
AuString {};
}
AUKN_SYM AuSPtr<IPCPipe> NewPipe()
{
IPCHandle handle;
handle.NewId();
auto name = "\\\\.\\pipe\\" + handle.ToNTPath();
auto maxLength = 16 * AuHwInfo::GetPageSize() ? AuHwInfo::GetPageSize() : 4096;
auto pipeServer = CreateNamedPipeA(name.c_str(),
PIPE_ACCESS_DUPLEX | FILE_FLAG_FIRST_PIPE_INSTANCE | FILE_FLAG_WRITE_THROUGH | FILE_FLAG_OVERLAPPED,
PIPE_WAIT,
1,
maxLength,
maxLength,
NMPWAIT_WAIT_FOREVER,
nullptr);
if (pipeServer == INVALID_HANDLE_VALUE)
{
SysPushErrorIO("{}", GetLastError());
return {};
}
auto object = AuMakeShared<IPCPipeImpl>(INVALID_HANDLE_VALUE, pipeServer, handle);
if (!object)
{
SysPushErrorMem();
AuWin32CloseHandle(pipeServer);
return {};
}
return object;
}
AUKN_SYM AuSPtr<IPCPipe> ImportPipe(const AuString &handleString)
{
IPCHandle handle;
HANDLE pipe;
if (!handle.FromString(handleString))
{
SysPushErrorParseError();
return {};
}
auto name = "\\\\.\\pipe\\" + handle.ToNTPath();
pipe = CreateFileA(name.c_str(),
GENERIC_WRITE | GENERIC_READ,
0,
NULL,
OPEN_ALWAYS,
FILE_ATTRIBUTE_NORMAL | FILE_FLAG_OVERLAPPED,
NULL);
if (pipe == INVALID_HANDLE_VALUE)
{
if (GetLastError() == ERROR_PIPE_BUSY)
{
SysPushErrorIO("Pipe is used -> a client has already connected or the nt server is not ready");
return {};
}
SysPushErrorIO("{}", GetLastError());
return {};
}
auto object = AuMakeShared<IPCPipeImpl>(pipe, INVALID_HANDLE_VALUE, handle);
if (!object)
{
SysPushErrorMem();
AuWin32CloseHandle(pipe);
return {};
}
return object;
}
}
// > The pipe created by UWP process with name \\.\pipe\Local\PipeName is converted to \\.\pipe\Sessions\<SessionId>\AppContainerNamedObjects\<AppContainerSid>\PipeName.
// > I can use this to communicate between UWP as server and Win32 as client.
// https://jike.in/qa/?qa=103904/
// ...good to know

2
Source/Loop/LoopQueue.NT.cpp
View File

@ -28,7 +28,7 @@ namespace Aurora::Loop
{
//////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////
// Loopq Queue :: Extended Handle
// Loop Queue :: Extended Handle
//////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////
LoopQueue::ExtendedSourceInfo::ExtendedSourceInfo(const AuSPtr<ILoopSourceEx> &in) : source(in), timeoutAbs(0)