J Reece Wilson
fd0c5b51b2
[+] Begin work on IO futexes for io release on process/thread exit [+] Linux ::readdir iteration [+] AuConsole buffering API [*] Fix sleep as to not get interrupted by signals [*] Switch the type of FS lock used under Linux [*] Linux: Use new IPCHandle encoding scheme [*] Fix undefined behaviour: unintialized timeout values (AuLoop/Linux) [*] Fix undefined behaviour: ConsoleTTY clear line was called of a color of a random value on stack [-] Remainings of std dir iterator [*] Fix pthread_kill (aka send signal to pthread handle) always kills process. This is what you expect bc signal handler inheritance. [*] Reformat the build Aurora.json file [+] Added clang warning ignores to the build file [*] Fix: UNIX need to use STDOUT_FILENO. Was using CRT handle in place of fd by mistake. [+] Linux implementation for IO yield (AuIO::IOYield() - UNIX::LinuxOverlappedYield()) [*] Fix: Linux async end of stream processing. res 0 = zero bytes consumed. <= was detecting this as an error of code 0. Should succeed with zero bytes. [+] Linux LoopQueue missing epilogue hook for the IO processor [*] Various refactors and minor bug fixes [*] Linux fix: Handle pipe EOS as zero [*] Linux fix: thread termination via a user signal of 77. Need a force terminate. [*] IPC handle: fix improper int to bool cast in the header setup within ToString [*] Linux fix: HWInfo CPU topology regression [-] Linux fix: remove SIGABRT handler [*] Missing override in compression, exit, and consoletty headers. [+] Unix Syslog logger backend
451 lines
11 KiB
C++
451 lines
11 KiB
C++
/***
|
|
Copyright (C) 2022 J Reece Wilson (a/k/a "Reece"). All rights reserved.
|
|
|
|
File: IPCPipe.Linux.cpp
|
|
Date: 2022-4-14
|
|
Author: Reece
|
|
***/
|
|
#include <Source/RuntimeInternal.hpp>
|
|
#include "IPC.hpp"
|
|
#include "IPCHandle.hpp"
|
|
#include "IPCPipe.Unix.hpp"
|
|
|
|
#include <Source/IO/UNIX/FDIpcServer.hpp>
|
|
|
|
#include <Source/IO/Loop/ILoopSourceEx.hpp>
|
|
#include <Source/IO/Loop/LSHandle.hpp>
|
|
#include <Source/IO/Loop/LSEvent.hpp>
|
|
|
|
#if defined(AURORA_IS_LINUX_DERIVED)
|
|
#include <Source/IO/UNIX/IOSubmit.Linux.hpp>
|
|
#include <Source/IO/FS/Async.Linux.hpp>
|
|
#include "IPCPrimitives.Linux.hpp"
|
|
#include "IPCMutexFutex.Linux.hpp"
|
|
#endif
|
|
|
|
#include "IPCMemory.Unix.hpp" // required for handle
|
|
|
|
#include <fcntl.h>
|
|
|
|
namespace Aurora::IO::IPC
|
|
{
|
|
//////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////
|
|
// Pipes
|
|
//////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////
|
|
|
|
struct IPCPipeImpl : IPCPipe, Loop::LSHandle, AuEnableSharedFromThis<IPCPipeImpl>
|
|
{
|
|
IPCPipeImpl(int (fds)[2], int (secondary)[2], IPCToken readEnd, IPCToken writeEnd, AuSPtr<IPCEvent> event, AuSPtr<IPCMutex> mutex);
|
|
~IPCPipeImpl();
|
|
|
|
PROXY_INTERNAL_INTERFACE_(LSHandle::)
|
|
|
|
virtual AuSPtr<IO::IAsyncTransaction> NewAsyncTransaction() override;
|
|
|
|
virtual AuSPtr<Loop::ILoopSource> AsReadChannelIsOpen() override;
|
|
virtual AuSPtr<Loop::ILoopSource> AsReadChannelHasData() override;
|
|
|
|
virtual bool Read(const Memory::MemoryViewStreamWrite &write, bool nonblock) 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;
|
|
|
|
private:
|
|
int fds[2] {-1, -1};
|
|
int secondary[2] {-1, -1};
|
|
|
|
AuSPtr<IO::FS::FileHandle> fsHandle_;
|
|
AuSPtr<IO::FS::LinuxAsyncFileStream> fsStream_;
|
|
|
|
IPCToken readEnd_;
|
|
IPCToken writeEnd_;
|
|
AuSPtr<IPCEvent> event_;
|
|
AuSPtr<IPCMutex> mutex_;
|
|
};
|
|
|
|
IPCPipeImpl::IPCPipeImpl(int (fds2)[2], int (fds3)[2], IPCToken readEnd, IPCToken writeEnd, AuSPtr<IPCEvent> event, AuSPtr<IPCMutex> mutex) :
|
|
fds {fds2[0], fds2[1]}, secondary {fds3[0], fds3[1]},
|
|
readEnd_(readEnd), writeEnd_(writeEnd), event_(event), mutex_(mutex)
|
|
{
|
|
this->handle = fds[0];
|
|
|
|
this->fsHandle_ = AuMakeShared<IO::FS::FileHandle>();
|
|
this->fsStream_ = AuMakeShared<IO::FS::LinuxAsyncFileStream>();
|
|
|
|
this->fsHandle_->Init(fds2[0], fds2[1]);
|
|
this->fsStream_->Init(this->fsHandle_);
|
|
}
|
|
|
|
IPCPipeImpl::~IPCPipeImpl()
|
|
{
|
|
int fd {-1};
|
|
|
|
if ((fd = AuExchange(fds[0], -1)) != -1)
|
|
{
|
|
IO::UNIX::FDServeEnd(readEnd_);
|
|
//::close(fd);
|
|
}
|
|
|
|
if ((fd = AuExchange(fds[1], -1)) != -1)
|
|
{
|
|
IO::UNIX::FDServeEnd(writeEnd_);
|
|
//::close(fd);
|
|
}
|
|
|
|
if ((fd = AuExchange(secondary[0], -1)) != -1)
|
|
{
|
|
::close(fd);
|
|
}
|
|
|
|
if ((fd = AuExchange(secondary[1], -1)) != -1)
|
|
{
|
|
::close(fd);
|
|
}
|
|
|
|
fsHandle_.reset();
|
|
fsStream_.reset();
|
|
|
|
if (secondary[0] == -1)
|
|
{
|
|
event_->Reset();
|
|
}
|
|
}
|
|
|
|
AuSPtr<Loop::ILoopSource> IPCPipeImpl::AsReadChannelIsOpen()
|
|
{
|
|
return this->event_;
|
|
}
|
|
|
|
AuSPtr<Loop::ILoopSource> IPCPipeImpl::AsReadChannelHasData()
|
|
{
|
|
return AuSharedFromThis();
|
|
}
|
|
|
|
AuSPtr<IO::IAsyncTransaction> IPCPipeImpl::NewAsyncTransaction()
|
|
{
|
|
return this->fsStream_->NewTransaction();
|
|
}
|
|
|
|
bool IPCPipeImpl::Read(const Memory::MemoryViewStreamWrite &write, bool nonblock)
|
|
{
|
|
auto handle = fds[0];
|
|
auto control = ::fcntl(handle, F_GETFL);
|
|
auto ref = control;
|
|
|
|
if (nonblock)
|
|
{
|
|
control |= O_NONBLOCK;
|
|
}
|
|
else
|
|
{
|
|
control &= ~O_NONBLOCK;
|
|
}
|
|
|
|
if (ref != control)
|
|
{
|
|
::fcntl(handle, F_SETFL, control);
|
|
}
|
|
|
|
if (!write.ptr)
|
|
{
|
|
#if defined(FIONREAD)
|
|
int available {};
|
|
if (::ioctl(ref, FIONREAD, &available) < 0)
|
|
{
|
|
return false;
|
|
}
|
|
|
|
write.outVariable = available;
|
|
return true;
|
|
#else
|
|
SysPushErrorHAL("Cannot pull bytes available in the IPC on this UNIX platform");
|
|
return false;
|
|
#endif
|
|
}
|
|
|
|
int tmp;
|
|
do
|
|
{
|
|
tmp = ::read(handle, write.ptr, write.length);
|
|
} while ((tmp == -1 && errno == EINTR));
|
|
|
|
if (tmp <= 0)
|
|
{
|
|
if (tmp == 0)
|
|
{
|
|
return nonblock;
|
|
}
|
|
|
|
SysPushErrorMem();
|
|
return false;
|
|
}
|
|
|
|
write.outVariable = tmp;
|
|
return true;
|
|
}
|
|
|
|
bool IPCPipeImpl::Write(const Memory::MemoryViewStreamRead &read)//, bool nonblock)
|
|
{
|
|
bool nonblock = true;
|
|
auto handle = this->fds[1];
|
|
auto control = ::fcntl(handle, F_GETFL);
|
|
auto ref = control;
|
|
|
|
if (nonblock)
|
|
{
|
|
control |= O_NONBLOCK;
|
|
}
|
|
else
|
|
{
|
|
control &= ~O_NONBLOCK;
|
|
}
|
|
|
|
if (ref != control)
|
|
{
|
|
::fcntl(handle, F_SETFL, control);
|
|
}
|
|
|
|
int tmp;
|
|
do
|
|
{
|
|
tmp = ::write(handle, read.ptr, read.length);
|
|
} while ((tmp == -1 && errno == EINTR));
|
|
|
|
if (tmp <= 0)
|
|
{
|
|
if (tmp == 0)
|
|
{
|
|
return nonblock;
|
|
}
|
|
|
|
SysPushErrorMem();
|
|
return false;
|
|
}
|
|
|
|
read.outVariable = tmp;
|
|
return true;
|
|
}
|
|
|
|
bool IPCPipeImpl::IsSignaled()
|
|
{
|
|
return LSHandle::IsSignaled();
|
|
}
|
|
|
|
bool IPCPipeImpl::WaitOn(AuUInt32 timeout)
|
|
{
|
|
return LSHandle::WaitOn(timeout);
|
|
}
|
|
|
|
Loop::ELoopSource IPCPipeImpl::GetType()
|
|
{
|
|
return Loop::ELoopSource::eSourceIPCReadPipe;
|
|
}
|
|
|
|
AuString IPCPipeImpl::ExportToString()
|
|
{
|
|
IPC::IPCHandle handle;
|
|
|
|
handle.PushId(EIPCHandleType::eIPCPipe, AuStaticCast<IPCEventProxy>(this->event_)->handle_.values[0].token);
|
|
#if defined(AURORA_IS_LINUX_DERIVED)
|
|
auto that = AuStaticCast<IPCMutexProxy>(this->mutex_);
|
|
handle.PushId(EIPCHandleType::eIPCPrimitiveMutex, that->token_);
|
|
SysAssert(that->mem_);
|
|
handle.PushId(EIPCHandleType::eIPCMemory, AuStaticCast<IPCSharedMemoryImpl>(that->mem_)->handle_.values[0].token);
|
|
handle.values[1].token.word = that->index_;
|
|
#else
|
|
handle.PushId(EIPCHandleType::eIPCPrimitiveMutex, ->handle_.values[0].token);
|
|
#endif
|
|
handle.PushId(EIPCHandleType::eIPCPipeEnd, this->readEnd_);
|
|
handle.PushId(EIPCHandleType::eIPCPipeEnd, this->writeEnd_);
|
|
|
|
return handle.ToString();
|
|
}
|
|
|
|
AUKN_SYM AuSPtr<IPCPipe> NewPipe()
|
|
{
|
|
IPCToken readEnd, writeEnd;
|
|
int fds1[2];
|
|
int fds2[2];
|
|
|
|
auto event = NewEvent(false, false);
|
|
if (!event)
|
|
{
|
|
SysPushErrorNested();
|
|
return {};
|
|
}
|
|
|
|
auto mutex = NewMutex();
|
|
if (!mutex)
|
|
{
|
|
SysPushErrorNested();
|
|
return {};
|
|
}
|
|
|
|
if (::pipe(fds1) == -1)
|
|
{
|
|
SysPushErrorIO();
|
|
return {};
|
|
}
|
|
|
|
if (::pipe(fds2) == -1)
|
|
{
|
|
SysPushErrorIO();
|
|
::close(fds1[0]);
|
|
::close(fds1[1]);
|
|
return {};
|
|
}
|
|
|
|
if (!IO::UNIX::FDServe(fds1[0], readEnd))
|
|
{
|
|
SysPushErrorIO();
|
|
::close(fds1[0]);
|
|
::close(fds1[1]);
|
|
::close(fds2[0]);
|
|
::close(fds2[1]);
|
|
return {};
|
|
}
|
|
|
|
if (!IO::UNIX::FDServe(fds2[1], writeEnd))
|
|
{
|
|
SysPushErrorIO();
|
|
IO::UNIX::FDServeEnd(readEnd);
|
|
::close(fds1[0]);
|
|
::close(fds1[1]);
|
|
::close(fds2[0]);
|
|
::close(fds2[1]);
|
|
return {};
|
|
}
|
|
|
|
int serverPair[2] {fds2[0], fds1[1]};
|
|
int clientPair[2] {fds1[0], fds2[1]};
|
|
|
|
auto handle = AuMakeShared<IPCPipeImpl>(serverPair, clientPair, readEnd, writeEnd, event, mutex);
|
|
if (!handle)
|
|
{
|
|
SysPushErrorMem();
|
|
IO::UNIX::FDServeEnd(readEnd);
|
|
IO::UNIX::FDServeEnd(writeEnd);
|
|
::close(fds1[0]);
|
|
::close(fds1[1]);
|
|
::close(fds2[0]);
|
|
::close(fds2[1]);
|
|
return {};
|
|
}
|
|
|
|
return handle;
|
|
}
|
|
|
|
AUKN_SYM AuSPtr<IPCPipe> ImportPipe(const AuString &handleString)
|
|
{
|
|
int fds[2] {-1, -1};
|
|
|
|
IPCHandle handle;
|
|
|
|
if (!handle.FromString(handleString))
|
|
{
|
|
SysPushErrorParseError();
|
|
return {};
|
|
}
|
|
|
|
auto eventVal = handle.GetToken(IPC::EIPCHandleType::eIPCPipe, 0);
|
|
if (!eventVal)
|
|
{
|
|
SysPushErrorParseError();
|
|
return {};
|
|
}
|
|
|
|
auto mutexVal = handle.GetToken(IPC::EIPCHandleType::eIPCPrimitiveMutex, 1);
|
|
if (!mutexVal)
|
|
{
|
|
SysPushErrorParseError();
|
|
return {};
|
|
}
|
|
|
|
int offset {};
|
|
|
|
#if defined(AURORA_IS_LINUX_DERIVED)
|
|
offset = 1;
|
|
|
|
auto ipcMem = handle.GetToken(IPC::EIPCHandleType::eIPCMemory, 2);
|
|
if (!ipcMem)
|
|
{
|
|
SysPushErrorParseError();
|
|
return {};
|
|
}
|
|
#endif
|
|
|
|
auto readVal = handle.GetToken(IPC::EIPCHandleType::eIPCPipeEnd, 2 + offset);
|
|
if (!readVal)
|
|
{
|
|
SysPushErrorParseError();
|
|
return {};
|
|
}
|
|
|
|
auto writeVal = handle.GetToken(IPC::EIPCHandleType::eIPCPipeEnd, 3 + offset);
|
|
if (!writeVal)
|
|
{
|
|
SysPushErrorParseError();
|
|
return {};
|
|
}
|
|
|
|
auto event = ImportEventEx(eventVal->token);
|
|
if (!event)
|
|
{
|
|
return {};
|
|
}
|
|
|
|
auto mutex = ImportMutexEx(mutexVal->token, ipcMem->token, mutexVal->token.word);
|
|
if (!mutex)
|
|
{
|
|
return {};
|
|
}
|
|
|
|
if (!IO::UNIX::FDAccept(readVal->token, fds[0]))
|
|
{
|
|
SysPushErrorNested();
|
|
return {};
|
|
}
|
|
|
|
if (!IO::UNIX::FDAccept(writeVal->token, fds[1]))
|
|
{
|
|
::close(fds[0]);
|
|
SysPushErrorNested();
|
|
return {};
|
|
}
|
|
|
|
int dummy[2] {-1, -1};
|
|
auto object = AuMakeShared<IPCPipeImpl>(fds,
|
|
dummy,
|
|
readVal->token,
|
|
writeVal->token,
|
|
event,
|
|
mutex);
|
|
if (!object)
|
|
{
|
|
SysPushErrorMem();
|
|
::close(fds[0]);
|
|
::close(fds[1]);
|
|
return {};
|
|
}
|
|
|
|
mutex->WaitOn();
|
|
|
|
if (event->IsSignaled())
|
|
{
|
|
mutex->Unlock();
|
|
SysPushErrorIO("Pipe Busy");
|
|
::close(fds[0]);
|
|
::close(fds[1]);
|
|
return {};
|
|
}
|
|
|
|
event->Set();
|
|
|
|
mutex->Unlock();
|
|
|
|
return object;
|
|
}
|
|
} |