AuroraRuntime/Source/IO/IPC/IPCMutexFutex.Linux.cpp

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

    File: IPCMutexFutex.Linux.cpp
    Date: 2022-
    Author: Reece
    Note: 
***/
#include <Source/RuntimeInternal.hpp>
#include "IPC.hpp"
#include "IPCHandle.hpp"
#include "IPCMutexFutex.Linux.hpp"

#include <linux/futex.h> 
#include <syscall.h> 
#include <Source/IO/UNIX/FDIpcServer.hpp>

////////////////////////////////////////////////////////////////////////////////////
// SysCalls
////////////////////////////////////////////////////////////////////////////////////

#if !defined(FUTEX_OWNER_DIED)
    #define FUTEX_OWNER_DIED        0x40000000
#endif

static int futex(uint32_t *uaddr, int futex_op, uint32_t val,
                 const struct timespec *timeout,   /* or: uint32_t val2 */
                 uint32_t *uaddr2, uint32_t val3)
{
    return syscall(SYS_futex, uaddr, futex_op, val, timeout, uaddr2, val3);
}

static int futex_wait(uint32_t *addr, uint32_t expected)
{
    return futex(addr, FUTEX_WAIT, expected, 0, 0, 0);
}

static int futex_wake(uint32_t *addr, uint32_t nthreads)
{
    return futex(addr, FUTEX_WAKE, nthreads, 0, 0, 0);
}

static long set_robust_list(struct procmon_head *head, size_t len)
{
    return syscall(SYS_set_robust_list, head, len);
}

static long get_robust_list(int pid, struct robust_list_head **head_ptr, size_t *len_ptr)
{
    return syscall(SYS_get_robust_list, pid, head_ptr, len_ptr);
}


////////////////////////////////////////////////////////////////////////////////////
// CONSTANTS
////////////////////////////////////////////////////////////////////////////////////

static const AuUInt8  kMaxFutexes = 128;
static const AuUInt32 kFutexArraySize = AuUInt32(kMaxFutexes) * 4;

static const AuUInt32 kFutexValueOwner     = 0x80000000;
static const AuUInt32 kFutexValueLocked    = kFutexValueOwner | 1;
static const AuUInt32 kFutexValueUnlocked  = kFutexValueOwner;
static const AuUInt32 kFutexValueNULL      = 0;
 
static const AuUInt32 kFutexValueMask      = kFutexValueLocked;
static const AuUInt32 kFutexValueMaskOwner = kFutexValueOwner | FUTEX_OWNER_DIED;

////////////////////////////////////////////////////////////////////////////////////
// VARIABLES
////////////////////////////////////////////////////////////////////////////////////

static AuSPtr<AuIPC::IPCSharedMemory> gFutexSharedMemory;
static AuUInt32 *gFutexArray;
static bool gFutexInit {};

static AuIOIPC::IMutexClosedHook * gFutexCallbacks[kMaxFutexes];


static void InitFutexAPI()
{
    static AuThreadPrimitives::SpinLock lock;

    AU_LOCK_GUARD(lock);

    if (AuExchange(gFutexInit, true))
    {
        return;
    }

    gFutexSharedMemory = AuIOIPC::NewSharedMemory(kFutexArraySize);
    SysAssert(gFutexSharedMemory);

    gFutexArray = gFutexSharedMemory->GetMemory().Begin<AuUInt32>(); 
}

struct FutexContext
{
    bool bInit {};

    // https://github.com/bminor/glibc/blob/78fb88827362fbd2cc8aa32892ae5b015106e25c/sysdeps/nptl/dl-tls_init_tp.c#L93
    // Every thread should have an arraylist head
    robust_list_head *head;


    bool Init();
};


bool FutexContext::Init()
{
    if (AuExchange(this->bInit, true))
    {
        return true;
    }

    InitFutexAPI();

    return this->bInit;
}

static thread_local FutexContext tlsFutexContext;

static AuUInt8 AllocateFutex()
{
    tlsFutexContext.Init();
    if (!gFutexArray)
    {
        return 255;
    }

    for (AuUInt32 i = 0;
         i < kMaxFutexes;
         i++)
    {
        if (AuAtomicCompareExchange<AuUInt32>(&gFutexArray[i], kFutexValueUnlocked, AuUInt32(0)) == 0)
        {
            return i;
        }
    }

    return 255;
}

static bool TryReleaseFutex(AuUInt8 index)
{
    auto &cb = gFutexCallbacks[index];
    if (cb)
    {
        if (!cb->OnClosed())
        {
            return false;
        }
    }

    cb = nullptr;
    gFutexArray[index] = 0;
    gFutexCallbacks[index] = nullptr;
    return true;
}

static void FreeFutex(AuUInt8 index)
{
    gFutexArray[index] = 0;
    gFutexCallbacks[index] = nullptr;
}

static void LinuxAddRobustFutexSlow(AuUInt32 *futex)
{
    tlsFutexContext.Init();

}

static void LinuxRemoveRobustFutexSlow(AuUInt32 *futex)
{
    tlsFutexContext.Init();

}

static void LinuxLockFutex(AuUInt32 *futex)
{
    
}

static void LinuxUnlockFutex(AuUInt32 *futex)
{
    
}

static void LinuxSuperSecretIOTick()
{
    if (!gFutexInit)
    {
        return;
    }

    if (!gFutexArray)
    {
        return;
    }

    for (AuUInt32 i = 0;
         i < kMaxFutexes;
         i++)
    {
        if (gFutexArray[i] == FUTEX_OWNER_DIED)
        {
            TryReleaseFutex(i);
        }
    }
}

#include "IPCPrimitives.Linux.hpp"
#include "IPCMemory.Unix.hpp"

namespace Aurora::IO::IPC
{
    static AuThreadPrimitives::SpinLock gLock;
    static AuBST<AuPair<AuUInt64 /*cookie*/, AuUInt32 /*pid*/>, AuWPtr<IPCSharedMemory>> gSharedViewCache;

    //////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////
    // Mutexes
    //////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////

    IPCMutexProxy::IPCMutexProxy(AuUInt32 index) : mutex_(), bOwned(true), index_(index)
    {
        if (this->mutex_.HasValidHandle())
        {
            if (IO::UNIX::FDServe(this->GetHandle(), this->token_))
            {
                //this->handle_.PushId(EIPCHandleType::eIPCPrimitiveMutex, token);
            }
            else
            {
                this->mutex_.~LSMutex();
            }
        }
        
        this->mem_ = gFutexSharedMemory;
    }

    IPCMutexProxy::IPCMutexProxy(int handle, AuSPtr<IPCSharedMemory> mem, AuUInt32 index) :
        mutex_(handle),
        mem_(mem),
        index_(index)
    {
        if (this->mutex_.HasValidHandle())
        {
            if (IO::UNIX::FDServe(this->GetHandle(), this->token_))
            {
                //this->handle_.PushId(EIPCHandleType::eIPCPrimitiveMutex, token);
            }
            else
            {
                this->mutex_.~LSMutex();
            }
        }
    }

    IPCMutexProxy::~IPCMutexProxy()
    {
        IO::UNIX::FDServeEnd(this->token_);

        if (this->bOwned)
        {
            FreeFutex(this->index_);
        }
    }

    bool IPCMutexProxy::Unlock()
    {
        return this->mutex_.Unlock();
    }

    bool IPCMutexProxy::IsSignaled()
    {
        return this->mutex_.IsSignaled();
    }

    bool IPCMutexProxy::WaitOn(AuUInt32 timeout)
    {
        return this->mutex_.WaitOn(timeout);
    }

    Loop::ELoopSource IPCMutexProxy::GetType()
    {
        return this->mutex_.GetType();
    }

    AuString IPCMutexProxy::ExportToString()
    {
        IPC::IPCHandle handle;

        handle.PushId(EIPCHandleType::eIPCPrimitiveMutex, this->token_);
        SysAssert(this->mem_);
        handle.PushId(EIPCHandleType::eIPCMemory, AuStaticCast<IPCSharedMemoryImpl>(this->mem_)->handle_.values[0].token);
        handle.values[0].token.word = this->index_;

        return handle.ToString();
    }

    AUKN_SYM AuSPtr<IPCMutex>     NewMutex()
    {
        auto futex = AllocateFutex();
        if (futex == 255)
        {
            return {};
        }

        auto object = AuMakeShared<IPCMutexProxy>(futex);
        if (!object)
        {
            SysPushErrorMem();
            return {};
        }

        if (!object->HasValidHandle())
        {
            SysPushErrorIO();
            return {};
        }

        return object;
    }

    static AuSPtr<IPCSharedMemory> GetFutexPagesFromCacheOrImport(const IPCToken &mem)
    {
        AU_LOCK_GUARD(gLock);

        auto id  = AuMakePair(mem.cookie, mem.pid);
        auto itr = gSharedViewCache.find(id);
        if (itr != gSharedViewCache.end())
        {
            auto test = itr->second.lock();
            if (test)
            {
                return test;
            }
        }

        auto shared = ImportSharedMemoryEx(mem);
        if (!shared)
        {
            SysPushErrorNested();
            return {};
        }

        if (!AuTryInsert(gSharedViewCache, id, shared))
        {
            SysPushErrorMem();
            // We don't need to fail. We can leak next time. It'll be fine.
        }

        return shared;
    }

    AuSPtr<IPCMutex> ImportMutexEx(const IPCToken &handle, const IPCToken &mem, AuUInt32 index)
    {
        int fd {-1};
        if (!IO::UNIX::FDAccept(handle, fd))
        {
            SysPushErrorNested();
            return {};
        }

        auto view = GetFutexPagesFromCacheOrImport(mem);
        if (!view)
        {
            SysPushErrorNested();
            return {};
        }

        auto object = AuMakeShared<IPCMutexProxy>(fd, view, index);
        if (!object)
        {
            SysPushErrorMem();
            ::close(fd);
            return {};
        }

        if (!object->HasValidHandle())
        {
            SysPushErrorIO();
            return {};
        }

        return object;
    }

    AUKN_SYM AuSPtr<IPCMutex> ImportMutex(const AuString &handle)
    {
        IPC::IPCHandle decodedHandle;

        if (!decodedHandle.FromString(handle))
        {
            SysPushErrorParseError("Invalid handle: {}", handle);
            return {};
        }

        auto val = decodedHandle.GetToken(IPC::EIPCHandleType::eIPCPrimitiveMutex, 0);
        if (!val)
        {
            SysPushErrorParseError("Invalid handle: {}", handle);
            return {};
        }

        auto mem = decodedHandle.GetToken(IPC::EIPCHandleType::eIPCMemory, 0);
        if (!mem)
        {
            SysPushErrorParseError("Invalid handle: {}", handle);
            return {};
        }

        return ImportMutexEx(val->token, mem->token, val->token.word);
    }
}

namespace Aurora::Grug
{
    void LinuxSuperSecretIOTick()
    {
        ::LinuxSuperSecretIOTick();
    }
}