510 lines
12 KiB
C++
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 ¶meters)
|
|
{
|
|
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 ¶meters)
|
|
{
|
|
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);
|
|
}
|
|
} |