AuroraRuntime/Source/IO/IPC/IPCPipe.NT.cpp

/***
    Copyright (C) 2022 J Reece Wilson (a/k/a "Reece"). All rights reserved.

    File: IPCPipe.NT.cpp
    Date: 2022-4-15
    Author: Reece
***/
#include <Source/RuntimeInternal.hpp>
#include "IPC.hpp"
#include "IPCHandle.hpp"

#include <Source/IO/Loop/ILoopSourceEx.hpp>
#include <Source/IO/Loop/LSHandle.hpp>
#include <Source/IO/Loop/LSEvent.hpp>
#include <Source/IO/FS/Async.NT.hpp>

#include "IPCPipe.NT.hpp"

namespace Aurora::IO::IPC
{
    //////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////
    // Pipes
    //////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////
    struct IPCPipeImpl;

    struct IPCHasConnectionEvent : Loop::LSHandle
    {
        IPCHasConnectionEvent(AuSPtr<IPCPipeImpl> parent);

        bool IsSignaled() override;
        Loop::ELoopSource GetType() override;

        bool OnTrigger(AuUInt handle) override;

    private:
        AuWPtr<IPCPipeImpl> parent_;
    };

    IPCHasConnectionEvent::IPCHasConnectionEvent(AuSPtr<IPCPipeImpl> parent) : parent_(parent), LSHandle((AuUInt)parent->GetConnectHandle())
    {

    }

    bool IPCHasConnectionEvent::IsSignaled()
    {
        return OnTrigger(0);
    }

    Loop::ELoopSource IPCHasConnectionEvent::GetType()
    {
        return Loop::ELoopSource::eSourceIPCHasClient;
    }

    bool IPCHasConnectionEvent::OnTrigger(AuUInt handle)
    {
        auto parent = this->parent_.lock();
        if (!parent)
        {
            SysPushErrorMem("IPC pipe is dead");
            return false;
        }

        parent->TryConnect();
        return WaitForSingleObject(parent->overlapped.hEvent, 0) == WAIT_OBJECT_0;
    }

    IPCPipeImpl::IPCPipeImpl(HANDLE clientHandle, HANDLE serverHandle, const IPCHandle &handle) :
        serverHandle_(serverHandle), clientHandle_(clientHandle), ipcHandle_(handle)
    {
        
        if (serverHandle != INVALID_HANDLE_VALUE)
        {
            this->hasClient_ = Loop::NewLSEvent(false, false, true);
        }

        this->fsHandle_ = AuMakeShared<IO::FS::FileHandle>();
        this->fsStream_ = AuMakeShared<IO::FS::NtAsyncFileStream>();

        this->fsHandle_->Init(this->GetPipeHandle(), this->GetPipeHandle());
        this->fsStream_->Init(this->fsHandle_);

        TryConnect();
    }

    void IPCPipeImpl::TryConnect()
    {
        DWORD idc;

        if (this->serverHandle_ == INVALID_HANDLE_VALUE)
        {
            return;
        }

        this->overlapped.hEvent = GetConnectHandle();
     
        bool firstTime = AuExchange(bFirstTime, false);
        if (firstTime || 
            (WaitForSingleObject(this->overlapped.hEvent, 0) == WAIT_OBJECT_0 &&
             (GetOverlappedResult(this->serverHandle_, &this->overlapped, &idc, false))))
        {
            ResetEvent(this->overlapped.hEvent);

            if (ConnectNamedPipe(this->serverHandle_, &this->overlapped))
            {
                bFirstTime = true;
                TryConnect();
            }
            else if (GetLastError() == ERROR_IO_PENDING)
            {
                // No-op
            }
            else if (GetLastError() == ERROR_PIPE_CONNECTED)
            {
                SetEvent(this->overlapped.hEvent);
            }
        }
        else
        {
        #if 0
            if (WaitForSingleObject(this->overlapped.hEvent, 0) == WAIT_OBJECT_0)
            {
                ResetEvent(this->overlapped.hEvent);
            }
        #endif
        }
    }

    IPCPipeImpl::~IPCPipeImpl()
    {
    }

    AuSPtr<Loop::ILoopSource> IPCPipeImpl::AsReadChannelIsOpen()
    {
        if (this->serverHandle_ == INVALID_HANDLE_VALUE)
        {
            return {};
        }

        if (!this->lshasConnection_)
        {
            this->lshasConnection_ = AuMakeShared<IPCHasConnectionEvent>(AuSharedFromThis());
        }

        return this->lshasConnection_;
    }

    AuSPtr<Loop::ILoopSource> IPCPipeImpl::AsReadChannelHasData()
    {
        // TODO (Hack): we should at least make a shared timer
        return AuUnsafeRaiiToShared(this);
    }
    
    AuSPtr<IO::IAsyncTransaction> IPCPipeImpl::NewAsyncTransaction()
    {
        auto transaction = AuStaticCast<AuFS::NtAsyncFileTransaction>(this->fsStream_->NewTransaction());
        if (transaction)
        {
            transaction->ntIpcPipeImpl = AuSharedFromThis();
        }
        return transaction;
    }

    bool IPCPipeImpl::Read(const Memory::MemoryViewStreamWrite &write, bool nonblocking)
    {
        DWORD size = write.length;

        TryConnect();

        auto h = this->GetPipeHandle();
        if (h == INVALID_HANDLE_VALUE)
        {
            SysPushErrorUninitialized();
            return false;
        }

        if (nonblocking || !write.ptr)
        {
            DWORD avail {};
            if (!PeekNamedPipe(h, NULL, NULL, NULL, &avail, NULL))
            {
                return false;
            }

            if (!avail)
            {
                return true;
            }

            size = AuMin(size, avail);
        }

        if (!write.ptr)
        {
            write.outVariable = size;
            return true;
        }

        auto ret = ::ReadFile(h, write.ptr, size, &size, NULL);
        write.outVariable = size;
        return ret;
    }

    bool IPCPipeImpl::Write(const Memory::MemoryViewStreamRead &read)
    {
        auto h = this->GetPipeHandle();
        if (h == INVALID_HANDLE_VALUE)
        {
            SysPushErrorUninitialized();
            return false;
        }

        TryConnect();

        DWORD temp;
        if (!::WriteFile(h, read.ptr, read.length, &temp, nullptr))
        {
            SysPushErrorIO();
            return false;
        }

        read.outVariable = temp;
        return true;
    }

    HANDLE IPCPipeImpl::GetPipeHandle()
    {
        return this->clientHandle_ == INVALID_HANDLE_VALUE ? this->serverHandle_ : this->clientHandle_;
    }

    HANDLE IPCPipeImpl::GetConnectHandle()
    {
        return (HANDLE)AuStaticCast<Loop::LSEvent>(this->hasClient_)->GetHandle();
    }

    void IPCPipeImpl::OnEndOfReadStream()
    {
        // TODO: fire inverse LS

        DisconnectNamedPipe(this->serverHandle_);
        this->bFirstTime = true;
        this->TryConnect();
    }

    bool IPCPipeImpl::IsSignaled()
    {
        DWORD avail {};

        TryConnect();

        if (!PeekNamedPipe(this->GetPipeHandle(), NULL, NULL, NULL, &avail, NULL))
        {
            return false;
        }

        return avail;
    }

    bool IPCPipeImpl::WaitOn(AuUInt32 timeout)
    {
        return LSHandle::WaitOn(timeout);
    }

    Loop::ELoopSource IPCPipeImpl::GetType()
    {
        return Loop::ELoopSource::eSourceIPCReadPipe;
    }

    AuString IPCPipeImpl::ExportToString()
    {
        TryConnect();

        return this->clientHandle_ == INVALID_HANDLE_VALUE ?
               this->ipcHandle_.ToString() :
               AuString {};
    }

    AUKN_SYM AuSPtr<IPCPipe>      NewPipe()
    {
        IPCHandle handle;
        handle.NewId();

        auto name = "\\\\.\\pipe\\" + handle.ToNTPath();
        auto maxLength = 16 * AuHwInfo::GetPageSize() ? AuHwInfo::GetPageSize() : 4096;
        auto pipeServer = CreateNamedPipeA(name.c_str(),
                                          PIPE_ACCESS_DUPLEX | FILE_FLAG_FIRST_PIPE_INSTANCE | FILE_FLAG_WRITE_THROUGH | FILE_FLAG_OVERLAPPED,
                                          PIPE_WAIT,
                                          1,
                                          maxLength,
                                          maxLength,
                                          NMPWAIT_WAIT_FOREVER,
                                          nullptr);

        if (pipeServer == INVALID_HANDLE_VALUE)
        {
            SysPushErrorIO("{}", GetLastError());
            return {};
        }

        auto object = AuMakeShared<IPCPipeImpl>(INVALID_HANDLE_VALUE, pipeServer, handle);
        if (!object)
        {
            SysPushErrorMem();
            AuWin32CloseHandle(pipeServer);
            return {};
        }

        return object;
    }

    AUKN_SYM AuSPtr<IPCPipe>      ImportPipe(const AuString &handleString)
    {
        IPCHandle handle;
        HANDLE pipe;

        if (!handle.FromString(handleString))
        {
            SysPushErrorParseError();
            return {};
        }

        auto name = "\\\\.\\pipe\\" + handle.ToNTPath();
        pipe = CreateFileA(name.c_str(),
                            GENERIC_WRITE | GENERIC_READ,
                            0,
                            NULL,
                            OPEN_ALWAYS,
                            FILE_ATTRIBUTE_NORMAL | FILE_FLAG_OVERLAPPED,
                            NULL);

        if (pipe == INVALID_HANDLE_VALUE)
        {
            if (GetLastError() == ERROR_PIPE_BUSY)
            {
                SysPushErrorIO("Pipe is used -> a client has already connected or the nt server is not ready");
                return {};
            }
            SysPushErrorIO("{}", GetLastError());
            return {};
        }

        auto object = AuMakeShared<IPCPipeImpl>(pipe, INVALID_HANDLE_VALUE, handle);
        if (!object)
        {
            SysPushErrorMem();
            AuWin32CloseHandle(pipe);
            return {};
        }

        return object;
    }
}

// > The pipe created by UWP process with name \\.\pipe\Local\PipeName is converted to \\.\pipe\Sessions\<SessionId>\AppContainerNamedObjects\<AppContainerSid>\PipeName. 
// > I can use this to communicate between UWP as server and Win32 as client.
//  https://jike.in/qa/?qa=103904/
//  ...good to know