Merge pull request #32 from maralla/sndbuf-full

fix bug of client writable events cannot be triggered when sndbuf full

Author: maralla

src/client.c

@@ -20,23 +20,24 @@ int client_trigger_event(struct connection *client, struct mbuf *buf)
     return CORVUS_OK;
 }
 
-int client_read(struct connection *client, int read_socket)
+int client_read(struct connection *client)
 {
     struct command *cmd;
     struct mbuf *buf;
     int status = CORVUS_OK, limit = 16;
 
-    if (!STAILQ_EMPTY(&client->info->cmd_queue)
-            && STAILQ_FIRST(&client->info->cmd_queue)->parse_done)
-    {
-        event_reregister(&client->ctx->loop, client, E_WRITABLE);
-        return CORVUS_OK;
+    cmd = STAILQ_FIRST(&client->info->cmd_queue);
+    if (cmd != NULL && cmd->parse_done) {
+        int size = socket_sndbuf_size(client->fd);
+        LOG(DEBUG, "%p %d %d", cmd, client->info->sndbuf, size);
+        if (size >= 0 && client->info->sndbuf > size) {
+            return CORVUS_OK;
+        }
     }
 
     do {
         buf = conn_get_buf(client);
         if (mbuf_read_size(buf) <= 0) {
-            if (!read_socket) break;
             buf = conn_get_buf(client);
             status = conn_read(client, buf);
             if (status != CORVUS_OK) return status;
@@ -89,6 +90,10 @@ int client_write(struct connection *client)
     }
 
     if (info->iov.len <= 0) {
+        if (!STAILQ_EMPTY(&info->cmd_queue) && conn_register(client) == CORVUS_ERR) {
+            LOG(ERROR, "client_write: fail to reregister client %d", client->fd);
+            return CORVUS_ERR;
+        }
         cmd_iov_reset(&info->iov);
         return CORVUS_OK;
     }
@@ -102,7 +107,7 @@ int client_write(struct connection *client)
     if (status == CORVUS_AGAIN) return CORVUS_OK;
 
     if (info->iov.cursor >= info->iov.len) {
-        cmd_iov_reset(&info->iov);
+        cmd_iov_free(&info->iov);
         if (event_reregister(&ctx->loop, client, E_READABLE) == CORVUS_ERR) {
             LOG(ERROR, "client_write: fail to reregister client %d", client->fd);
             return CORVUS_ERR;
@@ -112,7 +117,7 @@ int client_write(struct connection *client)
                     client->fd, client->ev->fd);
             return CORVUS_ERR;
         }
-    } else if (event_reregister(&ctx->loop, client, E_WRITABLE) == CORVUS_ERR) {
+    } else if (conn_register(client) == CORVUS_ERR) {
         LOG(ERROR, "client_write: fail to reregister client %d", client->fd);
         return CORVUS_ERR;
     }
@@ -147,7 +152,7 @@ void client_ready(struct connection *self, uint32_t mask)
     if (mask & E_READABLE) {
         LOG(DEBUG, "client readable");
 
-        int status = client_read(self, 1);
+        int status = client_read(self);
         if (status == CORVUS_ERR || status == CORVUS_EOF) {
             client_eof(self);
             return;
@@ -178,7 +183,7 @@ void client_event_ready(struct connection *self, uint32_t mask)
     client->info->last_active = time(NULL);
 
     if (mask & E_READABLE) {
-        if (client_read(client, 0) == CORVUS_ERR) {
+        if (client_read(client) == CORVUS_ERR) {
             client_eof(client);
             return;
         }
@@ -208,6 +213,12 @@ struct connection *client_create(struct context *ctx, int fd)
         return NULL;
     }
 
+    client->info->sndbuf = socket_get_sndbuf(client->fd);
+    if (client->info->sndbuf == -1) {
+        client->info->sndbuf = DEFAULT_SNDBUF;
+    }
+    client->info->sndbuf >>= 1;
+
     int evfd = socket_create_eventfd();
     client->ev = conn_create(ctx);
     if (evfd == -1 || client->ev == NULL) {
@@ -248,7 +259,7 @@ void client_eof(struct connection *client)
 
     // don't care response any more
     cmd_iov_clear(client->ctx, &client->info->iov);
-    cmd_iov_reset(&client->info->iov);
+    cmd_iov_free(&client->info->iov);
 
     // request may not write
     if (client->info->refcount <= 0) {

src/command.c

@@ -591,7 +591,7 @@ int cmd_extra(struct command *cmd, struct redis_data *data)
 
 int cmd_forward(struct command *cmd, struct redis_data *data)
 {
-    LOG(DEBUG, "forward command %d", cmd->cmd_type);
+    LOG(DEBUG, "forward command %p(%d)", cmd, cmd->cmd_type);
     switch (cmd->request_type) {
         case CMD_BASIC:
             cmd->slot = cmd_get_slot(data);
@@ -798,9 +798,7 @@ void cmd_mark(struct command *cmd, int fail)
         }
     }
 
-    if (root != NULL && event_reregister(&cmd->ctx->loop,
-                root->client, E_WRITABLE) == CORVUS_ERR)
-    {
+    if (root != NULL && conn_register(root->client) == CORVUS_ERR) {
         LOG(ERROR, "fail to reregister client %d", root->client->fd);
         client_eof(root->client);
     }

src/connection.c

@@ -96,6 +96,7 @@ void conn_info_init(struct conn_info *info)
     reader_init(&info->reader);
 
     info->last_active = -1;
+    info->sndbuf = 0;
 
     STAILQ_INIT(&info->cmd_queue);
     STAILQ_INIT(&info->ready_queue);

src/connection.h

@@ -44,6 +44,7 @@ struct conn_info {
     struct reader reader;
 
     int64_t last_active;
+    int sndbuf;
 
     struct cmd_tqh cmd_queue;
     struct cmd_tqh ready_queue;

src/mbuf.c

@@ -3,6 +3,8 @@
 #include "corvus.h"
 #include "logging.h"
 
+#define RECYCLE_LENGTH 8192 // 128mb
+
 static struct mbuf *_mbuf_get(struct context *ctx)
 {
     struct mbuf *mbuf;
@@ -69,6 +71,11 @@ void mbuf_recycle(struct context *ctx, struct mbuf *mbuf)
 {
     ATOMIC_DEC(ctx->mstats.buffers, 1);
 
+    if (ATOMIC_GET(ctx->mstats.free_buffers) > RECYCLE_LENGTH) {
+        mbuf_free(ctx, mbuf);
+        return;
+    }
+
     TAILQ_NEXT(mbuf, next) = NULL;
     TAILQ_INSERT_HEAD(&ctx->free_mbufq, mbuf, next);
 

src/server.c

@@ -85,7 +85,7 @@ int server_write(struct connection *server)
     ATOMIC_INC(info->send_bytes, status);
 
     if (info->iov.cursor >= info->iov.len) {
-        cmd_iov_reset(&info->iov);
+        cmd_iov_free(&info->iov);
     } else if (conn_register(server) == CORVUS_ERR) {
         LOG(ERROR, "server_write: fail to reregister server %d", server->fd);
         return CORVUS_ERR;
@@ -320,7 +320,7 @@ void server_eof(struct connection *server, const char *reason)
     event_deregister(&server->ctx->loop, server);
 
     // drop all unsent requests
-    cmd_iov_reset(&server->info->iov);
+    cmd_iov_free(&server->info->iov);
     conn_free(server);
     slot_create_job(SLOT_UPDATE);
 }

src/socket.c

@@ -3,6 +3,7 @@
 #include <netinet/tcp.h>
 #include <sys/uio.h>
 #include <sys/eventfd.h>
+#include <sys/ioctl.h>
 #include <netdb.h>
 #include <stdio.h>
 #include <stdlib.h>
@@ -400,3 +401,25 @@ int socket_trigger_event(int evfd)
     }
     return CORVUS_OK;
 }
+
+int socket_get_sndbuf(int fd)
+{
+    int n;
+    socklen_t len = sizeof(n);
+
+    if (getsockopt(fd, SOL_SOCKET, SO_SNDBUF, &n, &len) == -1) {
+        LOG(WARN, "getsockopt SO_SNDBUF: %s", strerror(errno));
+        return CORVUS_ERR;
+    }
+    return n;
+}
+
+int socket_sndbuf_size(int fd)
+{
+    int size = 0;
+    if (ioctl(fd, TIOCOUTQ, &size) == -1) {
+        LOG(WARN, "ioctl TIOCOUTQ: %s", strerror(errno));
+        return CORVUS_ERR;
+    }
+    return size;
+}

src/socket.h

@@ -5,6 +5,7 @@
 #include <limits.h>
 #include "mbuf.h"
 
+#define DEFAULT_SNDBUF 32768
 #define IP_LEN 45
 #define DSN_LEN (IP_LEN + 8)
 
@@ -33,5 +34,7 @@ int socket_parse_ip(char *addr, struct address *address);
 void socket_get_key(struct address *addr, char *dst);
 int socket_create_eventfd();
 int socket_trigger_event(int evfd);
+int socket_sndbuf_size(int fd);
+int socket_get_sndbuf(int fd);
 
 #endif /* end of include guard: __SOCKET_H */

tests/test_corvus.py

@@ -3126,3 +3126,13 @@ def test_delete_node(delete_keys):
     cluster.wait()
 
     assert r.get("hello") == "123"
+
+
+def test_writable(delete_keys):
+    delete_keys.keys("hello")
+
+    with r.pipeline(transaction=False) as p:
+        for u in range(20):
+            p.set('hello', 'x' * 1024 * 1024 * 17)
+            p.get('hello')
+        p.execute()