Reply Copy Avoidance (#2078)

### Overview
This PR introduces the ability to avoid copying the content of string
object into replies (i.e. bulk string replies) and to allow I/O threads
refer directly to obj->ptr in writev iov.

### Key Changes
* Added capability to reply construction allowing to interleave regular
replies with copy avoid replies in client reply buffers
* Extended write-to-client handlers to support copy avoid replies
* Added copy avoidance of string bulk replies when copy avoidance
indicated by I/O threads
* Minor changes in cluster slots stats in order to support
`network-bytes-out` for copy avoid replies
* Copy avoidance is beneficial for performance despite object size only
starting certain number of threads. So it will be enabled only starting
certain number of threads. Internal configuration
``min-io-threads-copy-avoid`` introduced to manage this number of
threads

**Note**: When copy avoidance disabled content and handling of client
reply buffers remains as before this PR

### Implementation Details
####  ``client`` and  ``clientReplyBlock`` structs:
1. ``buf_encoded`` flag has been added to ``clientReplyBlock`` struct
and to ``client`` struct for static ``c->buf`` to indicate if reply
buffer is in copy avoidance mode (i.e. include headers and payloads) or
not (i.e. plain replies only).
2. ``io_last_written_buf``, ``io_last_written_bufpos``,
``io_last_written_data_len`` fields added ``client`` struct to to keep
track of write state between ``writevToClient`` invocations
####  Reply construction:
1. Original ```_addReplyToBuffer``` and ```_addReplyProtoToList``` have
been renamed to ```_addReplyPayloadToBuffer``` and
```_addReplyPayloadToList``` and extended to support different types of
payloads - regular replies and copy avoid replies.
3. New ```_addReplyToBuffer``` and ```_addReplyProtoToList``` calls now
```_addReplyPayloadToBuffer``` and ```_addReplyPayloadToList``` and used
for adding **regular** replies to client reply buffers.
4. Newly introduced ```_addBulkOffloadToBuffer``` and
```_addBulkOffloadToList``` are used for adding **copy avoid** replies
to client reply buffers.
 
####  Write-to-client infrastructure:
The ```writevToClient``` and ```_postWriteToClient``` has been
significantly changed to support copy avoidance capability.

####  Debug configuration:
1. ``min-io-threads-avoid-copy-reply`` - Minimum number of IO threads
for copy avoidance
2. ``min-string-size-avoid-copy-reply`` - Minimum bulk string size for
copy avoidance when IO threads disabled
3. ``min-string-size-avoid-copy-reply-threaded`` - Minimum bulk string
size for copy avoidance when IO threads enabled

### Testing
1. Existing unit and integration tests passed. Copy avoidance enabled on
tests with ``--io-threads`` flag
2. Added unit tests for copy avoidance functionality

### Performance Tests

Note: pay attention `io-threads 1` config means only main thread with no
additional io-threads, `io-threads 2` means main thread plus 1 I/O
thread, `io-threads 9` means main thread plus 8 I/O threads.

#### 512 byte object size
Tests are conducted on memory optimized instances using:
* 3,000,000 keys
* 512 bytes object size 
* 1000 clients

|io-threads (including main thread)	|Plain Reply	|Copy Avoidance	|
|---	|---	|---	|
|7	|1,160,000	|1,160,000	|
|8	|1,150,000	|1,280,000	|
|9	|1,150,000	|1,330,000	|
|10	|N/A	|1,380,000	|
|11	|N/A	|1,420,000	|

#### Various object size, small number of threads

|iothreads |Data size |Keys |Clients |Instance type |Unstable branch
|Copy Avoidance On |
|---	|---	|---	|---	|---	|---	|---	|
|1	|512 byte	|3,000,000	|1,000	|memory optimized	|195,000	|195,000	|
|2	|512 byte	|3,000,000	|1,000	|memory optimized	|245,000	|245,000	|
|3	|512 byte	|3,000,000	|1,000	|memory optimized	|455,000	|459,000	|
|4	|512 byte	|3,000,000	|1,000	|memory optimized	|685,000	|685,000	|
|	|	|	|	|	|	|	|
|1	|1K	|3,000,000	|1,000	|memory optimized	|185,000	|185,000	|
|2	|1K	|3,000,000	|1,000	|memory optimized	|235,000	|235,000	|
|3	|1K	|3,000,000	|1,000	|memory optimized	|450,000	|450,000	|
|	|	|	|	|	|	|	|
|1	|4K	|1,000,000	|1,000	|network optimized	|182,000	|187,000	|
|2	|4K	|1,000,000	|1,000	|network optimized	|240,000	|238,000	|
|	|	|	|	|	|	|	|
|1	|16K	|1,000,000	|500	|network optimized	|100,000	|120,000	|
|2	|16K	|1,000,000	|500	|network optimized	|140,000	|140,000	|
|3	|16K	|1,000,000	|500	|network optimized	|275,000	|260,000	|
|	|	|	|	|	|	|	|
|1	|32K	|500,000	|500	|network optimized	|57,000	|90,000	|
|2	|32K	|500,000	|500	|network optimized	|110,000	|110,000	|
|3	|32K	|500,000	|500	|network optimized	|215,000	|215,000	|
|	|	|	|	|	|	|	|
|1	|64K	|100,000	|500	|network optimized	|30,000	|57,000	|
|2	|64K	|100,000	|500	|network optimized	|69,000	|61,000	|
|3	|64K	|100,000	|500	|network optimized	|120,000	|120,000	|
|4	|64K	|100,000	|500	|network optimized	|115,000 - 175,000	|175,000	|
|5	|64K	|100,000	|500	|network optimized	|115,000 - 165,000	|230,000	|

---------

Signed-off-by: Alexander Shabanov <[email protected]>
Signed-off-by: xbasel <[email protected]>
Signed-off-by: Madelyn Olson <[email protected]>
Co-authored-by: Alexander Shabanov <[email protected]>
Co-authored-by: Madelyn Olson <[email protected]>

Author: 3 people

src/cluster_slot_stats.c

@@ -131,22 +131,23 @@ static void addReplySortedSlotStats(client *c, slotStatForSort slot_stats[], lon
     }
 }
 
-static int canAddNetworkBytesOut(client *c) {
-    return server.cluster_slot_stats_enabled && server.cluster_enabled && c->slot != -1;
+/* Accumulates egress bytes for the slot. */
+void clusterSlotStatsAddNetworkBytesOutForSlot(int slot, unsigned long long net_bytes_out) {
+    if (!clusterSlotStatsEnabled(slot)) return;
+
+    serverAssert(slot >= 0 && slot < CLUSTER_SLOTS);
+    server.cluster->slot_stats[slot].network_bytes_out += net_bytes_out;
 }
 
 /* Accumulates egress bytes upon sending RESP responses back to user clients. */
 void clusterSlotStatsAddNetworkBytesOutForUserClient(client *c) {
-    if (!canAddNetworkBytesOut(c)) return;
-
-    serverAssert(c->slot >= 0 && c->slot < CLUSTER_SLOTS);
-    server.cluster->slot_stats[c->slot].network_bytes_out += c->net_output_bytes_curr_cmd;
+    clusterSlotStatsAddNetworkBytesOutForSlot(c->slot, c->net_output_bytes_curr_cmd);
 }
 
 /* Accumulates egress bytes upon sending replication stream. This only applies for primary nodes. */
 static void clusterSlotStatsUpdateNetworkBytesOutForReplication(long long len) {
     client *c = server.current_client;
-    if (c == NULL || !canAddNetworkBytesOut(c)) return;
+    if (c == NULL || !clusterSlotStatsEnabled(c->slot)) return;
 
     /* We multiply the bytes len by the number of replicas to account for us broadcasting to multiple replicas at once. */
     len *= (long long)listLength(server.replicas);
@@ -177,24 +178,14 @@ void clusterSlotStatsDecrNetworkBytesOutForReplication(long long len) {
  *    This type is not aggregated, to stay consistent with server.stat_net_output_bytes aggregation.
  * This function covers the internal propagation component. */
 void clusterSlotStatsAddNetworkBytesOutForShardedPubSubInternalPropagation(client *c, int slot) {
-    /* For a blocked client, c->slot could be pre-filled.
-     * Thus c->slot is backed-up for restoration after aggregation is completed. */
-    int _slot = c->slot;
-    c->slot = slot;
-    if (!canAddNetworkBytesOut(c)) {
-        /* c->slot should not change as a side effect of this function,
-         * regardless of the function's early return condition. */
-        c->slot = _slot;
-        return;
-    }
+    if (!clusterSlotStatsEnabled(slot)) return;
 
-    serverAssert(c->slot >= 0 && c->slot < CLUSTER_SLOTS);
-    server.cluster->slot_stats[c->slot].network_bytes_out += c->net_output_bytes_curr_cmd;
+    serverAssert(slot >= 0 && slot < CLUSTER_SLOTS);
+    server.cluster->slot_stats[slot].network_bytes_out += c->net_output_bytes_curr_cmd;
 
     /* For sharded pubsub, the client's network bytes metrics must be reset here,
      * as resetClient() is not called until subscription ends. */
     c->net_output_bytes_curr_cmd = 0;
-    c->slot = _slot;
 }
 
 /* Adds reply for the ORDERBY variant.
@@ -222,9 +213,7 @@ void clusterSlotStatResetAll(void) {
  * would equate to repeating the same calculation twice.
  */
 static int canAddCpuDuration(client *c) {
-    return server.cluster_slot_stats_enabled &&  /* Config should be enabled. */
-           server.cluster_enabled &&             /* Cluster mode should be enabled. */
-           c->slot != -1 &&                      /* Command should be slot specific. */
+    return clusterSlotStatsEnabled(c->slot) &&
            (!server.execution_nesting ||         /* Either; */
             (server.execution_nesting &&         /* 1) Command should not be nested, or */
              c->realcmd->flags & CMD_BLOCKING)); /* 2) If command is nested, it must be due to unblocking. */
@@ -251,8 +240,7 @@ static int canAddNetworkBytesIn(client *c) {
      * Third, blocked client is not aggregated, to avoid duplicate aggregation upon unblocking.
      * Fourth, the server is not under a MULTI/EXEC transaction, to avoid duplicate aggregation of
      * EXEC's 14 bytes RESP upon nested call()'s afterCommand(). */
-    return server.cluster_enabled && server.cluster_slot_stats_enabled && c->slot != -1 && !(c->flag.blocked) &&
-           !server.in_exec;
+    return clusterSlotStatsEnabled(c->slot) && !(c->flag.blocked) && !server.in_exec;
 }
 
 /* Adds network ingress bytes of the current command in execution,
@@ -346,3 +334,7 @@ void clusterSlotStatsCommand(client *c) {
         addReplySubcommandSyntaxError(c);
     }
 }
+
+int clusterSlotStatsEnabled(int slot) {
+    return server.cluster_slot_stats_enabled && server.cluster_enabled && slot != -1;
+}

src/cluster_slot_stats.h

@@ -6,6 +6,7 @@
 /* General use-cases. */
 void clusterSlotStatReset(int slot);
 void clusterSlotStatResetAll(void);
+int clusterSlotStatsEnabled(int slot);
 
 /* cpu-usec metric. */
 void clusterSlotStatsAddCpuDuration(client *c, ustime_t duration);
@@ -17,6 +18,7 @@ void clusterSlotStatsSetClusterMsgLength(uint32_t len);
 void clusterSlotStatsResetClusterMsgLength(void);
 
 /* network-bytes-out metric. */
+void clusterSlotStatsAddNetworkBytesOutForSlot(int slot, unsigned long long net_bytes_out);
 void clusterSlotStatsAddNetworkBytesOutForUserClient(client *c);
 void clusterSlotStatsIncrNetworkBytesOutForReplication(long long len);
 void clusterSlotStatsDecrNetworkBytesOutForReplication(long long len);

src/config.c

@@ -3262,6 +3262,9 @@ standardConfig static_configs[] = {
     createIntConfig("port", NULL, MODIFIABLE_CONFIG, 0, 65535, server.port, 6379, INTEGER_CONFIG, NULL, updatePort),                                               /* TCP port. */
     createIntConfig("io-threads", NULL, DEBUG_CONFIG | MODIFIABLE_CONFIG, 1, IO_THREADS_MAX_NUM, server.io_threads_num, 1, INTEGER_CONFIG, NULL, updateIOThreads), /* Single threaded by default */
     createIntConfig("events-per-io-thread", NULL, MODIFIABLE_CONFIG | HIDDEN_CONFIG, 0, INT_MAX, server.events_per_io_thread, 2, INTEGER_CONFIG, NULL, NULL),
+    createIntConfig("min-io-threads-avoid-copy-reply", NULL, MODIFIABLE_CONFIG | HIDDEN_CONFIG, 0, INT_MAX, server.min_io_threads_copy_avoid, 7, INTEGER_CONFIG, NULL, NULL),
+    createIntConfig("min-string-size-avoid-copy-reply", NULL, MODIFIABLE_CONFIG | HIDDEN_CONFIG, 0, INT_MAX, server.min_string_size_copy_avoid, 16384, INTEGER_CONFIG, NULL, NULL),
+    createIntConfig("min-string-size-avoid-copy-reply-threaded", NULL, MODIFIABLE_CONFIG | HIDDEN_CONFIG, 0, INT_MAX, server.min_string_size_copy_avoid_threaded, 65536, INTEGER_CONFIG, NULL, NULL),
     createIntConfig("prefetch-batch-max-size", NULL, MODIFIABLE_CONFIG, 0, 128, server.prefetch_batch_max_size, 16, INTEGER_CONFIG, NULL, NULL),
     createIntConfig("auto-aof-rewrite-percentage", NULL, MODIFIABLE_CONFIG, 0, INT_MAX, server.aof_rewrite_perc, 100, INTEGER_CONFIG, NULL, NULL),
     createIntConfig("cluster-replica-validity-factor", "cluster-slave-validity-factor", MODIFIABLE_CONFIG, 0, INT_MAX, server.cluster_replica_validity_factor, 10, INTEGER_CONFIG, NULL, NULL), /* replica max data age factor. */

src/io_threads.c

@@ -447,9 +447,14 @@ int trySendWriteToIOThreads(client *c) {
          * threads from reading data that might be invalid in their local CPU cache. */
         c->io_last_reply_block = listLast(c->reply);
         if (c->io_last_reply_block) {
-            c->io_last_bufpos = ((clientReplyBlock *)listNodeValue(c->io_last_reply_block))->used;
+            clientReplyBlock *block = (clientReplyBlock *)listNodeValue(c->io_last_reply_block);
+            c->io_last_bufpos = block->used;
+            /* If buffer is encoded force new header */
+            if (block->flag.buf_encoded) block->last_header = NULL;
         } else {
             c->io_last_bufpos = (size_t)c->bufpos;
+            /* If buffer is encoded force new header */
+            if (c->flag.buf_encoded) c->last_header = NULL;
         }
     }
 

src/memory_prefetch.c

@@ -10,6 +10,7 @@
 
 #include "memory_prefetch.h"
 #include "server.h"
+#include "io_threads.h"
 
 typedef enum {
     PREFETCH_ENTRY, /* Initial state, prefetch entries associated with the given key's hash */
@@ -120,6 +121,10 @@ static void prefetchEntry(KeyPrefetchInfo *info) {
     if (hashtableIncrementalFindStep(&info->hashtab_state) == 1) {
         /* Not done yet */
         moveToNextKey();
+    } else if (server.io_threads_num >= server.min_io_threads_copy_avoid) {
+        /* Copy avoidance should be more efficient without value prefetch
+         * starting certain number of I/O threads */
+        markKeyAsdone(info);
     } else {
         info->state = PREFETCH_VALUE;
     }