AuroraRuntime/Source/IO/FS/Async.Linux.cpp
2022-08-29 00:39:12 +01:00

510 lines
12 KiB
C++

/***
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>
#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/IPCPipe.Unix.hpp>
namespace Aurora::IO::FS
{
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;
private:
bool bExMode {};
AuWPtr<LinuxAsyncFileTransaction> caller_;
AuList<AuUInt> handles_;
};
LinuxAsyncFileTransactionLoopSource::LinuxAsyncFileTransactionLoopSource(AuSPtr<LinuxAsyncFileTransaction> that) : caller_(that), Loop::LSEvent(false, false, true)
{
if (that)
{
auto possiblePipe = that->GetFileHandle()->pIPCPipe;
if (possiblePipe)
{
this->bExMode = true;
this->handles_ = {possiblePipe->GetPreemptFd(), Loop::LSEvent::GetHandle()};
}
}
}
const AuList<AuUInt> &LinuxAsyncFileTransactionLoopSource::GetHandles()
{
return this->handles_;
}
bool LinuxAsyncFileTransactionLoopSource::Singular()
{
return !this->bExMode;
}
bool LinuxAsyncFileTransactionLoopSource::OnTrigger(AuUInt handle)
{
auto lock = caller_.lock();
if (lock)
{
return lock->Complete();
}
return true;
}
bool LinuxAsyncFileTransactionLoopSource::IsSignaled()
{
if (LSEvent::IsSignaled())
{
auto lock = caller_.lock();
return lock->Complete();
}
return false;
}
Loop::ELoopSource LinuxAsyncFileTransactionLoopSource::GetType()
{
return Loop::ELoopSource::eSourceAIO;
}
LinuxAsyncFileTransaction::~LinuxAsyncFileTransaction()
{
}
FileHandle::~FileHandle()
{
if ((this->readHandle != 0) &&
(this->readHandle != -1))
{
::close(this->readHandle);
}
if ((this->writeHandle != 0) &&
(this->writeHandle != -1) &&
(this->writeHandle != this->readHandle))
{
::close(this->writeHandle);
}
this->readHandle = this->writeHandle = -1;
}
bool FileHandle::Init(const AuString &path, EFileOpenMode openMode, bool bDirectIO, EFileAdvisoryLockLevel lock)
{
int fileHandle;
int fOpenMode;
auto pathex = NormalizePathRet(path);
if (pathex.empty())
{
return false;
}
fOpenMode = openMode == EFileOpenMode::eRead ? O_RDONLY : (O_RDWR | O_CREAT);
if (bDirectIO)
{
fOpenMode |= O_DIRECT;
}
fileHandle = ::open(pathex.c_str(),
fOpenMode,
0664);
if (fileHandle == -1)
{
SysPushErrorIO("Couldn't open file: {} ({}) {}", path, pathex, errno);
return false;
}
if (!ApplyDumbAdvisoryLock(fileHandle, lock))
{
SysPushErrorIO("Couldn't open file: {}. File node (not section) is locked.", path);
return false;
}
this->directIO = bDirectIO;
this->readHandle = this->writeHandle = fileHandle;
this->readOnly = openMode == EFileOpenMode::eRead;
return true;
}
void FileHandle::Init(int read, int write)
{
this->readHandle = read;
this->writeHandle = write;
this->directIO = true;
this->readOnly = false;
}
AuSPtr<FileHandle> LinuxAsyncFileStream::GetHandle()
{
return handle_;
}
void LinuxAsyncFileStream::Init(const AuSPtr<FileHandle> &handle)
{
this->handle_ = handle;
}
AuSPtr<IAsyncTransaction> LinuxAsyncFileStream::NewTransaction()
{
auto shared = AuMakeShared<LinuxAsyncFileTransaction>();
if (!shared)
{
return {};
}
if (!shared->Init(this->handle_))
{
return {};
}
return shared;
}
bool LinuxAsyncFileStream::BlockingTruncate(AuUInt64 length)
{
return ::ftruncate(this->handle_->writeHandle, length) != -1;
}
bool LinuxAsyncFileStream::BlockingRead(AuUInt64 offset, const Memory::MemoryViewStreamWrite &parameters)
{
if (this->handle_->bReadLock)
{
return false;
}
if (this->handle_->pIPCPipe)
{
if (this->handle_->pIPCPipe->LIOS_PopOne())
{
parameters.outVariable = 0;
return true;
}
}
if (!PosixSetOffset(this->handle_->readHandle, offset))
{
return false;
}
AuUInt32 read;
if (!PosixRead(this->handle_->readHandle, parameters.ptr, parameters.length, &read))
{
return false;
}
return true;
}
bool LinuxAsyncFileStream::BlockingWrite(AuUInt64 offset, const Memory::MemoryViewStreamRead &parameters)
{
if (this->handle_->bWriteLock)
{
return false;
}
if (!PosixSetOffset(this->handle_->writeHandle, offset))
{
return false;
}
AuUInt32 read;
if (!PosixWrite(this->handle_->writeHandle, parameters.ptr, parameters.length, &read))
{
return false;
}
return true;
}
bool LinuxAsyncFileTransaction::Init(const AuSPtr<FileHandle> &handle)
{
this->handle_ = handle;
this->loopSource_ = AuMakeShared<LinuxAsyncFileTransactionLoopSource>(AuSharedFromThis());
return bool(this->loopSource_);
}
bool LinuxAsyncFileTransaction::StartRead(AuUInt64 offset, const AuSPtr<AuMemoryViewWrite> &memoryView)
{
if (HasState())
{
SysPushErrorIO("IO Operation can not be reused yet.");
return false;
}
auto fd = this->handle_->readHandle;
if (fd == -1)
{
SysPushErrorUninitialized();
return false;
}
if (this->handle_->bReadLock)
{
return false;
}
this->latch_ = false;
this->hasError_ = false;
this->bTxFinished_ = false;
this->lastFinishedStat_ = 0;
if (!this->loopSource_)
{
SysPushErrorUninitialized();
return false;
}
this->loopSource_->Reset();
this->lastAbstractOffset_ = offset;
LIOS_Init(AuSharedFromThis());
SetMemory(memoryView);
if (this->handle_->pIPCPipe)
{
if (this->handle_->pIPCPipe->LIOS_PopOne())
{
LIOS_SendProcess(0, false, errno);
return true;
}
}
if (!UNIX::LinuxOverlappedSubmitRead(fd, offset, this, this->loopSource_.get(), bool(this->handle_->pIPCPipe)))
{
LIOS_SendProcess(0, true, errno);
return true;
}
else
{
if (gRuntimeConfig.bFIODisableBatching)
{
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;
}
if (this->handle_->bWriteLock)
{
return false;
}
auto fd = this->handle_->writeHandle;
if (fd == -1)
{
SysPushErrorUninitialized();
return false;
}
this->latch_ = false;
this->bTxFinished_ = false;
this->hasError_ = false;
this->lastFinishedStat_ = 0;
if (!this->loopSource_)
{
SysPushErrorUninitialized();
return false;
}
this->loopSource_->Reset();
this->lastAbstractOffset_ = offset;
LIOS_Init(AuSharedFromThis());
SetMemory(memoryView);
if (!UNIX::LinuxOverlappedSubmitWrite(fd, offset, this, this->loopSource_.get()))
{
LIOS_SendProcess(0, true, errno);
return false;
}
else
{
if (gRuntimeConfig.bFIODisableBatching)
{
UNIX::SendIOBuffers();
}
return true;
}
}
void LinuxAsyncFileTransaction::Reset()
{
if (this->loopSource_)
{
this->loopSource_->Reset();
}
}
void LinuxAsyncFileTransaction::LIOS_Process(AuUInt32 read, bool failure, int err, bool mark)
{
this->lastFinishedStat_ = failure ? 0 : read;
this->hasError_ = failure;
this->error_ = err;
this->bTxFinished_ = true;
if (mark)
{
return;
}
this->DispatchCb();
if (read)
{
if (this->handle_->pIPCPipe)
{
// Return value intentionally ignored
// We just need to poke on read...
this->handle_->pIPCPipe->LIOS_PopOne();
}
}
}
void LinuxAsyncFileTransaction::DispatchCb()
{
if (AuExchange(this->latch_, true))
{
// TODO (Reece): urgent
//SysPushErrorGeneric();
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();
}
return true;
}
return false;
}
bool LinuxAsyncFileTransaction::Failed()
{
return this->hasError_;
}
AuUInt LinuxAsyncFileTransaction::GetOSErrorCode()
{
return AuUInt(this->error_);
}
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<FileHandle> LinuxAsyncFileTransaction::GetFileHandle()
{
return this->handle_;
}
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)
{
AuSPtr<FileHandle> fileHandle;
LinuxAsyncFileStream *stream;
if (path.empty())
{
SysPushErrorParam("Empty path");
return {};
}
if (!EFileOpenModeIsValid(openMode))
{
SysPushErrorParam("Invalid open mode");
return {};
}
fileHandle = AuMakeShared<FileHandle>();
if (!fileHandle->Init(path, openMode, bDirectIO, lock))
{
SysPushErrorNested();
return {};
}
stream = _new LinuxAsyncFileStream();
if (!stream)
{
SysPushErrorMem();
return {};
}
stream->Init(fileHandle);
return stream;
}
AUKN_SYM void OpenAsyncRelease(IAsyncFileStream *handle)
{
AuSafeDelete<LinuxAsyncFileStream *>(handle);
}
}