Add multiplex, a bunch of cleanup

Signed-off-by: Andrew Opalach <andrew@akon.city>

3 files changed, 101 insertions, 107 deletions

diff --git a/src/socket/socket.h b/src/socket/socket.h
index c0cf796..bb42a3d 100644
--- a/src/socket/socket.h
+++ b/src/socket/socket.h
@@ -25,11 +25,11 @@ enum {
 struct aki_socket {
     u8 type;
 #ifndef _WIN32
-    s32 sock;
+    s32 fd;
     struct sockaddr_in addr_in;
     struct sockaddr_un addr_un;
 #else
-    SOCKET sock;
+    SOCKET fd;
     SOCKADDR_IN addr_in;
 #endif
 };
@@ -45,6 +45,7 @@ struct aki_socket {
 #define aki_htons htons
 #define aki_htonl htonl
 #define aki_ntohs ntohs
+#define aki_ntohl ntohl
 
 #ifndef _WIN32 // Posix-only helpers.
 #define aki_pollfd pollfd
@@ -53,24 +54,24 @@ void aki_fd_set_blocking(s32 fd, bool blocking);
 s32 aki_poll_fds(struct aki_pollfd *fds, aki_nfds nfds, s64 timeout_ns);
 #endif
 
-bool aki_socket_init(struct aki_socket *s);
+bool aki_socket_init(struct aki_socket *sock);
 
-void aki_socket_set_blocking(struct aki_socket *s, bool blocking);
-void aki_socket_set_no_delay(struct aki_socket *s, s32 no_delay);
+void aki_socket_set_blocking(struct aki_socket *sock, bool blocking);
+void aki_socket_set_nodelay(struct aki_socket *sock, s32 nodelay);
 
-void aki_socket_set_send_buf(struct aki_socket *s, u32 sndbuf);
-u32 aki_socket_get_send_buf(struct aki_socket *s);
+void aki_socket_set_send_buf(struct aki_socket *sock, u32 sndbuf);
+u32 aki_socket_get_send_buf(struct aki_socket *sock);
 
-bool aki_socket_listen(struct aki_socket *s, str *addr, s32 port);
-bool aki_socket_accept(struct aki_socket *s, struct aki_socket *c);
-bool aki_socket_connect(struct aki_socket *s, str *addr, s32 port);
+bool aki_socket_listen(struct aki_socket *sock, str *addr, s32 port);
+bool aki_socket_accept(struct aki_socket *sock, struct aki_socket *c);
+bool aki_socket_connect(struct aki_socket *sock, str *addr, s32 port);
 
-s32 aki_socket_get_fd(struct aki_socket *s);
+s32 aki_socket_get_fd(struct aki_socket *sock);
 
-ssize_t aki_socket_read(struct aki_socket *s, void *buf, size_t size);
-ssize_t aki_socket_write(struct aki_socket *s, void *buf, size_t size);
-ssize_t aki_socket_sendto(struct aki_socket *s, void *buf, size_t size);
-ssize_t aki_socket_recvfrom(struct aki_socket *s, void *buf, size_t size);
+ssize_t aki_socket_read(struct aki_socket *sock, void *buf, size_t size);
+ssize_t aki_socket_write(struct aki_socket *sock, void *buf, size_t size);
+ssize_t aki_socket_sendto(struct aki_socket *sock, void *buf, size_t size);
+ssize_t aki_socket_recvfrom(struct aki_socket *sock, void *buf, size_t size);
 
-void aki_socket_shutdown(struct aki_socket *s);
-void aki_socket_close(struct aki_socket *s);
+void aki_socket_shutdown(struct aki_socket *sock);
+void aki_socket_close(struct aki_socket *sock);
diff --git a/src/socket/socket_linux.c b/src/socket/socket_linux.c
index b442806..adab91b 100644
--- a/src/socket/socket_linux.c
+++ b/src/socket/socket_linux.c
@@ -1,9 +1,8 @@
+#include <al/log.h>
 #include <unistd.h>
 #include <sys/fcntl.h>
 #include <netinet/tcp.h>
 #include <netdb.h>
-#include <signal.h>
-#include <al/log.h>
 
 #include "../util/error.h"
 
@@ -11,40 +10,38 @@
 
 void aki_fd_set_blocking(s32 fd, bool blocking)
 {
-    s32 flags = fcntl(fd, F_GETFL, 0);
-    al_assert(flags != -1);
-    flags = blocking ? (flags & ~O_NONBLOCK) : (flags | O_NONBLOCK);
-    s32 ret = fcntl(fd, F_SETFL, flags);
+    s32 ret = fcntl(fd, F_GETFL, 0); // ret = flags
+    al_assert(ret != -1);
+    ret = fcntl(fd, F_SETFL, blocking ? (ret & ~O_NONBLOCK) : (ret | O_NONBLOCK)); // ret = success/error
     al_assert(ret == 0);
 }
 
-#define AKI_TIME_S_TO_NS(s) ((s) * INT64_C(1000000000))
-
 // https://github.com/mpv-player/mpv/blob/e575ec4fc3654387c7358bd3640877ef32628d2c/osdep/poll_wrapper.c#L29
+#define AKI_TIME_S_TO_NS(s) ((s) * INT64_C(1000000000))
 s32 aki_poll_fds(struct aki_pollfd *fds, aki_nfds nfds, s64 timeout_ns)
 {
     struct timespec ts;
-    ts.tv_sec  = timeout_ns / AKI_TIME_S_TO_NS(1);
+    ts.tv_sec = timeout_ns / AKI_TIME_S_TO_NS(1);
     ts.tv_nsec = timeout_ns % AKI_TIME_S_TO_NS(1);
     struct timespec *tsp = timeout_ns >= 0 ? &ts : NULL;
     return ppoll(fds, nfds, tsp, NULL);
 }
 
-bool aki_socket_init(struct aki_socket *s)
+bool aki_socket_init(struct aki_socket *sock)
 {
-    switch (s->type) {
+    switch (sock->type) {
     case AKI_SOCKET_TCP:
-        s->sock = socket(AF_INET, SOCK_STREAM, 0);
+        sock->fd = socket(AF_INET, SOCK_STREAM, 0);
         break;
     case AKI_SOCKET_UDP:
-        s->sock = socket(AF_INET, SOCK_DGRAM, IPPROTO_UDP);
+        sock->fd = socket(AF_INET, SOCK_DGRAM, IPPROTO_UDP);
         break;
     case AKI_SOCKET_UNIX:
-        s->sock = socket(AF_UNIX, SOCK_STREAM, 0);
+        sock->fd = socket(AF_UNIX, SOCK_STREAM, 0);
         break;
     }
 
-    if (s->sock < 0) {
+    if (sock->fd < 0) {
         al_log_error("socket", "socket() failed: %s (%d).", aki_strerror(errno), errno);
         return false;
     }
@@ -52,125 +49,123 @@ bool aki_socket_init(struct aki_socket *s)
     return true;
 }
 
-void aki_socket_set_blocking(struct aki_socket *s, bool blocking)
+void aki_socket_set_blocking(struct aki_socket *sock, bool blocking)
 {
-    aki_fd_set_blocking(s->sock, blocking);
+    aki_fd_set_blocking(sock->fd, blocking);
 }
 
-void aki_socket_set_no_delay(struct aki_socket *s, s32 no_delay)
+void aki_socket_set_nodelay(struct aki_socket *sock, s32 nodelay)
 {
-    if (s->type != AKI_SOCKET_TCP) return;
-    setsockopt(s->sock, IPPROTO_TCP, TCP_NODELAY, &no_delay, sizeof(no_delay));
+    //al_assert(sock->type == AKI_SOCKET_TCP);
+    if (sock->type != AKI_SOCKET_TCP) return; // TODO: 
+    setsockopt(sock->fd, IPPROTO_TCP, TCP_NODELAY, &nodelay, sizeof(nodelay));
 }
 
-u32 aki_socket_get_send_buf(struct aki_socket *s)
+u32 aki_socket_get_send_buf(struct aki_socket *sock)
 {
     u32 send_queue_size;
     socklen_t optlen = sizeof(send_queue_size);
-    getsockopt(s->sock, SOL_SOCKET, SO_SNDBUF, &send_queue_size, &optlen);
+    getsockopt(sock->fd, SOL_SOCKET, SO_SNDBUF, &send_queue_size, &optlen);
     return send_queue_size / 2;
 }
 
-void aki_socket_set_send_buf(struct aki_socket *s, u32 sndbuf)
+void aki_socket_set_send_buf(struct aki_socket *sock, u32 sndbuf)
 {
     u32 send_queue_size = sndbuf;
-    setsockopt(s->sock, SOL_SOCKET, SO_SNDBUF, &send_queue_size, sizeof(send_queue_size));
-    al_assert(aki_socket_get_send_buf(s) == send_queue_size);
+    setsockopt(sock->fd, SOL_SOCKET, SO_SNDBUF, &send_queue_size, sizeof(send_queue_size));
+    al_assert(aki_socket_get_send_buf(sock) == send_queue_size);
 }
 
-u32 aki_socket_get_recv_buf(struct aki_socket *s)
+u32 aki_socket_get_recv_buf(struct aki_socket *sock)
 {
     u32 receive_queue_size;
     socklen_t optlen = sizeof(receive_queue_size);
-    getsockopt(s->sock, SOL_SOCKET, SO_RCVBUF, &receive_queue_size, &optlen);
+    getsockopt(sock->fd, SOL_SOCKET, SO_RCVBUF, &receive_queue_size, &optlen);
     return receive_queue_size / 2;
 }
 
-void aki_socket_set_recv_buf(struct aki_socket *s, u32 rcvbuf)
+void aki_socket_set_recv_buf(struct aki_socket *sock, u32 rcvbuf)
 {
     u32 receive_queue_size = rcvbuf;
-    setsockopt(s->sock, SOL_SOCKET, SO_RCVBUF, &receive_queue_size, sizeof(receive_queue_size));
-    al_assert(aki_socket_get_recv_buf(s) == receive_queue_size);
+    setsockopt(sock->fd, SOL_SOCKET, SO_RCVBUF, &receive_queue_size, sizeof(receive_queue_size));
+    al_assert(aki_socket_get_recv_buf(sock) == receive_queue_size);
 }
 
 #define al_log_socket_err(func) \
     al_log_error("socket", #func"(%.*s:%d) failed: %s (%d).", \
         AL_STR_PRINTF(addr), port, aki_strerror(errno), errno);
 
-bool aki_socket_listen(struct aki_socket *s, str *addr, s32 port)
+bool aki_socket_listen(struct aki_socket *sock, str *addr, s32 port)
 {
     struct sockaddr *saddr = NULL;
     socklen_t addrlen = 0;
 
-    switch (s->type) {
+    switch (sock->type) {
     case AKI_SOCKET_TCP: {
-        s->addr_in.sin_family = AF_INET;
-        s->addr_in.sin_addr.s_addr = htonl(INADDR_ANY);
-        s->addr_in.sin_port = htons(port);
-        saddr = (struct sockaddr *)&s->addr_in;
-        addrlen = sizeof(s->addr_in);
+        sock->addr_in.sin_family = AF_INET;
+        sock->addr_in.sin_addr.s_addr = htonl(INADDR_ANY);
+        sock->addr_in.sin_port = htons(port);
+        saddr = (struct sockaddr *)&sock->addr_in;
+        addrlen = sizeof(sock->addr_in);
         break;
     }
     case AKI_SOCKET_UNIX: {
+        sock->addr_un.sun_family = AF_UNIX;
         char *c_str = al_str_to_c_str(addr);
-        s->addr_un.sun_family = AF_UNIX;
-        size_t len = al_strlen(c_str);
-        al_memcpy(s->addr_un.sun_path, c_str, len);
-        s->addr_un.sun_path[len] = '\0';
+        al_memcpy(sock->addr_un.sun_path, c_str, al_strlen(c_str) + 1);
         al_free(c_str);
-        saddr = (struct sockaddr *)&s->addr_un;
-        addrlen = sizeof(s->addr_un);
+        saddr = (struct sockaddr *)&sock->addr_un;
+        addrlen = sizeof(sock->addr_un);
         break;
     }
     case AKI_SOCKET_UDP:
         al_assert(false);
     }
 
-    if (bind(s->sock, saddr, addrlen) < 0) {
+    if (bind(sock->fd, saddr, addrlen) < 0) {
         al_log_socket_err(bind)
         return false;
     }
 
-    if (listen(s->sock, SOMAXCONN) < 0) {
+    if (listen(sock->fd, SOMAXCONN) < 0) {
         al_log_socket_err(listen)
         return false;
     }
 
     al_log_info("socket", "Listening on %.*s:%d.", AL_STR_PRINTF(addr), port);
 
-    signal(SIGPIPE, SIG_IGN);
-
     return true;
 }
 
-bool aki_socket_accept(struct aki_socket *s, struct aki_socket *c)
+bool aki_socket_accept(struct aki_socket *sock, struct aki_socket *cl)
 {
-    c->type = s->type;
+    cl->type = sock->type;
 
     struct sockaddr_storage addr;
     socklen_t len = sizeof(addr);
-    if ((c->sock = accept(s->sock, (struct sockaddr *)&addr, &len)) == -1) {
+    if ((cl->fd = accept(sock->fd, (struct sockaddr *)&addr, &len)) == -1) {
+        al_log_error("socket", "accept() failed: %s (%d).", aki_strerror(errno), errno);
         return false;
     }
 
-    switch (s->type) {
+    switch (cl->type) {
     case AKI_SOCKET_TCP: {
         s32 port = 0;
-        char ipstr[INET6_ADDRSTRLEN] = {0};
+        char ipstr[INET6_ADDRSTRLEN] = { 0 };
         if (addr.ss_family == AF_INET) {
-            struct sockaddr_in *s = (struct sockaddr_in *)&addr;
-            port = ntohs(s->sin_port);
-            inet_ntop(AF_INET, &s->sin_addr, ipstr, sizeof(ipstr));
+            struct sockaddr_in *saddr = (struct sockaddr_in *)&addr;
+            port = ntohs(saddr->sin_port);
+            inet_ntop(AF_INET, &saddr->sin_addr, ipstr, sizeof(ipstr));
         } else if (addr.ss_family == AF_INET6) {
-            struct sockaddr_in6 *s = (struct sockaddr_in6 *)&addr;
-            port = ntohs(s->sin6_port);
-            inet_ntop(AF_INET6, &s->sin6_addr, ipstr, sizeof(ipstr));
+            struct sockaddr_in6 *saddr = (struct sockaddr_in6 *)&addr;
+            port = ntohs(saddr->sin6_port);
+            inet_ntop(AF_INET6, &saddr->sin6_addr, ipstr, sizeof(ipstr));
         }
         al_log_info("socket", "Connection from %s:%d.", ipstr, port);
         break;
     }
     case AKI_SOCKET_UNIX:
-        al_log_info("socket", "Connection from a UNIX socket.");
+        al_log_info("socket", "Connection on UNIX socket.");
         break;
     case AKI_SOCKET_UDP:
         al_assert(false);
@@ -179,9 +174,9 @@ bool aki_socket_accept(struct aki_socket *s, struct aki_socket *c)
     return true;
 }
 
-bool aki_socket_connect(struct aki_socket *s, str *addr, s32 port)
+bool aki_socket_connect(struct aki_socket *sock, str *addr, s32 port)
 {
-    switch (s->type) {
+    switch (sock->type) {
     case AKI_SOCKET_TCP:
     case AKI_SOCKET_UDP: {
         char *c_str = al_str_to_c_str(addr);
@@ -198,9 +193,9 @@ bool aki_socket_connect(struct aki_socket *s, str *addr, s32 port)
             return false;
         }
 
-        s->addr_in.sin_family = AF_INET;
-        s->addr_in.sin_addr.s_addr = ((struct in_addr *)hptr->h_addr_list[0])->s_addr;
-        s->addr_in.sin_port = htons(port);
+        sock->addr_in.sin_family = AF_INET;
+        sock->addr_in.sin_addr.s_addr = ((struct in_addr *)hptr->h_addr_list[0])->s_addr;
+        sock->addr_in.sin_port = htons(port);
 
         break;
     }
@@ -208,21 +203,21 @@ bool aki_socket_connect(struct aki_socket *s, str *addr, s32 port)
         break;
     }
 
-    switch (s->type) {
+    switch (sock->type) {
     case AKI_SOCKET_TCP:
     case AKI_SOCKET_UDP: {
-        s32 ret = connect(s->sock, (struct sockaddr *)&s->addr_in, sizeof(s->addr_in));
-        if (ret != 0 && errno != 115) {
+        s32 ret = connect(sock->fd, (struct sockaddr *)&sock->addr_in, sizeof(sock->addr_in));
+        if (ret != 0 && errno != EINPROGRESS) { 
             al_log_socket_err(connect)
             return false;
         }
         break;
     }
     case AKI_SOCKET_UNIX: {
-        s->addr_un.sun_family = AF_UNIX;
-        al_memcpy(s->addr_un.sun_path, addr->data, addr->len);
-        s->addr_un.sun_path[addr->len] = '\0';
-        if (connect(s->sock, (struct sockaddr *)&s->addr_un, sizeof(s->addr_un)) < 0) {
+        sock->addr_un.sun_family = AF_UNIX;
+        al_memcpy(sock->addr_un.sun_path, addr->data, addr->len);
+        sock->addr_un.sun_path[addr->len] = '\0';
+        if (connect(sock->fd, (struct sockaddr *)&sock->addr_un, sizeof(sock->addr_un)) < 0) {
             al_log_socket_err(connect)
             return false;
         }
@@ -233,43 +228,41 @@ bool aki_socket_connect(struct aki_socket *s, str *addr, s32 port)
         break;
     }
 
-    signal(SIGPIPE, SIG_IGN);
-
     return true;
 }
 
-ssize_t aki_socket_read(struct aki_socket *s, void *buf, size_t size)
+ssize_t aki_socket_read(struct aki_socket *sock, void *buf, size_t size)
 {
-    return recv(s->sock, buf, size, 0);
+    return recv(sock->fd, buf, size, 0);
 }
 
-ssize_t aki_socket_write(struct aki_socket *s, void *buf, size_t size)
+ssize_t aki_socket_write(struct aki_socket *sock, void *buf, size_t size)
 {
-    return send(s->sock, buf, size, MSG_NOSIGNAL);
+    return send(sock->fd, buf, size, MSG_NOSIGNAL);
 }
 
-ssize_t aki_socket_sendto(struct aki_socket *s, void *buf, size_t size)
+ssize_t aki_socket_sendto(struct aki_socket *sock, void *buf, size_t size)
 {
-    return sendto(s->sock, buf, size, 0, (const struct sockaddr *)&s->addr_in, sizeof(s->addr_in));
+    return sendto(sock->fd, buf, size, 0, (const struct sockaddr *)&sock->addr_in, sizeof(sock->addr_in));
 }
 
-ssize_t aki_socket_recvfrom(struct aki_socket *s, void *buf, size_t size)
+ssize_t aki_socket_recvfrom(struct aki_socket *sock, void *buf, size_t size)
 {
-    socklen_t addrlen = sizeof(s->addr_in);
-    return recvfrom(s->sock, buf, size, 0, (struct sockaddr *)&s->addr_in, &addrlen);
+    socklen_t addrlen = sizeof(sock->addr_in);
+    return recvfrom(sock->fd, buf, size, 0, (struct sockaddr *)&sock->addr_in, &addrlen);
 }
 
-s32 aki_socket_get_fd(struct aki_socket *s)
+s32 aki_socket_get_fd(struct aki_socket *sock)
 {
-    return s->sock;
+    return sock->fd;
 }
 
-void aki_socket_shutdown(struct aki_socket *s)
+void aki_socket_shutdown(struct aki_socket *sock)
 {
-    shutdown(s->sock, SHUT_RDWR);
+    shutdown(sock->fd, SHUT_RDWR);
 }
 
-void aki_socket_close(struct aki_socket *s)
+void aki_socket_close(struct aki_socket *sock)
 {
-    close(s->sock);
+    close(sock->fd);
 }
diff --git a/src/socket/socket_windows.c b/src/socket/socket_windows.c
index dcb7e08..a7192d8 100644
--- a/src/socket/socket_windows.c
+++ b/src/socket/socket_windows.c
@@ -30,10 +30,10 @@ void aki_socket_set_blocking(struct aki_socket *s, bool blocking)
     ioctlsocket(s->sock, FIONBIO, &mode);
 }
 
-void aki_socket_set_no_delay(struct aki_socket *s, s32 no_delay)
+void aki_socket_set_nodelay(struct aki_socket *s, s32 nodelay)
 {
     (void)s;
-    (void)no_delay;
+    (void)nodelay;
 }
 
 #define al_log_socket_err(func) \