Execute MGET and MSET in batches

Signed-off-by: Viktor Söderqvist <[email protected]>

Author: zuiderkwast

src/memory_prefetch.c

@@ -282,13 +282,13 @@ void removeClientFromPendingCommandsBatch(client *c) {
 
 /* A simple way to prefetch the keys for a client command execution. This can be
  * used for optimizing multi-key commands like mget, mset, etc. */
-void prefetchKeys(client *c, int first, int step) {
+void prefetchKeys(client *c, int first, int step, int count) {
     /* Skip this for IO threads. Keys are already batch-prefetched. */
     if (c->io_parsed_cmd) return;
 
     int n = (c->argc - first) / step;
-    if (n < 2) return;  /* There's no benefit in prefetcing only one key. */
-    if (n > 32) n = 32; /* Prefetching too many keys doesn't help the cache. */
+    if (count < n) n = count;
+    if (n < 2) return; /* There's no benefit in prefetcing only one key. */
     hashtableIncrementalFindState states[n];
     int i, arg;
     for (i = 0, arg = first; i < n && arg < c->argc; i++, arg += step) {

src/memory_prefetch.h

@@ -8,6 +8,6 @@ void processClientsCommandsBatch(void);
 int addCommandToBatchAndProcessIfFull(struct client *c);
 void removeClientFromPendingCommandsBatch(struct client *c);
 
-void prefetchKeys(struct client *c, int first, int step);
+void prefetchKeys(struct client *c, int first, int step, int count);
 
 #endif /* MEMORY_PREFETCH_H */

src/t_string.c

@@ -606,10 +606,14 @@ void getrangeCommand(client *c) {
 }
 
 void mgetCommand(client *c) {
+    const int batch_size = 16;
     int j;
-    prefetchKeys(c, 1, 1);
     addReplyArrayLen(c, c->argc - 1);
     for (j = 1; j < c->argc; j++) {
+        if (j % batch_size == 1) {
+            /* Prefetch the next batch of keys */
+            prefetchKeys(c, j, 1, batch_size);
+        }
         robj *o = lookupKeyRead(c->db, c->argv[j]);
         if (o == NULL) {
             addReplyNull(c);
@@ -624,20 +628,22 @@ void mgetCommand(client *c) {
 }
 
 void msetGenericCommand(client *c, int nx) {
+    const int batch_size = 16;
     int j;
 
     if ((c->argc % 2) == 0) {
         addReplyErrorArity(c);
         return;
     }
 
-    /* Prefetch the keys from memory in parallel. */
-    prefetchKeys(c, 1, 2);
-
     /* Handle the NX flag. The MSETNX semantic is to return zero and don't
      * set anything if at least one key already exists. */
     if (nx) {
         for (j = 1; j < c->argc; j += 2) {
+            if (j % (batch_size * 2) == 1) {
+                /* Prefetch the next batch of keys */
+                prefetchKeys(c, j, 2, batch_size);
+            }
             if (lookupKeyWrite(c->db, c->argv[j]) != NULL) {
                 addReply(c, shared.czero);
                 return;
@@ -647,6 +653,10 @@ void msetGenericCommand(client *c, int nx) {
 
     int setkey_flags = nx ? SETKEY_DOESNT_EXIST : 0;
     for (j = 1; j < c->argc; j += 2) {
+        if (!nx && j % (batch_size * 2) == 1) {
+            /* Prefetch the next batch of keys */
+            prefetchKeys(c, j, 2, batch_size);
+        }
         robj *val = tryObjectEncoding(c->argv[j + 1]);
         setKey(c, c->db, c->argv[j], &val, setkey_flags);
         incrRefCount(val);