Per-thread defaultLoop and uWS::run()

15.pro

@@ -14,7 +14,6 @@ SOURCES += \
 
 HEADERS += \
     src/new_design/HttpRouter.h \
-    src/Loop.h \
     src/new_design/HttpParser.h \
     src/websocket/libwshandshake.hpp \
     src/websocket/WebSocketProtocol.h \
@@ -27,7 +26,8 @@ HEADERS += \
     src/new_design/StaticDispatch.h \
     src/new_design/LoopData.h \
     src/new_design/AsyncSocket.h \
-    src/new_design/AsyncSocketData.h
+    src/new_design/AsyncSocketData.h \
+    src/new_design/Loop.h
 
 INCLUDEPATH += uSockets/src src
 QMAKE_CXXFLAGS += -fsanitize=address

main.cpp

@@ -50,18 +50,20 @@ std::string_view getFile(std::string_view file) {
 
 int main(int argc, char **argv) {
 
-    // new stuff, hope it sticks together?
+    // test per-thread loopery
-
+    uWS::Loop::defaultLoop();
-
+    new std::thread([]() {
-
+        uWS::Loop::defaultLoop();
-    uWS::Loop loop;
+        uWS::Loop::defaultLoop();
+    });
+    uWS::Loop::defaultLoop();
 
     us_ssl_socket_context_options ssl_options;
     ssl_options.key_file_name = "/home/alexhultman/uWebSockets/misc/ssl/key.pem";
     ssl_options.cert_file_name = "/home/alexhultman/uWebSockets/misc/ssl/cert.pem";
     ssl_options.passphrase = "1234";
 
-    uWS::HttpContext<true> *httpContext = uWS::HttpContext<true>::create(loop.loop, &ssl_options);
+    uWS::HttpContext<true> *httpContext = uWS::HttpContext<true>::create(uWS::Loop::defaultLoop(), &ssl_options);
 
     // req, res?
     httpContext->onGet("/:folder/:file", [](auto *res, auto *req) {
@@ -87,9 +89,10 @@ int main(int argc, char **argv) {
 
     httpContext->listen(nullptr, 3000, 0);
 
-    loop.run();
+    uWS::run();
 
     httpContext->free();
+    uWS::Loop::defaultLoop()->free();
 
     return 0;
 
@@ -192,6 +195,5 @@ int main(int argc, char **argv) {
         std::cout << "Connections: " << --connections << std::endl;
     }).listen(nullptr, 3000, 0);*/
 
-    uWS::run();
     // loop.run();
 }

src/Loop.h

@@ -1,50 +0,0 @@
-#ifndef HUB_H
-#define HUB_H
-
-// this header needs fixing! should not depend on source files!
-
-#include "new_design/LoopData.h"
-
-#include <libusockets.h>
-
-// Loop is a bit different, it holds the loop pointer, not as "this"
-
-namespace uWS {
-struct Loop {
-    us_loop *loop;
-
-    static void wakeupCb(us_loop *loop) {
-
-    }
-
-    static void preCb(us_loop *loop) {
-
-    }
-
-    static void postCb(us_loop *loop) {
-
-    }
-
-    Loop() : loop(us_create_loop(1, wakeupCb, preCb, postCb, sizeof(LoopData))) {
-
-        new (us_loop_ext(loop)) LoopData();
-    }
-
-    void run() {
-        us_loop_run(loop);
-    }
-
-    ~Loop() {
-        // deconstruct the loopdata
-    }
-};
-
-// dessa måste ligga i en sourcefil
-thread_local Loop defaultLoop;
-
-static void run() {
-    defaultLoop.run();
-}
-}
-
-#endif // HUB_H

src/new_design/AsyncSocket.h

@@ -56,11 +56,11 @@ public:
     int write(const char *src, int length, bool optionally = false, int nextLength = 0) {
         LoopData *loopData = getLoopData();
 
-        std::cout << "Write called with length: " << length << ", optionally: " << optionally << std::endl;
+        //std::cout << "Write called with length: " << length << ", optionally: " << optionally << std::endl;
 
         /* Do nothing for a null sized chunk */
         if (length == 0) {
-            std::cout << "Write returned: 0" << std::endl;
+            //std::cout << "Write returned: 0" << std::endl;
             return 0;
         }
 
@@ -69,7 +69,7 @@ public:
         /* Do not write anything if we have a per-socket buffer */
         if (asyncSocketData->buffer.length()) {
             if (optionally) {
-                std::cout << "Write returned: 0" << std::endl;
+                //std::cout << "Write returned: 0" << std::endl;
                 return 0;
             } else {
                 std::cout << "Buffering at top of write!" << std::endl;
@@ -81,7 +81,7 @@ public:
 
                 /* Buffer this chunk */
                 asyncSocketData->buffer.append(src, length);
-                std::cout << "Write returned: " << length << std::endl;
+                //std::cout << "Write returned: " << length << std::endl;
                 return length;
             }
         }
@@ -100,12 +100,12 @@ public:
 
                     /* Optionally matters here though */
                     written += uncork(src + written, length - written, optionally);
-                    std::cout << "Write returned: " << written << std::endl;
+                    //std::cout << "Write returned: " << written << std::endl;
                     return written;
                 } else {
                     /* For non-SSL we take the penalty of two syscalls */
                     int written = uncork(src, length, optionally);
-                    std::cout << "Write returned: " << written << std::endl;
+                    //std::cout << "Write returned: " << written << std::endl;
                     return written;
                 }
             }
@@ -117,7 +117,7 @@ public:
             if (written < length) {
                 /* If the write was optional then just bail out */
                 if (optionally) {
-                    std::cout << "Write returned: " << written << std::endl;
+                    //std::cout << "Write returned: " << written << std::endl;
                     return written;
                 }
 
@@ -134,7 +134,7 @@ public:
             }
         }
 
-        std::cout << "Write returned: " << length << std::endl;
+        //std::cout << "Write returned: " << length << std::endl;
         return length;
     }
 

src/new_design/HttpContext.h

@@ -167,13 +167,13 @@ private:
 
 public:
     /* Construct a new HttpContext using specified loop */
-    static HttpContext *create(us_loop *loop, us_ssl_socket_context_options *ssl_options = nullptr) {
+    static HttpContext *create(Loop *loop, us_ssl_socket_context_options *ssl_options = nullptr) {
         HttpContext *httpContext;
 
         if constexpr(SSL) {
-            httpContext = (HttpContext *) us_create_ssl_socket_context(loop, sizeof(HttpContextData<SSL>), *ssl_options);
+            httpContext = (HttpContext *) us_create_ssl_socket_context((us_loop *) loop, sizeof(HttpContextData<SSL>), *ssl_options);
         } else {
-            httpContext = (HttpContext *) us_create_socket_context(loop, sizeof(HttpContextData<SSL>));
+            httpContext = (HttpContext *) us_create_socket_context((us_loop *) loop, sizeof(HttpContextData<SSL>));
         }
 
 

src/new_design/Loop.h

@@ -0,0 +1,87 @@
+#ifndef LOOP_H
+#define LOOP_H
+
+#include "LoopData.h"
+
+#include <libusockets.h>
+
+#include <thread>
+
+namespace uWS {
+struct Loop {
+private:
+    static void wakeupCb(us_loop *loop) {
+
+    }
+
+    static void preCb(us_loop *loop) {
+
+    }
+
+    static void postCb(us_loop *loop) {
+
+    }
+
+    Loop() = delete;
+
+    Loop *init() {
+        new (us_loop_ext((us_loop *) this)) LoopData();
+        return this;
+    }
+
+    static Loop *create(bool defaultLoop) {
+        return ((Loop *) us_create_loop(defaultLoop, wakeupCb, preCb, postCb, sizeof(LoopData)))->init();
+    }
+
+public:
+    /* Returns the default loop if called from one thread, or a dedicated per-thread loop if called from multiple threads */
+    static Loop *defaultLoop() {
+        /* Deliver and attach the default loop to the first thread who calls us */
+        static thread_local bool ownsDefaultLoop;
+        static Loop *defaultLoop;
+        if (!defaultLoop) {
+            ownsDefaultLoop = true;
+            defaultLoop = create(true);
+            std::cout << "Created default loop " << defaultLoop << " for thread " << std::this_thread::get_id() << std::endl;
+            return defaultLoop;
+        } else if (ownsDefaultLoop) {
+            std::cout << "Returned default loop " << defaultLoop << " for thread " << std::this_thread::get_id() << std::endl;
+            return defaultLoop;
+        }
+
+        /* Other threads get their non-default loops lazily created */
+        static thread_local Loop *threadLocalLoop;
+        if (!threadLocalLoop) {
+            threadLocalLoop = create(false);
+            std::cout << "Created non-default loop " << threadLocalLoop << " for thread " << std::this_thread::get_id() << std::endl;
+            return threadLocalLoop;
+        }
+        std::cout << "Returned non-default loop " << threadLocalLoop << " for thread " << std::this_thread::get_id() << std::endl;
+        return threadLocalLoop;
+    }
+
+    /* Freeing the default loop should be done once */
+    void free() {
+        us_loop_free((us_loop *) this);
+    }
+
+    /* Actively block and run this loop */
+    void run() {
+        us_loop_run((us_loop *) this);
+    }
+
+    /* Passively integrate with the underlying default loop */
+    /* Used to seamlessly integrate with third parties such as Node.js */
+    void integrate() {
+
+    }
+};
+
+/* Can be called from any thread to run the thread local loop */
+void run() {
+    Loop::defaultLoop()->run();
+}
+
+}
+
+#endif // LOOP_H