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>
#include <Source/IO/AuIOHandle.hpp>

namespace Aurora::IO::FS
{
    #define IPC_PIPE AuStaticCast<AFileHandle>(this->pHandle_)->pIPCPipe

    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)
        {
            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;
    }

    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(AuSPtr<ProcessBlock> pProcessBlock) :
        pProcessBlock_(pProcessBlock)
    {

    }

    LinuxAsyncFileTransaction::~LinuxAsyncFileTransaction()
    {
        if (this->pProcessBlock_)
        {
            AU_LOCK_GUARD(this->pProcessBlock_->externalLock);
            AuTryRemove(this->pProcessBlock_->submits, this);
        }
    }

    void LinuxAsyncFileStream::Init(const AuSPtr<IIOHandle> &handle)
    {
        this->pHandle_ = handle;
    }

    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();
                }
            }
        }
    }

    AuSPtr<IAsyncTransaction> LinuxAsyncFileStream::NewTransaction()
    {
        auto shared = AuMakeShared<LinuxAsyncFileTransaction>(this->pProcessBlock_);
        if (!shared)
        {
            return {};
        }

        if (!shared->Init(this->pHandle_))
        {
            return {};
        }

        if (this->pProcessBlock_)
        {
            AU_LOCK_GUARD(this->pProcessBlock_->externalLock);
            if (!AuTryInsert(this->pProcessBlock_->submits, shared.get()))
            {
                SysPushErrorMemory();
                return {};
            }
        }

        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)
    {
        if (!handle)
        {
            return false;
        }
        this->pHandle_ = handle;
        this->loopSource_ = AuMakeShared<LinuxAsyncFileTransactionLoopSource>(AuSharedFromThis());
        return bool(this->loopSource_);
    }

    void LinuxAsyncFileTransaction::SetBaseOffset(AuUInt64 uBaseOffset)
    {
        this->uBaseOffset = uBaseOffset;
    }

    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)
        {
            return false;
        }

        auto fd = (int)iOptSafe.Value();
        if (fd == -1)
        {
            SysPushErrorUninitialized();
            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 (IPC_PIPE)
        {
            if (IPC_PIPE->LIOS_PopOne())
            {
                LIOS_SendProcess(0, false, errno);
                return true;
            }
        }

        if (this->pProcessBlock_ &&
            this->pProcessBlock_->bDead)
        {
            LIOS_SendProcess(0, false, errno);
            return true;
        }

        offset += this->uBaseOffset;
        
        if (!UNIX::LinuxOverlappedSubmitRead(fd, offset, this, this->loopSource_.get(), bool(IPC_PIPE)))
        {
            LIOS_SendProcess(0, true, errno);
            return true;
        }
        else
        {
            if (gRuntimeConfig.linuxConfig.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;
        }

        auto iOptSafe = this->pHandle_->GetOSWriteHandleSafe();
        if (!iOptSafe)
        {
            return false;
        }

        auto fd = (int)iOptSafe.Value();
        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 (this->pProcessBlock_ &&
            this->pProcessBlock_->bDead)
        {
            LIOS_SendProcess(0, false, errno);
            return true;
        }

        offset += this->uBaseOffset;

        if (!UNIX::LinuxOverlappedSubmitWrite(fd, offset, this, this->loopSource_.get()))
        {
            LIOS_SendProcess(0, true, errno);
            return false;
        }
        else
        {
            if (gRuntimeConfig.linuxConfig.bFIODisableBatching)
            {
                UNIX::SendIOBuffers(); 
            }

            return true;
        }
    }

    void LinuxAsyncFileTransaction::Reset()
    {
        (void)this->LIOS_Cancel();

        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 (IPC_PIPE)
            {
                // Return value intentionally ignored
                // We just need to poke on read...
                IPC_PIPE->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::HasCompleted()
    {
        return this->bTxFinished_;
    }

    bool LinuxAsyncFileTransaction::HasFailed()
    {
        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<IIOHandle> LinuxAsyncFileTransaction::GetFileHandle()
    {
        return this->pHandle_;
    }

    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)
    {
        auto pHandle = AuIO::IOHandleShared();
        if (!pHandle)
        {
            SysPushErrorMemory();
            return nullptr;
        }

        AuIO::IIOHandle::HandleCreate createhandle(path);
        createhandle.eAdvisoryLevel = lock;
        createhandle.eMode = openMode;
        createhandle.bFailIfNonEmptyFile = false;
        createhandle.bDirectIOMode = bDirectIO;
        createhandle.bAsyncHandle = true;

        if (!pHandle->InitFromPath(createhandle))
        {
            SysPushErrorNested();
            return nullptr;
        }

        auto pStream = _new LinuxAsyncFileStream();
        if (!pStream)
        {
            SysPushErrorMemory();
            return nullptr;
        }

        pStream->Init(pHandle);
        return pStream;
    }

    AUKN_SYM void OpenAsyncRelease(IAsyncFileStream *handle)
    {
        AuSafeDelete<LinuxAsyncFileStream *>(handle);
    }
}