Lua uvFifoListener ffi

src/lua/fileffi.lua

@@ -20,6 +20,7 @@ ffi.cdef [[
 typedef struct { char opaque[?]; } LuaFile;
 typedef struct { uint32_t size; uint8_t data[?]; } LuaBuffer;
 typedef struct { char opaque[?]; } LuaSlice;
+typedef struct { char opaque[?]; } LuaServer;
 
 enum FileOps {
     READ,
@@ -49,6 +50,12 @@ LuaFile* subFile(LuaFile*, uint64_t start, int64_t size);
 
 LuaFile* uvFifo(const char* address, int port);
 
+LuaServer* uvFifoListener();
+
+void stopListener(LuaServer* server);
+void startListener(LuaServer* server, unsigned port, void (*cb)(LuaFile* fifo));
+void deleteListener(LuaServer* server);
+
 void closeFile(LuaFile* wrapper);
 
 uint64_t readFileRawPtr(LuaFile* wrapper, void* dst, uint64_t size);
@@ -339,6 +346,36 @@ local function createFileWrapper(wrapper)
     return file
 end
 
+local function createUvListener(port, cb)
+    if type(port) ~= 'number' then error 'port argument must be an unsigned number' end
+    if (port) == nil then error 'must provide a port' end
+    if cb == nil then error 'must provide a callback function' end
+    if type(cb) ~= 'function' then error 'callback must be a function' end
+
+    local _callback = function(fifo)
+        file = createFileWrapper(fifo)
+        cb(file)
+    end
+
+    _callback = ffi.cast('void (*)(LuaFile* fifo)', _callback)
+
+    local wrapper = C.uvFifoListener()
+
+    local listener = {
+        _type = "LuaServer",
+        _wrapper = wrapper,
+        _proxy = newproxy(),
+        _cb = _callback,
+        _port = port,
+        start = function(listener) C.startListener(listener._wrapper, listener._port, listener._cb)  end,
+        stop = function(listener) C.stopListener(listener._wrapper) end,
+    }
+    -- Use a proxy instead of doing this on the wrapper directly using ffi.gc, because of a bug in LuaJIT,
+    -- where circular references on finalizers using ffi.gc won't actually collect anything.
+    debug.setmetatable(listener._proxy, { __gc = function()  C.deleteListener(listener._wrapper) listener._cb = nil end })
+    return listener
+end
+
 local function open(filename, t)
     if (t == nil) then t = 'READ' end
     if (t == 'DOWNLOAD_URL_AND_WAIT') then
@@ -395,6 +432,10 @@ local function uvFifo(address, port)
     return createFileWrapper(C.uvFifo(address, port))
 end
 
+local function uvFifoListener(port, callback)
+    return createUvListener(port, callback)
+end
+
 if (type(Support) ~= 'table') then Support = {} end
 
 Support.NewLuaBuffer = function(size)
@@ -408,6 +449,8 @@ Support.File = {
     buffer = buffer,
     zReader = zReader,
     uvFifo = uvFifo,
+    uvFifoListener = uvFifoListener,
+    _createUvListener = createUvListener,
     _createFileWrapper = createFileWrapper,
     _createSliceWrapper = createSliceWrapper,
 }

src/lua/luafile.cc

@@ -27,6 +27,7 @@
 namespace {
 
 using LuaFile = PCSX::LuaFFI::LuaFile;
+using LuaServer = PCSX::LuaFFI::LuaServer;
 
 enum FileOps {
     READ,
@@ -73,6 +74,44 @@ LuaFile* subFile(LuaFile* wrapper, uint64_t start, int64_t size) {
 }
 LuaFile* uvFifo(const char* address, int port) { return new LuaFile(new PCSX::UvFifo(address, port)); }
 
+LuaServer* uvFifoListener() {return new LuaServer(new PCSX::UvFifoListener());}
+
+void startListener(LuaServer* server, unsigned port, void (*cb)(LuaFile* fifo)) {
+    server->m_listener->start(port, PCSX::g_system->getLoop(), &server->m_async, [cb, server](PCSX::UvFifo* fifo) {
+        if (fifo) {
+            cb(new LuaFile(fifo));
+            server->m_status = LuaServer::Status::STARTED;
+        } else {
+            server->m_async.data = server;
+            uv_close(reinterpret_cast<uv_handle_t*>(&server->m_async), [](uv_handle_t* handle) {
+                LuaServer* server = reinterpret_cast<LuaServer*>(handle->data);
+                server->m_status = LuaServer::Status::STOPPED;
+                if (server->m_deleting) {
+                    delete server->m_listener;
+                    delete server;
+                }
+            });
+        }
+    });
+}
+
+void stopListener(LuaServer* server) {
+    if (server->m_status == LuaServer::Status::STOPPED) return;
+
+    server->m_status = LuaServer::Status::STOPPING;
+    server->m_listener->stop();
+}
+
+void deleteListener(LuaServer* server) {
+    server->m_deleting = true;
+    if (server->m_status == LuaServer::Status::STARTED) {
+        stopListener(server);
+    } else if (server->m_status == LuaServer::Status::STOPPED) {
+        delete server->m_listener;
+        delete server;
+    }
+}
+
 void closeFile(LuaFile* wrapper) { wrapper->file->close(); }
 
 uint64_t readFileRawPtr(LuaFile* wrapper, void* dst, uint64_t size) { return wrapper->file->read(dst, size); }
@@ -194,6 +233,10 @@ static void registerAllSymbols(PCSX::Lua L) {
     REGISTER(L, bufferFileEmpty);
     REGISTER(L, subFile);
     REGISTER(L, uvFifo);
+    REGISTER(L, uvFifoListener);
+    REGISTER(L, stopListener);
+    REGISTER(L, startListener);
+    REGISTER(L, deleteListener);
 
     REGISTER(L, closeFile);
 

src/lua/luafile.h

@@ -23,6 +23,7 @@
 
 #include "lua/luawrapper.h"
 #include "support/file.h"
+#include "support/uvfile.h"
 
 namespace PCSX {
 
@@ -52,6 +53,15 @@ struct LuaFile {
     IO<File> file;
 };
 
+struct LuaServer {
+    LuaServer(UvFifoListener* listener) : m_listener(listener) {}
+    UvFifoListener* m_listener = nullptr;
+    uv_async_t m_async;
+    bool m_deleting = false;
+
+    enum class Status { STARTED, STOPPED, STOPPING } m_status = Status::STOPPED;
+};
+
 void open_file(Lua);
 }  // namespace LuaFFI
 

src/support/uvfile.cc

@@ -702,6 +702,8 @@ PCSX::UvFifo::UvFifo(const std::string_view address, unsigned port) : File(File:
     m_tcp = tcp;
     request([this, host = std::string(address), port](auto loop) {
         uv_tcp_init(loop, m_tcp);
+        uv_tcp_nodelay(m_tcp, 1);
+
         struct sockaddr_in connectAddr;
         int result = uv_ip4_addr(host.c_str(), port, &connectAddr);
         if (result != 0) {
@@ -852,6 +854,8 @@ void PCSX::UvFifoListener::start(unsigned port, uv_loop_t *loop, uv_async_t *asy
     });
     request([this, port](auto loop) {
         uv_tcp_init(loop, &m_server);
+        uv_tcp_nodelay(&m_server, 1);
+
         m_server.data = this;
 
         struct sockaddr_in bindAddr;
@@ -871,6 +875,7 @@ void PCSX::UvFifoListener::start(unsigned port, uv_loop_t *loop, uv_async_t *asy
             uv_tcp_t *tcp = new uv_tcp_t();
             auto loop = server->loop;
             uv_tcp_init(loop, tcp);
+            uv_tcp_nodelay(tcp, 1);
             if (uv_accept(reinterpret_cast<uv_stream_t *>(server), reinterpret_cast<uv_stream_t *>(tcp)) == 0) {
                 UvFifo *fifo = new UvFifo(tcp);
                 listener->m_pending.Enqueue(fifo);