MOD-8009: Allow Users To Configure Cursor Index Limitation Through Global Config (#5137)

kei-nan · GuyAv46 · oshadmi · web-flow · commit 602fe7685da6 · 2024-11-05T13:58:15.000Z
* * move cursor limitation per index to be a configuration parameter * solve a bug where the coordinator would stop reading from internal cursors * * remove int return type for iterate callback functions, since their return value is ignored and has no real meaning. * * Code Review - Round #1 * * Code Review - Round #1, missing file change * * fix crash in testAggregate.py * * Code Review - Round #2 * * replace comma with a period * * if we somehow ended up with a closed channel, free the reply * Update src/config.c Co-authored-by: GuyAv46 <47632673+GuyAv46@users.noreply.github.com> * Update src/cursor.c Co-authored-by: Omer Shadmi <76992134+oshadmi@users.noreply.github.com> * - fix channel test * - add logging for the broadcasting failure flow * - changed broadcast failure to be verbose instead of warning * Update tests/pytests/test_cursors.py Co-authored-by: Omer Shadmi <76992134+oshadmi@users.noreply.github.com> --------- Co-authored-by: GuyAv46 <47632673+GuyAv46@users.noreply.github.com> Co-authored-by: Omer Shadmi <76992134+oshadmi@users.noreply.github.com>
diff --git a/src/config.c b/src/config.c
@@ -636,6 +636,16 @@ CONFIG_GETTER(getBGIndexSleepGap) {
 CONFIG_BOOLEAN_SETTER(set_PrioritizeIntersectUnionChildren, prioritizeIntersectUnionChildren)
 CONFIG_BOOLEAN_GETTER(get_PrioritizeIntersectUnionChildren, prioritizeIntersectUnionChildren, 0)
 
+CONFIG_SETTER(setIndexCursorLimit) {
+  int acrc = AC_GetLongLong(ac, &config->indexCursorLimit, AC_F_GE0);
+  RETURN_STATUS(acrc);
+}
+
+CONFIG_GETTER(getIndexCursorLimit) {
+  sds ss = sdsempty();
+  return sdscatprintf(ss, "%lld", config->indexCursorLimit);
+}
+
 RSConfig RSGlobalConfig = RS_DEFAULT_CONFIG;
 
 static RSConfigVar *findConfigVar(const RSConfigOptions *config, const char *name) {
@@ -840,6 +850,10 @@ RSConfigOptions RSGlobalConfigOptions = {
                      "high memory consumption.",
          .setValue = setCursorMaxIdle,
          .getValue = getCursorMaxIdle},
+        {.name = "INDEX_CURSOR_LIMIT",
+         .helpText = "Max number of cursors for a given index that can be opened inside of a shard. Default is 128",
+         .setValue = setIndexCursorLimit,
+         .getValue = getIndexCursorLimit},
         {.name = "NO_MEM_POOLS",
          .helpText = "Set RediSearch to run without memory pools",
          .setValue = setNoMemPools,
diff --git a/src/config.h b/src/config.h
@@ -139,6 +139,8 @@ typedef struct {
   // If set, we use an optimization that sorts the children of an intersection iterator in a way
   // where union iterators are being factorize by the number of their own children.
   int prioritizeIntersectUnionChildren;
+  // Limit the number of cursors that can be created for a single index
+  long long indexCursorLimit;
 } RSConfig;
 
 typedef enum {
@@ -217,6 +219,7 @@ void UpgradeDeprecatedMTConfigs();
 #define GC_SCANSIZE 100
 #define DEFAULT_MIN_PHONETIC_TERM_LEN 3
 #define DEFAULT_FORK_GC_RUN_INTERVAL 30
+#define DEFAULT_INDEX_CURSOR_LIMIT 128
 #define SEARCH_REQUEST_RESULTS_MAX 1000000
 #define NR_MAX_DEPTH_BALANCE 2
 #define VECSIM_DEFAULT_BLOCK_SIZE   1024
@@ -262,7 +265,8 @@ void UpgradeDeprecatedMTConfigs();
     .vssMaxResize = 0,                                                                                                \
     .multiTextOffsetDelta = 100,                                                                                      \
     .numBGIndexingIterationsBeforeSleep = 100,                                                                        \
-    .prioritizeIntersectUnionChildren = false                                                                         \
+    .prioritizeIntersectUnionChildren = false,                                                                        \
+    .indexCursorLimit = DEFAULT_INDEX_CURSOR_LIMIT                                                                    \
   }
 
 #define REDIS_ARRAY_LIMIT 7
diff --git a/src/coord/dist_aggregate.c b/src/coord/dist_aggregate.c
@@ -16,18 +16,13 @@
 #include "resp3.h"
 #include "coord/config.h"
 #include "dist_profile.h"
+#include "util/misc.h"
 
 #include <err.h>
 
 // Get cursor command using a cursor id and an existing aggregate command
 // Returns true if the cursor is not done (i.e., not depleted)
-static bool getCursorCommand(MRReply *res, MRCommand *cmd, MRIteratorCtx *ctx) {
-  long long cursorId;
-  if (!MRReply_ToInteger(MRReply_ArrayElement(res, 1), &cursorId)) {
-    // Invalid format?!
-    return false;
-  }
-
+static bool getCursorCommand(long long cursorId, MRCommand *cmd, MRIteratorCtx *ctx) {
   if (cursorId == 0) {
     // Cursor was set to 0, end of reply chain. cmd->depleted will be set in `MRIteratorCallback_Done`.
     return false;
@@ -73,7 +68,7 @@ static bool getCursorCommand(MRReply *res, MRCommand *cmd, MRIteratorCtx *ctx) {
 }
 
 
-static int netCursorCallback(MRIteratorCallbackCtx *ctx, MRReply *rep) {
+static void netCursorCallback(MRIteratorCallbackCtx *ctx, MRReply *rep) {
   MRCommand *cmd = MRIteratorCallback_GetCommand(ctx);
 
   // If the root command of this reply is a DEL command, we don't want to
@@ -85,23 +80,25 @@ static int netCursorCallback(MRIteratorCallbackCtx *ctx, MRReply *rep) {
     // Discard the response, and return REDIS_OK
     MRIteratorCallback_Done(ctx, MRReply_Type(rep) == MR_REPLY_ERROR);
     MRReply_Free(rep);
-    return REDIS_OK;
+    return;
   }
 
   // Check if an error returned from the shard
   if (MRReply_Type(rep) == MR_REPLY_ERROR) {
-    RedisModule_Log(RSDummyContext, "notice", "Coordinator got an error from a shard");
-    RedisModule_Log(RSDummyContext, "verbose", "Shard error: %s", MRReply_String(rep, NULL));
+    const char* error = MRReply_String(rep, NULL);
+    RedisModule_Log(RSDummyContext, "notice", "Coordinator got an error '%.*s' from a shard", GetRedisErrorCodeLength(error), error);
+    RedisModule_Log(RSDummyContext, "verbose", "Shard error: %s", error);
     MRIteratorCallback_AddReply(ctx, rep); // to be picked up by getNextReply
     MRIteratorCallback_Done(ctx, 1);
-    return REDIS_ERR;
+    return;
   }
 
+  const bool isResp3 = cmd->protocol == 3;
   bool bail_out = MRReply_Type(rep) != MR_REPLY_ARRAY;
 
   if (!bail_out) {
     size_t len = MRReply_Length(rep);
-    if (cmd->protocol == 3) {
+    if (isResp3) {
       bail_out = len != 2; // (map, cursor)
       if (bail_out) {
         RedisModule_Log(RSDummyContext, "warning", "Expected reply of length 2, got %ld", len);
@@ -118,15 +115,17 @@ static int netCursorCallback(MRIteratorCallbackCtx *ctx, MRReply *rep) {
     RedisModule_Log(RSDummyContext, "warning", "An unexpected reply was received from a shard");
     MRReply_Free(rep);
     MRIteratorCallback_Done(ctx, 1);
-    return REDIS_ERR;
+    return;
   }
 
-  // rewrite and resend the cursor command if needed
-  int rc = REDIS_OK;
-  bool done = !getCursorCommand(rep, cmd, MRIteratorCallback_GetCtx(ctx));
+  long long cursorId;
+  MRReply* cursor = MRReply_ArrayElement(rep, 1);
+  if (!MRReply_ToInteger(cursor, &cursorId)) {
+    cursorId = 0;
+  }
 
   // Push the reply down the chain
-  if (cmd->protocol == 3) // RESP3
+  if (isResp3) // RESP3
   {
     MRReply *map = MRReply_ArrayElement(rep, 0);
     MRReply *results = NULL;
@@ -137,11 +136,7 @@ static int netCursorCallback(MRIteratorCallbackCtx *ctx, MRReply *rep) {
         MRIteratorCallback_AddReply(ctx, rep); // to be picked up by getNextReply
         // User code now owns the reply, so we can't free it here ourselves!
         rep = NULL;
-      } else {
-        done = true;
       }
-    } else {
-      done = true;
     }
   }
   else // RESP2
@@ -151,28 +146,23 @@ static int netCursorCallback(MRIteratorCallbackCtx *ctx, MRReply *rep) {
       MRIteratorCallback_AddReply(ctx, rep); // to be picked up by getNextReply
       // User code now owns the reply, so we can't free it here ourselves!
       rep = NULL;
-    } else {
-      done = true;
     }
   }
 
-  if (done) {
+  // rewrite and resend the cursor command if needed
+  // should only be determined based on the cursor and not on the set of results we get
+  if (!getCursorCommand(cursorId, cmd, MRIteratorCallback_GetCtx(ctx))) {
     MRIteratorCallback_Done(ctx, 0);
   } else if (cmd->forCursor) {
     MRIteratorCallback_ProcessDone(ctx);
-  } else {
-    // resend command
-    if (MRIteratorCallback_ResendCommand(ctx) == REDIS_ERR) {
-      MRIteratorCallback_Done(ctx, 1);
-      rc = REDIS_ERR;
-    }
+  } else if (MRIteratorCallback_ResendCommand(ctx) == REDIS_ERR) {
+    MRIteratorCallback_Done(ctx, 1);
   }
 
   if (rep != NULL) {
     // If rep has been set to NULL, it means the callback has been invoked
     MRReply_Free(rep);
   }
-  return rc;
 }
 
 RSValue *MRReply_ToValue(MRReply *r) {
diff --git a/src/coord/rmr/chan.c b/src/coord/rmr/chan.c
@@ -73,12 +73,12 @@ size_t MRChannel_Size(MRChannel *chan) {
   return ret;
 }
 
-int MRChannel_Push(MRChannel *chan, void *ptr) {
+PushErrorMask MRChannel_Push(MRChannel *chan, void *ptr) {
 
   pthread_mutex_lock(&chan->lock);
-  int rc = 1;
+  int rc = 0;
   if (!chan->open) {
-    rc = 0;
+    rc = CHANNEL_CLOSED;
     goto end;
   }
 
@@ -95,7 +95,7 @@ int MRChannel_Push(MRChannel *chan, void *ptr) {
   }
   chan->size++;
 end:
-  if (pthread_cond_broadcast(&chan->cond)) rc = 0;
+  if (pthread_cond_broadcast(&chan->cond)) rc |= BROADCAST_FAILURE;
   pthread_mutex_unlock(&chan->lock);
   return rc;
 }
diff --git a/src/coord/rmr/chan.h b/src/coord/rmr/chan.h
@@ -15,8 +15,14 @@ typedef struct MRChannel MRChannel;
 
 extern void *MRCHANNEL_CLOSED;
 
+typedef enum {
+  CHANNEL_CLOSED = 0x1,
+  BROADCAST_FAILURE = 0x2
+} PushErrorMask;
+
+
 MRChannel *MR_NewChannel();
-int MRChannel_Push(MRChannel *chan, void *ptr);
+PushErrorMask MRChannel_Push(MRChannel *chan, void *ptr);
 /* Pop an item, wait indefinitely or until the channel is closed for an item.
  * Return MRCHANNEL_CLOSED if the channel is closed*/
 void *MRChannel_Pop(MRChannel *chan);
diff --git a/src/coord/rmr/rmr.c b/src/coord/rmr/rmr.c
@@ -529,7 +529,15 @@ MRIteratorCtx *MRIteratorCallback_GetCtx(MRIteratorCallbackCtx *ctx) {
 }
 
 void MRIteratorCallback_AddReply(MRIteratorCallbackCtx *ctx, MRReply *rep) {
-  MRChannel_Push(ctx->ic->chan, rep);
+  PushErrorMask mask = MRChannel_Push(ctx->ic->chan, rep);
+  if (mask & CHANNEL_CLOSED) {
+    // We are using a verbose log level to prevent flodding the logs with this message in production
+    RedisModule_Log(RSDummyContext, "verbose", "Tried pushing a reply to a closed channel");
+    MRReply_Free(rep);
+  }
+  if (mask & BROADCAST_FAILURE) {
+    RedisModule_Log(RSDummyContext, "verbose", "Failed broadcasting the reply in the channel");
+  }
 }
 
 void iterStartCb(void *p) {
diff --git a/src/coord/rmr/rmr.h b/src/coord/rmr/rmr.h
@@ -72,7 +72,7 @@ typedef struct MRIteratorCallbackCtx MRIteratorCallbackCtx;
 typedef struct MRIteratorCtx MRIteratorCtx;
 typedef struct MRIterator MRIterator;
 
-typedef int (*MRIteratorCallback)(MRIteratorCallbackCtx *ctx, MRReply *rep);
+typedef void (*MRIteratorCallback)(MRIteratorCallbackCtx *ctx, MRReply *rep);
 
 // Trigger all the commands in the iterator to be sent.
 // Returns true if there may be more replies to come, false if we are done.
diff --git a/src/cursor.c b/src/cursor.c
@@ -206,16 +206,14 @@ Cursor *Cursors_Reserve(CursorList *cl, StrongRef global_spec_ref, unsigned inte
   // If the cursor should be associated with a spec,
   // we assume that global_spec_ref points to a valid spec, else the function returns NULL.
   IndexSpec *spec = StrongRef_Get(global_spec_ref);
-
   // If we are in a coordinator ctx, the spec is NULL
-  if (spec && (spec->activeCursors >= spec->cursorsCap)) {
+  if (spec && spec->activeCursors >= RSGlobalConfig.indexCursorLimit) {
     /** Collect idle cursors now */
     Cursors_GCInternal(cl, 0);
-    if (spec->activeCursors >= spec->cursorsCap) {
-      QueryError_SetError(status, QUERY_ELIMIT, "Too many cursors allocated for index");
+    if (spec->activeCursors >= RSGlobalConfig.indexCursorLimit) {
+      QueryError_SetErrorFmt(status, QUERY_ELIMIT, "INDEX_CURSOR_LIMIT of %lld has been reached for an index", RSGlobalConfig.indexCursorLimit);
       goto done;
     }
-
   }
 
   cur = rm_calloc(1, sizeof(*cur));
@@ -313,7 +311,7 @@ void Cursors_RenderStats(CursorList *cl, CursorList *cl_coord, IndexSpec *spec,
     RedisModule_ReplyKV_LongLong(reply, "global_idle", ARRAY_GETSIZE_AS(&cl->idle, Cursor **) +
                                                         ARRAY_GETSIZE_AS(&cl_coord->idle, Cursor **));
     RedisModule_ReplyKV_LongLong(reply, "global_total", kh_size(cl->lookup) + kh_size(cl_coord->lookup));
-    RedisModule_ReplyKV_LongLong(reply, "index_capacity", spec->cursorsCap);
+    RedisModule_ReplyKV_LongLong(reply, "index_capacity", RSGlobalConfig.indexCursorLimit);
     RedisModule_ReplyKV_LongLong(reply, "index_total", spec->activeCursors);
 
   RedisModule_Reply_MapEnd(reply);
@@ -330,7 +328,7 @@ void Cursors_RenderStatsForInfo(CursorList *cl, CursorList *cl_coord, IndexSpec
   RedisModule_InfoAddFieldLongLong(ctx, "global_idle", ARRAY_GETSIZE_AS(&cl->idle, Cursor **) +
                                                         ARRAY_GETSIZE_AS(&cl_coord->idle, Cursor **));
   RedisModule_InfoAddFieldLongLong(ctx, "global_total", kh_size(cl->lookup) + kh_size(cl_coord->lookup));
-  RedisModule_InfoAddFieldLongLong(ctx, "index_capacity", spec->cursorsCap);
+  RedisModule_InfoAddFieldLongLong(ctx, "index_capacity", RSGlobalConfig.indexCursorLimit);
   RedisModule_InfoAddFieldLongLong(ctx, "index_total", spec->activeCursors);
   RedisModule_InfoEndDictField(ctx);
 
diff --git a/src/cursor.h b/src/cursor.h
@@ -136,7 +136,6 @@ void CursorList_Empty(CursorList *cl);
  */
 void CursorList_Expire(CursorList *cl);
 
-#define RSCURSORS_DEFAULT_CAPACITY 128
 #define RSCURSORS_SWEEP_INTERVAL 500                /* GC Every 500 requests */
 #define RSCURSORS_SWEEP_THROTTLE (1 * (1000000000)) /* Throttle, in NS */
 
diff --git a/src/spec.c b/src/spec.c
@@ -86,9 +86,8 @@ static void setMemoryInfo(RedisModuleCtx *ctx) {
  * Initialize the spec's fields that are related to the cursors.
  */
 
-static void Cursors_initSpec(IndexSpec *spec, size_t capacity) {
+static void Cursors_initSpec(IndexSpec *spec) {
   spec->activeCursors = 0;
-  spec->cursorsCap = capacity;
 }
 
 /*
@@ -405,7 +404,7 @@ IndexSpec *IndexSpec_CreateNew(RedisModuleCtx *ctx, RedisModuleString **argv, in
   // Start the garbage collector
   IndexSpec_StartGC(ctx, spec_ref, sp);
 
-  Cursors_initSpec(sp, RSCURSORS_DEFAULT_CAPACITY);
+  Cursors_initSpec(sp);
 
   // Create the indexer
   sp->indexer = NewIndexer(sp);
@@ -2568,7 +2567,7 @@ int IndexSpec_CreateFromRdb(RedisModuleCtx *ctx, RedisModuleIO *rdb, int encver,
   sp->uniqueId = spec_unique_ids++;
 
   IndexSpec_StartGC(ctx, spec_ref, sp);
-  Cursors_initSpec(sp, RSCURSORS_DEFAULT_CAPACITY);
+  Cursors_initSpec(sp);
 
   if (sp->flags & Index_HasSmap) {
     sp->smap = SynonymMap_RdbLoad(rdb, encver);
@@ -2724,7 +2723,7 @@ void *IndexSpec_LegacyRdbLoad(RedisModuleIO *rdb, int encver) {
 
   // start the gc and add the spec to the cursor list
   IndexSpec_StartGC(RSDummyContext, spec_ref, sp);
-  Cursors_initSpec(sp, RSCURSORS_DEFAULT_CAPACITY);
+  Cursors_initSpec(sp);
 
   dictAdd(legacySpecDict, sp->name, spec_ref.rm);
   return spec_ref.rm;
diff --git a/src/spec.h b/src/spec.h
@@ -326,7 +326,6 @@ typedef struct IndexSpec {
   pthread_rwlock_t rwlock;
 
   // Cursors counters
-  size_t cursorsCap;
   size_t activeCursors;
 
   // Quick access to the spec's strong ref
diff --git a/src/util/misc.c b/src/util/misc.c
@@ -7,6 +7,7 @@
 #include "misc.h"
 #include <stdlib.h>
 #include <ctype.h>
+#include <string.h>
 
 void GenericAofRewrite_DisabledHandler(RedisModuleIO *aof, RedisModuleString *key, void *value) {
   RedisModule_Log(RedisModule_GetContextFromIO(aof), "error",
@@ -22,3 +23,8 @@ char *strtolower(char *str) {
   }
   return str;
 }
+
+int GetRedisErrorCodeLength(const char* error) {
+  const char* errorSpace = strchr(error, ' ');
+  return errorSpace ? errorSpace - error : 0;
+}
diff --git a/src/util/misc.h b/src/util/misc.h
@@ -16,4 +16,6 @@ void GenericAofRewrite_DisabledHandler(RedisModuleIO *aof, RedisModuleString *ke
 
 char *strtolower(char *str);
 
-#endif
+int GetRedisErrorCodeLength(const char* error);
+
+#endif
diff --git a/tests/ctests/coord_tests/test_chan.c b/tests/ctests/coord_tests/test_chan.c
@@ -17,8 +17,8 @@ void testChan() {
   for (int i = 0; i < 100; i++) {
     int *ptr = rm_malloc(sizeof(*ptr));
     *ptr = i;
-    int rc = MRChannel_Push(c, ptr);
-    mu_assert_int_eq(1, rc);
+    PushErrorMask mask = MRChannel_Push(c, ptr);
+    mu_assert_int_eq(0, mask);
     mu_assert_int_eq(i + 1, MRChannel_Size(c));
   }
 
diff --git a/tests/ctests/test_error_parsing.c b/tests/ctests/test_error_parsing.c
diff --git a/tests/pytests/test_config.py b/tests/pytests/test_config.py
diff --git a/tests/pytests/test_cursors.py b/tests/pytests/test_cursors.py

Original file line number	Diff line number	Diff line change
`@@ -73,12 +73,12 @@ size_t MRChannel_Size(MRChannel *chan) {`
`73`	`73`	`return ret;`
`74`	`74`	`}`
`75`	`75`
`76`		`-int MRChannel_Push(MRChannel chan, void ptr) {`
	`76`	`+PushErrorMask MRChannel_Push(MRChannel chan, void ptr) {`
`77`	`77`
`78`	`78`	`pthread_mutex_lock(&chan->lock);`
`79`		`- int rc = 1;`
	`79`	`+ int rc = 0;`
`80`	`80`	`if (!chan->open) {`
`81`		`- rc = 0;`
	`81`	`+ rc = CHANNEL_CLOSED;`
`82`	`82`	`goto end;`
`83`	`83`	`}`
`84`	`84`
`@@ -95,7 +95,7 @@ int MRChannel_Push(MRChannel chan, void ptr) {`
`95`	`95`	`}`
`96`	`96`	`chan->size++;`
`97`	`97`	`end:`
`98`		`- if (pthread_cond_broadcast(&chan->cond)) rc = 0;`
	`98`	`+ if (pthread_cond_broadcast(&chan->cond)) rc \|= BROADCAST_FAILURE;`
`99`	`99`	`pthread_mutex_unlock(&chan->lock);`
`100`	`100`	`return rc;`
`101`	`101`	`}`