add wqueue.

Author: Ruk33

bin/.game_server.dll

@@ -8,10 +8,10 @@ del *.pdb
 
 set flags=/Z7 /W4 /wd4996 /wd4201 /wd4477 /nologo /diagnostics:column -FC -Gm- -GR-
 
-cl %flags% ../login_server.c ../directory.c ../net_windows.c /Fe:login_server.exe -I../openssl-1.1/x64/include ../openssl-1.1/x64/lib/libcrypto.lib Ws2_32.lib
+cl %flags% ../login_server.c ../directory.c ../net_windows.c ../wqueue.c /Fe:login_server.exe -I../openssl-1.1/x64/include ../openssl-1.1/x64/lib/libcrypto.lib Ws2_32.lib
 
 cl %flags% /LD ../game_server.c ../directory.c ../net_windows.c /Fe:game_server.dll /link -incremental:no /PDB:game_server_%random%.pdb /EXPORT:on_init /EXPORT:on_connection /EXPORT:on_request /EXPORT:on_response /EXPORT:on_disconnect /EXPORT:on_tick Ws2_32.lib
 
-cl %flags% ../runner.c ../net_windows.c /Fe:game_server.exe /link /PDB:game_server.pdb Ws2_32.lib
+cl %flags% ../runner.c ../net_windows.c ../wqueue.c /Fe:game_server.exe /link /PDB:game_server.pdb Ws2_32.lib
 
 popd

bin/game_server.dll

@@ -601,26 +601,25 @@ void on_request(void **buf, int socket, void *request, size_t len)
     handle_request(state, conn);
 }
 
-void on_response(void **buf, int socket)
+void on_response(void **buf)
 {
     assert(buf);
     struct state *state = *(struct state **) buf;
     assert(state);
-    struct connection *conn = get_connection_from_socket(state, socket);
-    if (!conn) {
-        trace("answering a non connected client?" nl);
-        return;
-    }
-    if (conn->sent < conn->to_send_count) {
-        trace("sending %llu bytes of data" nl, conn->to_send_count - conn->sent);
-        conn->sent += net_send(conn->socket, 
-                               conn->to_send + conn->sent,
-                               conn->to_send_count - conn->sent);
-    }
-    // reset counters when all data has been sent.
-    if (conn->sent >= conn->to_send_count) {
-        conn->sent = 0;
-        conn->to_send_count = 0;
+
+    for (size_t i = 0; i < countof(state->connections); i++) {
+        struct connection *conn = state->connections + i;
+        if (conn->sent < conn->to_send_count) {
+            void *head = conn->to_send + conn->sent;
+            unsigned long long to_send = conn->to_send_count - conn->sent;
+            trace("sending %d bytes of data" nl, (s32) to_send);
+            conn->sent += net_send(conn->socket, head, to_send);
+        }
+        // reset counters when all data has been sent.
+        if (conn->sent >= conn->to_send_count) {
+            conn->sent = 0;
+            conn->to_send_count = 0;
+        }
     }
 }
 
@@ -656,23 +655,19 @@ int on_tick(void **buf)
 
     state->d = 1000.0f;
     state->run_time += (double) state->d;
-    // state->ticks++;
     // 1000/10 = 100
     state->ticks = (u64) (state->run_time / 100.0);
 
-    // trace("ticks %u" nl, state->ticks);
-    
     if (state->ticks == old_ticks)
-        return 1;
-    
-    // returns 17
-    // trace("clock %d" nl, millis());
-    // trace("tick" nl);
+        return 0;
+
+    int flush_responses = 0;
 
     for (size_t i = 0; i < countof(state->characters); i++) {
         struct character *character = state->characters + i;
         if (!character->active)
             continue;
+        flush_responses = 1;
         switch (character->action_type) {
             case idle:
             // idle_update(state, character);
@@ -688,7 +683,7 @@ int on_tick(void **buf)
         }
     }
 
-    return 1;
+    return flush_responses;
 }
 
 static struct connection *get_connection_from_socket(struct state *state, int socket)

bin/game_server.exe

@@ -16,6 +16,7 @@
 
 #include "directory.h"
 #include "net.h"
+#include "wqueue.h"
 
 typedef uint8_t byte;
 
@@ -67,6 +68,7 @@ struct connection {
     RSA *rsa_key;
 };
 
+static struct wqueue send_responses_worker = {0};
 static struct connection connections[32] = {0};
 
 static u32 ip_to_u32(char *ip)
@@ -704,31 +706,46 @@ static void handle_event(int socket, enum net_event event, void *read, size_t le
             on_request(conn);
         } break;
 
-        case net_write: {
-            if (conn->sent < conn->to_send_count) {
-                trace("sending %d bytes of data" nl, 
-                      (s32) (conn->to_send_count - conn->sent));
-                conn->sent += net_send(conn->socket,
-                                       conn->to_send + conn->sent,
-                                       conn->to_send_count - conn->sent);
-            }
-            // reset counters when all data has been sent.
-            if (conn->sent >= conn->to_send_count) {
-                conn->sent = 0;
-                conn->to_send_count = 0;
-            }
-        } break;
-        
         default:
         break;
     }
+    // we assume we always need to reply with something
+    // when we get a request
+    wpush(&send_responses_worker, conn);
+}
+
+static void send_responses(struct wqueue *q, void *w)
+{
+    struct connection *conn = (struct connection *) w;
+
+    void *head = conn->to_send + conn->sent;
+    unsigned long long to_send = conn->to_send_count - conn->sent;
+    trace("sending %d bytes of data" nl, to_send);
+    conn->sent += net_send(conn->socket, head, to_send);
+
+    // if we couldn't sent the entire response, re-add
+    // this connection to the worker so we try
+    // to flush later again.
+    // NOTE(fmontenegro) do we want to try up to n times
+    // and then dropping the connection?
+    if (conn->sent < conn->to_send_count) {
+        wpush(q, conn);
+        return;
+    }
+
+    // all data has been sent, reset the counters.
+    conn->sent = 0;
+    conn->to_send_count = 0;
 }
 
 int main()
 {
-#define port 2106
+    wstart(&send_responses_worker, send_responses);
+
+    unsigned short port = 2106;
     int socket = net_port(port);
     trace("login server, listening for connection on port %d" nl, port);
     net_listen(socket, handle_event);
+
     return 0;
 }

bin/login_server.exe

@@ -5,8 +5,6 @@ enum net_event {
     net_closed,
     // can read
     net_read,
-    // can write
-    net_write,
 };
 
 typedef void (net_handler)
@@ -20,8 +18,8 @@ int net_sock(char *path);
 void net_listen(int server, net_handler *handler);
 // try to send the full buffer. returns the amount
 // of bytes that was able to send. if it writes
-// less than the intended amount, wait for net_write
-// to keep sending past the bytes already sent.
+// less than the intended amount, you will have to 
+// keep trying sending the rest of the data.
 unsigned long long net_send(int socket, void *buf, unsigned long long n);
 // close the socket.
 void net_close(int socket);

build.bat

@@ -126,7 +126,7 @@ void net_listen(int server, net_handler *handler)
                         break;
                     }
                     event.data.fd = client;
-                    event.events = EPOLLIN | EPOLLOUT | EPOLLET;
+                    event.events = EPOLLIN | EPOLLET;
                     if (epoll_ctl(epoll_fd, EPOLL_CTL_ADD, client, &event) == -1) {
                         print_err("failed to add new client to epoll");
                         close(client);
@@ -155,8 +155,6 @@ void net_listen(int server, net_handler *handler)
                     handler(events[i].data.fd, net_read, read_buf, (size_t) received);
                 }
             }
-            if (events[i].events & EPOLLOUT)
-                handler(events[i].data.fd, net_write, 0, 0);
         }
     }
 

game_server.c

@@ -55,7 +55,7 @@ void net_listen(int server, net_handler *handler)
     static SOCKET clients[max_clients];
     static unsigned char read_buf[8192] = {0};
 
-    int highest = 0;
+    SOCKET highest = 0;
 
     if (!handler) {
         printf("net error: no socket request handler provided.\n");
@@ -67,9 +67,7 @@ void net_listen(int server, net_handler *handler)
 
     while (1) {
         fd_set readfds;
-        fd_set writefds;
         FD_ZERO(&readfds);
-        FD_ZERO(&writefds);
         FD_SET(server, &readfds);
 
         highest = server;
@@ -83,15 +81,14 @@ void net_listen(int server, net_handler *handler)
             int optlen = sizeof(optval);
             if (getsockopt(clients[i], SOL_SOCKET, SO_ERROR, (char*) &optval, &optlen) != SOCKET_ERROR && optval == 0) {
                 FD_SET(clients[i], &readfds);
-                FD_SET(clients[i], &writefds);
                 if (clients[i] > highest)
                     highest = clients[i];
             } else {
                 clients[i] = INVALID_SOCKET;
             }
         }
 
-        int activity = select(highest + 1, &readfds, &writefds, 0, 0);
+        int activity = select((int) (highest + 1), &readfds, 0, 0, 0);
         if (activity == SOCKET_ERROR) {
             printf("select error %d\n", WSAGetLastError());
             break;
@@ -153,10 +150,6 @@ void net_listen(int server, net_handler *handler)
                     }
                 }
             }
-            
-            // write.
-            if (FD_ISSET(clients[i], &writefds))
-                handler((int) clients[i], net_write, 0, 0);
         }
     }
 }

login_server.c

@@ -5,11 +5,12 @@
 #endif
 
 #include "net.h"
+#include "wqueue.h"
 
 typedef int on_init(void **buf);
 typedef void on_connection(void **buf, int socket);
 typedef void on_request(void **buf, int socket, void *request, size_t len);
-typedef void on_response(void **buf, int socket);
+typedef void on_response(void **buf);
 typedef void on_disconnect(void **buf, int socket);
 typedef int on_tick(void **buf);
 
@@ -26,6 +27,8 @@ struct state {
     on_response *on_response;
     on_disconnect *on_disconnect;
     on_tick *on_tick;
+
+    struct wqueue send_responses_worker;
 
 #ifdef _WIN32
     FILETIME lib_write_time;
@@ -92,7 +95,7 @@ static void load_lib_if_required(void)
 #endif
 }
 
-void handle_event(int socket, enum net_event event, void *read, size_t len)
+static void handle_event(int socket, enum net_event event, void *read, size_t len)
 {
     load_lib_if_required();
 
@@ -109,26 +112,33 @@ void handle_event(int socket, enum net_event event, void *read, size_t len)
         state.on_request(&state.buf, socket, read, len);
         break;
 
-        case net_write:
-        state.on_response(&state.buf, socket);
-        break;
-        
         default:
         break;
     }
+
+    wpush(&state.send_responses_worker, 0);
 }
 
-DWORD timer_thread(LPVOID param)
+static DWORD timer_thread(LPVOID param)
 {
     param = param;
     while (1) {
         Sleep(1000);
         load_lib_if_required();
-        state.on_tick(&state.buf);
+        if (state.on_tick(&state.buf))
+            wpush(&state.send_responses_worker, 0);
     }
     return 0;
 }
 
+static void send_responses(struct wqueue *q, void *w)
+{
+    q = q;
+    w = w;
+    load_lib_if_required();
+    state.on_response(&state.buf);
+}
+
 int main()
 {
     if (!lib_load())
@@ -138,6 +148,8 @@ int main()
         return 0;
 
     CreateThread(0, 0, timer_thread, 0, 0, 0);
+
+    wstart(&state.send_responses_worker, send_responses);
 
     int socket = net_port(7777);
     net_listen(socket, handle_event);

net.h

@@ -0,0 +1,88 @@
+#include <windows.h>
+#include "wqueue.h"
+
+static unsigned long wqueue_thread(void *p)
+{
+    struct wqueue *q = (struct wqueue *) p;
+    while (1) {
+        int close = 0;
+        EnterCriticalSection(&q->lock);
+        close = q->closed && !q->wcount;
+        LeaveCriticalSection(&q->lock);
+        if (close)
+            break;
+
+        WaitForSingleObject(q->wpending, INFINITE);
+        EnterCriticalSection(&q->lock);
+        void *work = q->work[0];
+        for (size_t i = 1; i < q->wcount; i++)
+            q->work[i - 1] = q->work[i];
+        q->wcount--;
+        LeaveCriticalSection(&q->lock);
+        q->worker((void *) q, work);
+    }
+
+    CloseHandle(q->wpending);
+    CloseHandle(q->thread);
+    DeleteCriticalSection(&q->lock);
+
+    return 0;
+}
+
+void wstart(struct wqueue *q, worker *worker)
+{
+    InitializeCriticalSection(&q->lock);
+    q->closed = 0;
+    q->worker = worker;
+    q->wpending = CreateEvent(0, FALSE, FALSE, 0);
+    q->thread = CreateThread(0, 0, wqueue_thread, q, 0, 0);
+}
+
+void wpush(struct wqueue *q, void *work)
+{
+    if (q->closed)
+        return;
+    EnterCriticalSection(&q->lock);
+    q->work[q->wcount++] = work;
+    LeaveCriticalSection(&q->lock);
+    SetEvent(q->wpending);
+}
+
+void wclose(struct wqueue *q)
+{
+    if (q->closed)
+        return;
+    EnterCriticalSection(&q->lock);
+    q->closed = 1;
+    LeaveCriticalSection(&q->lock);
+}
+
+#ifdef run_wqueue
+// example on how to use these functions
+#include <stdio.h>
+
+void do_work(struct wqueue *q, void *w)
+{
+    static int tmp = 42;
+    int value = *(int *) w;
+    if (value == 2) {
+        fprintf(stderr, "adding new value because we found 2\n");
+        wpush(q, &tmp);
+    }
+    fprintf(stderr, "new work! value is %d\n", value);
+}
+
+int main()
+{
+    struct wqueue q = {0};
+    wstart(&q, do_work);
+
+    int values[] = {1, 2, 3, 4, 5};
+    for (int i = 0; i < 5; i++) {
+        wpush(&q, values + i);
+        // wclose(&q);
+        Sleep(1000);
+    }
+    fprintf(stderr, "all done!\n");
+}
+#endif

net_linux.c

@@ -0,0 +1,18 @@
+#include <windows.h>
+
+struct wqueue;
+typedef void (worker)(struct wqueue *q, void *work);
+
+struct wqueue {
+    CRITICAL_SECTION lock;
+    HANDLE wpending;
+    HANDLE thread;
+    worker *worker;
+    void *work[256];
+    size_t wcount;
+    int closed;
+};
+
+void wstart(struct wqueue *q, worker *worker);
+void wpush(struct wqueue *q, void *work);
+void wclose(struct wqueue *q);