AuroraRuntime/Source/IO/FS/Async.Linux.cpp

561 lines
14 KiB
C++
Raw Normal View History

2022-04-13 11:00:35 +00:00
/***
Copyright (C) 2022 J Reece Wilson (a/k/a "Reece"). All rights reserved.
File: Async.Linux.cpp
Date: 2022-4-12
Author: Reece
***/
#include <Source/RuntimeInternal.hpp>
#include "FS.hpp"
#include "FileAdvisory.Unix.hpp"
#include <Source/IO/Loop/Loop.hpp>
#include <Source/IO/Loop/LSHandle.hpp>
#include <Source/IO/Loop/LSEvent.hpp>
2022-04-13 11:00:35 +00:00
#include <Source/IO/UNIX/IOSubmit.Linux.hpp>
#include <unistd.h>
#include <fcntl.h>
#include "FileStream.Unix.hpp"
#include "Async.Linux.hpp"
#include <Source/IO/IPC/AuIPCPipe.Unix.hpp>
#include <Source/IO/AuIOHandle.hpp>
2022-04-13 11:00:35 +00:00
namespace Aurora::IO::FS
{
#define IPC_PIPE AuStaticCast<AFileHandle>(this->pHandle_)->pIPCPipe
2022-06-12 13:53:35 +00:00
struct LinuxAsyncFileTransactionLoopSource : Aurora::IO::Loop::LSEvent
{
LinuxAsyncFileTransactionLoopSource(AuSPtr<LinuxAsyncFileTransaction> that);
virtual bool IsSignaled() override;
virtual bool OnTrigger(AuUInt handle) override;
virtual AuLoop::ELoopSource GetType() override;
virtual const AuList<AuUInt> &GetHandles() override;
virtual bool Singular() override;
2022-06-12 13:53:35 +00:00
private:
bool bExMode {};
2022-06-12 13:53:35 +00:00
AuWPtr<LinuxAsyncFileTransaction> caller_;
AuList<AuUInt> handles_;
2022-06-12 13:53:35 +00:00
};
LinuxAsyncFileTransactionLoopSource::LinuxAsyncFileTransactionLoopSource(AuSPtr<LinuxAsyncFileTransaction> that) :
caller_(that),
Loop::LSEvent(false, false, true)
2022-04-13 11:00:35 +00:00
{
if (that)
{
if (auto pPipe = AuStaticCast<AFileHandle>(that->GetFileHandle())->pIPCPipe)
{
this->bExMode = true;
this->handles_ = {pPipe->GetPreemptFd(), Loop::LSEvent::GetHandle()};
}
}
}
const AuList<AuUInt> &LinuxAsyncFileTransactionLoopSource::GetHandles()
{
return this->handles_;
}
bool LinuxAsyncFileTransactionLoopSource::Singular()
{
return !this->bExMode;
2022-04-13 11:00:35 +00:00
}
bool LinuxAsyncFileTransactionLoopSource::OnTrigger(AuUInt handle)
{
auto lock = caller_.lock();
if (lock)
{
return lock->Complete();
}
return true;
2022-04-13 11:00:35 +00:00
}
bool LinuxAsyncFileTransactionLoopSource::IsSignaled()
{
if (LSEvent::IsSignaled())
{
auto lock = caller_.lock();
return lock->Complete();
}
return false;
2022-04-13 11:00:35 +00:00
}
Loop::ELoopSource LinuxAsyncFileTransactionLoopSource::GetType()
{
return Loop::ELoopSource::eSourceAIO;
}
LinuxAsyncFileTransaction::LinuxAsyncFileTransaction(AuSPtr<ProcessBlock> pProcessBlock) :
pProcessBlock_(pProcessBlock)
{
}
2022-04-13 11:00:35 +00:00
LinuxAsyncFileTransaction::~LinuxAsyncFileTransaction()
{
if (this->pProcessBlock_)
{
AU_LOCK_GUARD(this->pProcessBlock_->externalLock);
AuTryRemove(this->pProcessBlock_->submits, this);
}
2022-04-13 11:00:35 +00:00
}
void LinuxAsyncFileStream::Init(const AuSPtr<IIOHandle> &handle)
2022-04-13 11:00:35 +00:00
{
this->pHandle_ = handle;
2022-04-13 11:00:35 +00:00
}
void LinuxAsyncFileStream::MakeProcess(Processes::IProcess *pProcess)
{
this->pProcessBlock_ = AuMakeSharedThrow<ProcessBlock>();
this->pProcessBlock_->pProcess = pProcess;
}
void LinuxAsyncFileStream::CheckProcess()
{
if (this->pProcessBlock_)
{
AU_LOCK_GUARD(this->pProcessBlock_->externalLock);
if (this->pProcessBlock_->pProcess &&
this->pProcessBlock_->pProcess->HasExited())
{
this->pProcessBlock_->pProcess = nullptr;
this->pProcessBlock_->bDead = true;
for (auto &pSubmittable : this->pProcessBlock_->submits)
{
pSubmittable->LIOS_Cancel();
}
}
}
}
2022-04-13 11:00:35 +00:00
AuSPtr<IAsyncTransaction> LinuxAsyncFileStream::NewTransaction()
{
auto shared = AuMakeShared<LinuxAsyncFileTransaction>(this->pProcessBlock_);
2022-04-13 11:00:35 +00:00
if (!shared)
{
return {};
}
if (!shared->Init(this->pHandle_))
2022-04-13 11:00:35 +00:00
{
return {};
}
if (this->pProcessBlock_)
{
AU_LOCK_GUARD(this->pProcessBlock_->externalLock);
if (!AuTryInsert(this->pProcessBlock_->submits, shared.get()))
{
SysPushErrorMemory();
return {};
}
}
2022-04-13 11:00:35 +00:00
return shared;
}
bool LinuxAsyncFileStream::BlockingTruncate(AuUInt64 length)
{
auto iOptSafe = this->pHandle_->GetOSWriteHandleSafe();
if (!iOptSafe)
{
return false;
}
auto fd = (int)iOptSafe.Value();
if (fd == -1)
{
SysPushErrorUninitialized();
return false;
}
return ::ftruncate(fd, length) != -1;
}
bool LinuxAsyncFileStream::BlockingRead(AuUInt64 offset, const Memory::MemoryViewStreamWrite &parameters)
{
auto iOptSafe = this->pHandle_->GetOSReadHandleSafe();
if (!iOptSafe)
{
return false;
}
auto fd = (int)iOptSafe.Value();
if (fd == -1)
{
SysPushErrorUninitialized();
return false;
}
if (IPC_PIPE)
{
if (IPC_PIPE->LIOS_PopOne())
{
parameters.outVariable = 0;
return true;
}
}
if (this->pProcessBlock_ &&
this->pProcessBlock_->bDead)
{
parameters.outVariable = 0;
return true;
}
if (!PosixSetOffset(fd, offset))
{
return false;
}
AuUInt32 read;
if (!PosixRead(fd, parameters.ptr, parameters.length, &read))
{
return false;
}
return true;
}
bool LinuxAsyncFileStream::BlockingWrite(AuUInt64 offset, const Memory::MemoryViewStreamRead &parameters)
{
auto iOptSafe = this->pHandle_->GetOSWriteHandleSafe();
if (!iOptSafe)
{
return false;
}
auto fd = (int)iOptSafe.Value();
if (fd == -1)
{
SysPushErrorUninitialized();
return false;
}
if (!PosixSetOffset(fd, offset))
{
return false;
}
if (this->pProcessBlock_ &&
this->pProcessBlock_->bDead)
{
return false;
}
AuUInt32 read;
if (!PosixWrite(fd, parameters.ptr, parameters.length, &read))
{
return false;
}
return true;
}
bool LinuxAsyncFileTransaction::Init(const AuSPtr<IIOHandle> &handle)
2022-04-13 11:00:35 +00:00
{
if (!handle)
{
return false;
}
this->pHandle_ = handle;
2022-04-13 11:00:35 +00:00
this->loopSource_ = AuMakeShared<LinuxAsyncFileTransactionLoopSource>(AuSharedFromThis());
return bool(this->loopSource_);
}
2023-08-08 23:02:35 +00:00
void LinuxAsyncFileTransaction::SetBaseOffset(AuUInt64 uBaseOffset)
{
this->uBaseOffset = uBaseOffset;
}
2022-04-13 11:00:35 +00:00
bool LinuxAsyncFileTransaction::StartRead(AuUInt64 offset, const AuSPtr<AuMemoryViewWrite> &memoryView)
{
if (HasState())
{
SysPushErrorIO("IO Operation can not be reused yet.");
return false;
}
auto iOptSafe = this->pHandle_->GetOSReadHandleSafe();
if (!iOptSafe)
2022-04-13 11:00:35 +00:00
{
return false;
}
auto fd = (int)iOptSafe.Value();
if (fd == -1)
{
SysPushErrorUninitialized();
return false;
}
2022-04-13 11:00:35 +00:00
this->latch_ = false;
2022-06-12 13:53:35 +00:00
this->hasError_ = false;
this->bTxFinished_ = false;
this->lastFinishedStat_ = 0;
if (!this->loopSource_)
{
SysPushErrorUninitialized();
return false;
}
this->loopSource_->Reset();
2022-04-13 11:00:35 +00:00
this->lastAbstractOffset_ = offset;
LIOS_Init(AuSharedFromThis());
SetMemory(memoryView);
if (IPC_PIPE)
{
if (IPC_PIPE->LIOS_PopOne())
{
LIOS_SendProcess(0, false, errno);
return true;
}
}
2023-08-08 23:02:35 +00:00
if (this->pProcessBlock_ &&
this->pProcessBlock_->bDead)
{
LIOS_SendProcess(0, false, errno);
return true;
}
2023-08-08 23:02:35 +00:00
offset += this->uBaseOffset;
if (!UNIX::LinuxOverlappedSubmitRead(fd, offset, this, this->loopSource_.get(), bool(IPC_PIPE)))
2022-04-13 11:00:35 +00:00
{
LIOS_SendProcess(0, true, errno);
return true;
2022-04-13 11:00:35 +00:00
}
else
{
if (gRuntimeConfig.linuxConfig.bFIODisableBatching)
2022-04-13 11:00:35 +00:00
{
UNIX::SendIOBuffers();
}
return true;
}
}
bool LinuxAsyncFileTransaction::StartWrite(AuUInt64 offset, const AuSPtr<AuMemoryViewRead> &memoryView)
{
if (HasState())
{
SysPushErrorIO("IO Operation can not be reused yet.");
return false;
}
auto iOptSafe = this->pHandle_->GetOSWriteHandleSafe();
if (!iOptSafe)
{
return false;
}
auto fd = (int)iOptSafe.Value();
2022-04-13 11:00:35 +00:00
if (fd == -1)
{
SysPushErrorUninitialized();
return false;
}
this->latch_ = false;
this->bTxFinished_ = false;
2022-06-12 13:53:35 +00:00
this->hasError_ = false;
this->lastFinishedStat_ = 0;
if (!this->loopSource_)
{
SysPushErrorUninitialized();
return false;
}
2022-04-13 11:00:35 +00:00
this->loopSource_->Reset();
2022-04-13 11:00:35 +00:00
this->lastAbstractOffset_ = offset;
LIOS_Init(AuSharedFromThis());
SetMemory(memoryView);
if (this->pProcessBlock_ &&
this->pProcessBlock_->bDead)
{
LIOS_SendProcess(0, false, errno);
return true;
}
2023-08-08 23:02:35 +00:00
offset += this->uBaseOffset;
2022-04-13 11:00:35 +00:00
if (!UNIX::LinuxOverlappedSubmitWrite(fd, offset, this, this->loopSource_.get()))
{
LIOS_SendProcess(0, true, errno);
return false;
}
else
{
if (gRuntimeConfig.linuxConfig.bFIODisableBatching)
2022-04-13 11:00:35 +00:00
{
UNIX::SendIOBuffers();
}
return true;
}
}
void LinuxAsyncFileTransaction::Reset()
{
(void)this->LIOS_Cancel();
if (this->loopSource_)
{
this->loopSource_->Reset();
}
}
2022-04-13 11:00:35 +00:00
void LinuxAsyncFileTransaction::LIOS_Process(AuUInt32 read, bool failure, int err, bool mark)
{
this->lastFinishedStat_ = failure ? 0 : read;
2022-06-12 13:53:35 +00:00
this->hasError_ = failure;
this->error_ = err;
2022-04-13 11:00:35 +00:00
this->bTxFinished_ = true;
if (mark)
{
return;
}
this->DispatchCb();
if (read)
{
if (IPC_PIPE)
{
// Return value intentionally ignored
// We just need to poke on read...
IPC_PIPE->LIOS_PopOne();
}
}
2022-04-13 11:00:35 +00:00
}
void LinuxAsyncFileTransaction::DispatchCb()
{
if (AuExchange(this->latch_, true))
{
// TODO (Reece): urgent
//SysPushErrorGeneric();
2022-04-13 11:00:35 +00:00
return;
}
if (this->sub_)
{
this->sub_->OnAsyncFileOpFinished(this->lastAbstractOffset_, this->lastFinishedStat_);
}
}
bool LinuxAsyncFileTransaction::Complete()
{
if (this->bTxFinished_)
{
if (!this->latch_)
{
LIOS_SendProcess(this->lastFinishedStat_, this->lastFinishedStat_ == 0, 0, false);
//DispatchCb();
2022-04-13 11:00:35 +00:00
}
2022-04-13 11:00:35 +00:00
return true;
}
return false;
}
2023-08-08 23:14:36 +00:00
bool LinuxAsyncFileTransaction::HasCompleted()
{
return this->bTxFinished_;
}
2023-12-28 16:49:11 +00:00
bool LinuxAsyncFileTransaction::HasFailed()
2022-06-12 13:53:35 +00:00
{
return this->hasError_;
}
AuUInt LinuxAsyncFileTransaction::GetOSErrorCode()
{
return AuUInt(this->error_);
}
2022-04-13 11:00:35 +00:00
AuUInt32 LinuxAsyncFileTransaction::GetLastPacketLength()
{
return this->lastFinishedStat_;
}
void LinuxAsyncFileTransaction::SetCallback(const AuSPtr<IAsyncFinishedSubscriber> &sub)
{
this->sub_ = sub;
}
bool LinuxAsyncFileTransaction::Wait(AuUInt32 timeout)
{
// TODO:
AuList<AuSPtr<IAsyncTransaction>> files {AuUnsafeRaiiToShared(this)};
return WaitMultiple(files, timeout);
}
AuSPtr<IIOHandle> LinuxAsyncFileTransaction::GetFileHandle()
2022-04-13 11:00:35 +00:00
{
return this->pHandle_;
2022-04-13 11:00:35 +00:00
}
AuSPtr<Loop::ILoopSource> LinuxAsyncFileTransaction::NewLoopSource()
{
return AuStaticCast<Loop::ILoopSource>(AuStaticCast<Loop::ILSEvent>(this->loopSource_));
}
AUKN_SYM IAsyncFileStream *OpenAsyncNew(const AuString &path, EFileOpenMode openMode, bool bDirectIO, EFileAdvisoryLockLevel lock)
2022-04-13 11:00:35 +00:00
{
auto pHandle = AuIO::IOHandleShared();
if (!pHandle)
2022-04-13 11:00:35 +00:00
{
SysPushErrorMemory();
return nullptr;
2022-04-13 11:00:35 +00:00
}
AuIO::IIOHandle::HandleCreate createhandle(path);
createhandle.eAdvisoryLevel = lock;
createhandle.eMode = openMode;
createhandle.bFailIfNonEmptyFile = false;
createhandle.bDirectIOMode = bDirectIO;
createhandle.bAsyncHandle = true;
2022-04-13 11:00:35 +00:00
if (!pHandle->InitFromPath(createhandle))
2022-04-13 11:00:35 +00:00
{
SysPushErrorNested();
return nullptr;
2022-04-13 11:00:35 +00:00
}
auto pStream = _new LinuxAsyncFileStream();
if (!pStream)
2022-04-13 11:00:35 +00:00
{
SysPushErrorMemory();
return nullptr;
2022-04-13 11:00:35 +00:00
}
pStream->Init(pHandle);
return pStream;
2022-04-13 11:00:35 +00:00
}
AUKN_SYM void OpenAsyncRelease(IAsyncFileStream *handle)
{
AuSafeDelete<LinuxAsyncFileStream *>(handle);
}
}