disable EVENT_READ when waiting for EVENT_WRITE, and vice versa

deps/ccommon/CMakeLists.txt

@@ -51,6 +51,7 @@ option(HAVE_ASSERT_LOG "assert_log enabled by default" ON)
 option(HAVE_ASSERT_PANIC "assert_panic disabled by default" OFF)
 option(HAVE_LOGGING "logging enabled by default" ON)
 option(HAVE_STATS "stats enabled by default" ON)
+option(HAVE_DEBUG_MM "debugging oriented memory management disabled by default" OFF)
 option(COVERAGE "code coverage" OFF)
 
 include(CheckIncludeFiles)

deps/ccommon/config.h.in

@@ -17,3 +17,5 @@
 #cmakedefine HAVE_LOGGING
 
 #cmakedefine HAVE_STATS
+
+#cmakedefine HAVE_DEBUG_MM

deps/ccommon/docs/c-styleguide.txt

@@ -16,11 +16,13 @@
   <stdint.h>. However, when interfacing with system calls and libraries
   you cannot get away from using int and char.
 - Use bool for boolean variables. You have to include <stdbool.h>
-- Avoid using a bool as type for struct member names. Instead use unsigned
-  1-bit bit field. Eg:
+- If memory usage or alignment is a concern, avoid using a bool as type for
+  struct member names. Instead use unsigned 1-bit bit field. e.g.
   struct foo {
       unsigned is_bar:1;
   };
+  However, if neither memory usage or alignment will be significantly impacted
+  by the struct, opt for using bool for the sake of readability.
 - Always use size_t type when dealing with sizes of objects or memory ranges.
 - Your code should be 64-bit and 32-bit friendly. Bear in mind problems
   of printing, comparisons, and structure alignment. You have to include

deps/ccommon/include/cc_array.h

@@ -46,7 +46,7 @@ struct array {
     uint32_t nalloc;    /* # allocated element */
     size_t   size;      /* element size */
     uint32_t nelem;     /* # element */
-    void     *data;     /* elements */
+    uint8_t  *data;     /* elements */
 };
 
 
@@ -93,7 +93,7 @@ array_data_assign(struct array *arr, uint32_t nalloc, size_t size, void *data)
  * element is out of bounds, return -1.
  */
 static inline int
-array_locate(struct array *arr, void *elem) {
+array_locate(struct array *arr, uint8_t *elem) {
     int idx;
 
     idx = (elem - arr->data) / arr->size;

deps/ccommon/include/cc_define.h

@@ -54,10 +54,9 @@ extern "C" {
 # define CC_BACKTRACE 1
 #endif
 
-/* TODO: add compile time option to turn chaining on/off */
-/*#ifdef HAVE_CHAINED*/
-# define CC_HAVE_CHAINED 1
-/*#endif*/
+#ifdef HAVE_DEBUG_MM
+#define CC_DEBUG_MM 1
+#endif
 
 #define CC_OK        0
 #define CC_ERROR    -1

deps/ccommon/include/cc_mm.h

@@ -17,6 +17,8 @@
 
 #pragma once
 
+#include <cc_define.h>
+
 #include <stddef.h>
 
 /*
@@ -41,8 +43,13 @@
 #define cc_calloc(_n, _s)                                       \
     _cc_calloc((size_t)(_n), (size_t)(_s), __FILE__, __LINE__)
 
+#if defined CC_DEBUG_MM && CC_DEBUG_MM == 1
+#define cc_realloc(_p, _s)                                      \
+    _cc_realloc_move(_p, (size_t)(_s), __FILE__, __LINE__)
+#else
 #define cc_realloc(_p, _s)                                      \
     _cc_realloc(_p, (size_t)(_s), __FILE__, __LINE__)
+#endif
 
 #define cc_free(_p) do {                                        \
     _cc_free(_p, __FILE__, __LINE__);                           \
@@ -59,6 +66,7 @@ void * _cc_alloc(size_t size, const char *name, int line);
 void * _cc_zalloc(size_t size, const char *name, int line);
 void * _cc_calloc(size_t nmemb, size_t size, const char *name, int line);
 void * _cc_realloc(void *ptr, size_t size, const char *name, int line);
+void * _cc_realloc_move(void *ptr, size_t size, const char *name, int line);
 void _cc_free(void *ptr, const char *name, int line);
 void * _cc_mmap(size_t size, const char *name, int line);
 int _cc_munmap(void *p, size_t size, const char *name, int line);

deps/ccommon/include/cc_signal.h

@@ -48,7 +48,7 @@ struct signal {
  * - SIGSEGV(debug): print stacktrace before reraise segfault again
  * - SIGPIPE(channel): ignored, this prevents service from exiting when pipe closes
  */
-struct signal signals[SIGNAL_MAX]; /* there are only 31 signals from 1 to 31 */
+extern struct signal signals[SIGNAL_MAX]; /* there are only 31 signals from 1 to 31 */
 
 int signal_override(int signo, char *info, int flags, uint32_t mask, sig_fn handler);
 

deps/ccommon/src/cc_debug.c

@@ -105,6 +105,12 @@ _logrotate(int signo)
 void
 debug_log_flush(void *arg)
 {
+    /*
+     * arg is unused but necessary for debug_log_flush to be used in conjunction
+     * with cc_timer and cc_wheel facilities, since to be inserted into a timing
+     * wheel the function must have the type signature of timeout_cb_fn.
+     */
+    (void)arg;
     log_flush(dlog->logger);
 }
 

deps/ccommon/src/cc_mm.c

@@ -34,7 +34,10 @@ _cc_alloc(size_t size, const char *name, int line)
 {
     void *p;
 
-    ASSERT(size != 0);
+    if (size == 0) {
+        log_debug("malloc(0) @ %s:%d", name, line);
+        return NULL;
+    }
 
     p = malloc(size);
     if (p == NULL) {
@@ -70,7 +73,11 @@ _cc_realloc(void *ptr, size_t size, const char *name, int line)
 {
     void *p;
 
-    ASSERT(size != 0);
+    if (size == 0) {
+        free(ptr);
+        log_debug("realloc(0) @ %s:%d", name, line);
+        return NULL;
+    }
 
     p = realloc(ptr, size);
     if (p == NULL) {
@@ -82,10 +89,37 @@ _cc_realloc(void *ptr, size_t size, const char *name, int line)
     return p;
 }
 
+void *
+_cc_realloc_move(void *ptr, size_t size, const char *name, int line)
+{
+    void *p = NULL, *pr;
+
+    if (size == 0) {
+        free(ptr);
+        log_debug("realloc(0) @ %s:%d", name, line);
+        return NULL;
+    }
+
+    /*
+     * Calling realloc then malloc allows us to force this function call to
+     * change the address of the allocated memory block. realloc ensures we can
+     * copy size bytes, and calling malloc before the realloc'd data is free'd
+     * gives us a new address for the memory object.
+     */
+    if (((pr = realloc(ptr, size)) == NULL || (p = malloc(size)) == NULL)) {
+        log_error("realloc(%zu) failed @ %s:%d", size, name, line);
+    } else {
+        log_vverb("realloc(%zu) at %p @ %s:%d", size, p, name, line);
+        memcpy(p, pr, size);
+    }
+
+    free(pr);
+    return p;
+}
+
 void
 _cc_free(void *ptr, const char *name, int line)
 {
-    ASSERT(ptr != NULL);
     log_vverb("free(%p) @ %s:%d", ptr, name, line);
     free(ptr);
 }
@@ -103,10 +137,10 @@ _cc_mmap(size_t size, const char *name, int line)
      * is set appropriately.
      */
     p = mmap(NULL, size, PROT_READ | PROT_WRITE, MAP_PRIVATE | MAP_ANONYMOUS,
-             -1, 0);
+            -1, 0);
     if (p == ((void *) -1)) {
         log_error("mmap %zu bytes @ %s:%d failed: %s", size, name, line,
-                  strerror(errno));
+                strerror(errno));
         return NULL;
     }
 
@@ -128,7 +162,7 @@ _cc_munmap(void *p, size_t size, const char *name, int line)
     status = munmap(p, size);
     if (status < 0) {
         log_error("munmap %p @ %s:%d failed: %s", p, name, line,
-                  strerror(errno));
+                strerror(errno));
     }
 
     return status;

deps/ccommon/src/cc_signal.c

@@ -8,6 +8,8 @@
 #include <signal.h>
 #include <string.h>
 
+struct signal signals[SIGNAL_MAX];
+
 #ifndef CC_HAVE_SIGNAME
 const char* sys_signame[SIGNAL_MAX + 1] = {
     "UNDEFINED",

src/core/data/worker.c

@@ -46,14 +46,27 @@ _worker_write(struct buf_sock *s)
     return status;
 }
 
-static inline void
+/* the caller only needs to check the return status of this function if
+ * it previously received a write event and wants to re-register the
+ * read event upon full, successful write.
+ */
+static inline rstatus_i
 _worker_event_write(struct buf_sock *s)
 {
     rstatus_i status;
     struct tcp_conn *c = s->ch;
 
     status = _worker_write(s);
     if (status == CC_ERETRY || status == CC_EAGAIN) { /* retry write */
+        /* by removing current masks and only listen to write event(s), we are
+         * effectively stopping processing incoming data until we can write
+         * something to the (kernel) buffer for the channel. This is sensible
+         * because either the local network or the client is backed up when
+         * kernel write buffer is full, and this allows us to propagate back
+         * pressure to the sending side.
+         */
+
+        event_del(ctx->evb, hdl->wid(c));
         event_add_write(ctx->evb, hdl->wid(c), s);
     } else if (status == CC_ERROR) {
         c->state = CHANNEL_TERM;
@@ -62,8 +75,10 @@ _worker_event_write(struct buf_sock *s)
     if (processor->post_write(&s->rbuf, &s->wbuf, &s->data) < 0) {
         log_debug("handler signals channel termination");
         s->ch->state = CHANNEL_TERM;
-        return;
+        return CC_ERROR;
     }
+
+    return status;
 }
 
 static inline void
@@ -171,9 +186,14 @@ _worker_event(void *arg, uint32_t events)
             INCR(worker_metrics, worker_event_read);
             _worker_event_read(s);
         } else if (events & EVENT_WRITE) {
+            /* got here only when a previous write was incompleted/retried */
             log_verb("processing worker write event on buf_sock %p", s);
             INCR(worker_metrics, worker_event_write);
-            _worker_event_write(s);
+            if (_worker_event_write(s) == CC_OK) {
+                /* write backlog cleared up, re-add read event (only) */
+                event_del(ctx->evb, hdl->wid(s->ch));
+                event_add_read(ctx->evb, hdl->rid(s->ch), s);
+            }
         } else if (events & EVENT_ERR) {
             s->ch->state = CHANNEL_TERM;
             INCR(worker_metrics, worker_event_error);

src/core/data/worker.h

@@ -22,8 +22,7 @@ typedef struct {
     ACTION( worker_event_loop,      METRIC_COUNTER, "# worker event loops returned" )\
     ACTION( worker_event_read,      METRIC_COUNTER, "# worker core_read events"     )\
     ACTION( worker_event_write,     METRIC_COUNTER, "# worker core_write events"    )\
-    ACTION( worker_event_error,     METRIC_COUNTER, "# worker core_error events"    )\
-    ACTION( worker_oom_ex,          METRIC_COUNTER, "# worker error due to oom"     )
+    ACTION( worker_event_error,     METRIC_COUNTER, "# worker core_error events"    )
 
 typedef struct {
     CORE_WORKER_METRIC(METRIC_DECLARE)

src/protocol/data/memcache/compose.c

@@ -121,7 +121,7 @@ compose_req(struct buf **buf, struct request *req)
 {
     request_type_t type = req->type;
     struct bstring *str = &req_strings[type];
-    struct bstring *key = req->keys->data;
+    struct bstring *key = (struct bstring *)req->keys->data;
     int noreply_len = req->noreply * NOREPLY_LEN;
     int cas_len = (req->type == REQ_CAS) ? CC_UINT64_MAXLEN : 0;
     uint32_t i;