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

/***
    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/AuIPCPipe.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);
    }
}