[2.6] MOD-8009: Allow Users To Configure Cursor Index Limitation Through Global Config (#5168)

kei-nan · web-flow · commit b86fcdaf35ec · 2024-11-06T11:39:04.000Z
* MOD-8009: Allow Users To Configure Cursor Index Limitation Through Global Config (#5137) * * 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> (cherry picked from commit 602fe76) * * use RSGlobalConfig.indexCursorLimit * * apply missing cherry-pick changes * * fix netCursorCallback callback return type * * code review comments + test crash fix * * fix remaining code review comments
diff --git a/coord/src/dist_aggregate.c b/coord/src/dist_aggregate.c
@@ -18,12 +18,7 @@
 #include <err.h>
 
 /* Get cursor command using a cursor id and an existing aggregate command */
-static int getCursorCommand(MRReply *prev, MRCommand *cmd) {
-  long long cursorId;
-  if (!MRReply_ToInteger(MRReply_ArrayElement(prev, 1), &cursorId)) {
-    // Invalid format?!
-    return 0;
-  }
+static int getCursorCommand(long long cursorId, MRCommand *cmd) {
   if (cursorId == 0) {
     // Cursor was set to 0, end of reply chain.
     cmd->depleted = true;
@@ -46,7 +41,7 @@ static int getCursorCommand(MRReply *prev, MRCommand *cmd) {
   return 1;
 }
 
-static int netCursorCallback(MRIteratorCallbackCtx *ctx, MRReply *rep, MRCommand *cmd) {
+static void netCursorCallback(MRIteratorCallbackCtx *ctx, MRReply *rep, MRCommand *cmd) {
   // Should we assert this??
   if (!rep || MRReply_Type(rep) != MR_REPLY_ARRAY ||
              (MRReply_Length(rep) != 2 && MRReply_Length(rep) != 3)) {
@@ -56,40 +51,37 @@ static int netCursorCallback(MRIteratorCallbackCtx *ctx, MRReply *rep, MRCommand
     MRReply_Free(rep);
     MRIteratorCallback_Done(ctx, 1);
     RedisModule_Log(NULL, "warning", "An empty reply was received from a shard");
-    return REDIS_ERR;
+    return;
   }
 
-  // rewrite and resend the cursor command if needed
-  int rc = REDIS_OK;
-  int isDone = !getCursorCommand(rep, cmd);
-
+  long long cursorId;
+  MRReply* cursor = MRReply_ArrayElement(rep, 1);
+  if (!MRReply_ToInteger(cursor, &cursorId)) {
+    cursorId = 0;
+  }
   // Push the reply down the chain
   MRReply *arr = MRReply_ArrayElement(rep, 0);
   if (arr && MRReply_Type(arr) == MR_REPLY_ARRAY && MRReply_Length(arr) > 1) {
     MRIteratorCallback_AddReply(ctx, rep);
     // User code now owns the reply, so we can't free it here ourselves!
     rep = NULL;
-  } else {
-    isDone = 1;
   }
 
-  if (isDone) {
+  if (!getCursorCommand(cursorId, cmd)) {
     MRIteratorCallback_Done(ctx, 0);
   } else if (cmd->forCursor) {
     MRIteratorCallback_ProcessDone(ctx);
   } else {
     // resend command
     if (REDIS_ERR == MRIteratorCallback_ResendCommand(ctx, cmd)) {
       MRIteratorCallback_Done(ctx, 1);
-      rc = REDIS_ERR;
     }
   }
 
   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/coord/src/module.c b/coord/src/module.c
@@ -1988,7 +1988,7 @@ static void addIndexCursor(const IndexSpec *sp) {
   char *end = strchr(s, '{');
   if (end) {
     *end = '\0';
-    CursorList_AddSpec(&RSCursorsCoord, s, RSCURSORS_DEFAULT_CAPACITY);
+    CursorList_AddSpec(&RSCursorsCoord, s);
   }
   rm_free(s);
 }
diff --git a/coord/src/rmr/chan.c b/coord/src/rmr/chan.c
@@ -82,12 +82,12 @@ size_t MRChannel_MaxSize(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 || (chan->maxSize > 0 && chan->size == chan->maxSize)) {
-    rc = 0;
+    rc = CHANNEL_CLOSED;
     goto end;
   }
 
@@ -104,7 +104,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/coord/src/rmr/chan.h b/coord/src/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(size_t max);
-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/coord/src/rmr/rmr.c b/coord/src/rmr/rmr.c
@@ -591,8 +591,16 @@ int MRIteratorCallback_Done(MRIteratorCallbackCtx *ctx, int error) {
   return 1;
 }
 
-int MRIteratorCallback_AddReply(MRIteratorCallbackCtx *ctx, MRReply *rep) {
-  return MRChannel_Push(ctx->ic->chan, rep);
+void MRIteratorCallback_AddReply(MRIteratorCallbackCtx *ctx, MRReply *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/coord/src/rmr/rmr.h b/coord/src/rmr/rmr.h
@@ -73,7 +73,7 @@ extern void *MRITERATOR_DONE;
 typedef struct MRIteratorCallbackCtx MRIteratorCallbackCtx;
 typedef struct MRIterator MRIterator;
 
-typedef int (*MRIteratorCallback)(MRIteratorCallbackCtx *ctx, MRReply *rep, MRCommand *cmd);
+typedef void (*MRIteratorCallback)(MRIteratorCallbackCtx *ctx, MRReply *rep, MRCommand *cmd);
 
 // 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.
@@ -83,7 +83,7 @@ MRReply *MRIterator_Next(MRIterator *it);
 
 MRIterator *MR_Iterate(MRCommandGenerator cg, MRIteratorCallback cb);
 
-int MRIteratorCallback_AddReply(MRIteratorCallbackCtx *ctx, MRReply *rep);
+void MRIteratorCallback_AddReply(MRIteratorCallbackCtx *ctx, MRReply *rep);
 
 int MRIteratorCallback_Done(MRIteratorCallbackCtx *ctx, int error);
 
diff --git a/coord/src/rmr/test/test_chan.c b/coord/src/rmr/test/test_chan.c
@@ -18,8 +18,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/src/config.c b/src/config.c
@@ -539,6 +539,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) {
@@ -723,6 +733,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
@@ -127,6 +127,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 {
@@ -197,6 +199,7 @@ void DialectsGlobalStats_AddToInfo(RedisModuleInfoCtx *ctx);
 #define DEFAULT_MIN_PHONETIC_TERM_LEN 3
 #define DEFAULT_FORK_GC_RUN_INTERVAL 30
 #define DEFAULT_MAX_RESULTS_TO_UNSORTED_MODE 1000
+#define DEFAULT_INDEX_CURSOR_LIMIT 128
 #define SEARCH_REQUEST_RESULTS_MAX 1000000
 #define NR_MAX_DEPTH_BALANCE 2
 #define MIN_DIALECT_VERSION 1 // MIN_DIALECT_VERSION is expected to change over time as dialects become deprecated.
@@ -223,7 +226,8 @@ void DialectsGlobalStats_AddToInfo(RedisModuleInfoCtx *ctx);
     .forkGCCleanNumericEmptyNodes = true, .freeResourcesThread = true, .defaultDialectVersion = 1,\
     .vssMaxResize = 0, .multiTextOffsetDelta = 100, .used_dialects = 0,                           \
     .numBGIndexingIterationsBeforeSleep = 100,                                                    \
-    .prioritizeIntersectUnionChildren = false \
+    .prioritizeIntersectUnionChildren = false,                                                                        \
+    .indexCursorLimit = DEFAULT_INDEX_CURSOR_LIMIT                                                                    \
   }
 
 #define REDIS_ARRAY_LIMIT 7
diff --git a/src/cursor.c b/src/cursor.c
@@ -144,15 +144,14 @@ int Cursors_CollectIdle(CursorList *cl) {
   return rc;
 }
 
-void CursorList_AddSpec(CursorList *cl, const char *k, size_t capacity) {
+void CursorList_AddSpec(CursorList *cl, const char *k) {
   CursorSpecInfo *info = findInfo(cl, k);
   if (!info) {
     info = rm_malloc(sizeof(*info));
     info->keyName = rm_strdup(k);
     info->used = 0;
     dictAdd(cl->specsDict, (void *)k, info);
   }
-  info->cap = capacity;
 }
 
 void CursorList_RemoveSpec(CursorList *cl, const char *k) {
@@ -204,10 +203,10 @@ Cursor *Cursors_Reserve(CursorList *cl, const char *lookupName, unsigned interva
     goto done;
   }
 
-  if (spec->used >= spec->cap) {
+  if (spec->used >= RSGlobalConfig.indexCursorLimit) {
     /** Collect idle cursors now */
     Cursors_GCInternal(cl, 0);
-    if (spec->used >= spec->cap) {
+    if (spec->used >= RSGlobalConfig.indexCursorLimit) {
       QueryError_SetError(status, QUERY_ELIMIT, "Too many cursors allocated for index");
       goto done;
     }
@@ -314,7 +313,7 @@ void Cursors_RenderStats(CursorList *cl, CursorList *cl_coord, const char *name,
   RedisModule_ReplyWithLongLong(ctx, kh_size(cl->lookup));
 
   RedisModule_ReplyWithSimpleString(ctx, "index_capacity");
-  RedisModule_ReplyWithLongLong(ctx, info->cap + (info_coord ? info_coord->cap : 0));
+  RedisModule_ReplyWithLongLong(ctx, RSGlobalConfig.indexCursorLimit);
 
   RedisModule_ReplyWithSimpleString(ctx, "index_total");
   RedisModule_ReplyWithLongLong(ctx, info->used + (info_coord ? info_coord->used : 0));
@@ -333,7 +332,7 @@ void Cursors_RenderStatsForInfo(CursorList *cl, CursorList *cl_coord, const char
   RedisModule_InfoBeginDictField(ctx, "cursor_stats");
   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", info->cap +  + (info_coord ? info_coord->cap : 0));
+  RedisModule_InfoAddFieldLongLong(ctx, "index_capacity", RSGlobalConfig.indexCursorLimit);
   RedisModule_InfoAddFieldLongLong(ctx, "index_total", info->used + (info_coord ? info_coord->used : 0));
   RedisModule_InfoEndDictField(ctx);
 
diff --git a/src/cursor.h b/src/cursor.h
@@ -15,7 +15,6 @@
 
 typedef struct {
   char *keyName; /** Name of the key that refers to the spec */
-  size_t cap;    /** Maximum number of cursors for the spec */
   size_t used;   /** Number of cursors currently open */
 } CursorSpecInfo;
 
@@ -143,15 +142,14 @@ void CursorList_Empty(CursorList *cl, bool coord);
  */
 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 */
 
 /**
  * Add an index spec to the cursor list. This has the effect of adding the
  * spec (via its key) along with its capacity
  */
-void CursorList_AddSpec(CursorList *cl, const char *k, size_t capacity);
+void CursorList_AddSpec(CursorList *cl, const char *k);
 
 void CursorList_RemoveSpec(CursorList *cl, const char *k);
 
diff --git a/src/spec.c b/src/spec.c
@@ -329,8 +329,8 @@ IndexSpec *IndexSpec_CreateNew(RedisModuleCtx *ctx, RedisModuleString **argv, in
   // Start the garbage collector
   IndexSpec_StartGC(ctx, sp, GC_DEFAULT_HZ);
 
-  CursorList_AddSpec(&RSCursors, sp->name, RSCURSORS_DEFAULT_CAPACITY);
-  CursorList_AddSpec(&RSCursorsCoord, sp->name, RSCURSORS_DEFAULT_CAPACITY);
+  CursorList_AddSpec(&RSCursors, sp->name);
+  CursorList_AddSpec(&RSCursorsCoord, sp->name);
 
   // Create the indexer
   sp->indexer = NewIndexer(sp);
@@ -2258,8 +2258,8 @@ IndexSpec *IndexSpec_CreateFromRdb(RedisModuleCtx *ctx, RedisModuleIO *rdb, int
 
   IndexSpec_StartGC(ctx, sp, GC_DEFAULT_HZ);
   RedisModuleString *specKey = RedisModule_CreateStringPrintf(ctx, INDEX_SPEC_KEY_FMT, sp->name);
-  CursorList_AddSpec(&RSCursors, sp->name, RSCURSORS_DEFAULT_CAPACITY);
-  CursorList_AddSpec(&RSCursorsCoord, sp->name, RSCURSORS_DEFAULT_CAPACITY);
+  CursorList_AddSpec(&RSCursors, sp->name);
+  CursorList_AddSpec(&RSCursorsCoord, sp->name);
   RedisModule_FreeString(ctx, specKey);
 
   if (sp->flags & Index_HasSmap) {
@@ -2406,8 +2406,8 @@ void *IndexSpec_LegacyRdbLoad(RedisModuleIO *rdb, int encver) {
 
   // start the gc and add the spec to the cursor list
   IndexSpec_StartGC(RSDummyContext, sp, GC_DEFAULT_HZ);
-  CursorList_AddSpec(&RSCursors, sp->name, RSCURSORS_DEFAULT_CAPACITY);
-  CursorList_AddSpec(&RSCursorsCoord, sp->name, RSCURSORS_DEFAULT_CAPACITY);
+  CursorList_AddSpec(&RSCursors, sp->name);
+  CursorList_AddSpec(&RSCursorsCoord, sp->name);
 
   dictAdd(legacySpecDict, sp->name, sp);
   return sp;
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/test_error_parsing.c b/tests/ctests/test_error_parsing.c
@@ -0,0 +1,33 @@
+#include "test_util.h"
+#include "src/util/misc.h"
+#include <string.h>
+
+int testErrorCodeLengthExtraction() {
+  ASSERT_EQUAL(GetRedisErrorCodeLength("ERR Error message"), 3);
+  ASSERT_EQUAL(GetRedisErrorCodeLength("ERR"), 0);
+  ASSERT_EQUAL(GetRedisErrorCodeLength("ERR "), 3);
+  ASSERT_EQUAL(GetRedisErrorCodeLength(""), 0);
+  ASSERT_EQUAL(GetRedisErrorCodeLength(" "), 0);
+  return 0;
+}
+
+int testErrorCodeFormat(const char* error, const char* expected) {
+  char buf[1024];
+  sprintf(buf, "%.*s", GetRedisErrorCodeLength(error), error);
+  ASSERT_STRING_EQ(buf, expected);
+  return 0;
+}
+
+int testErrorCodeFormatting() {
+  testErrorCodeFormat("ERR Error message", "ERR");
+  testErrorCodeFormat("ERR-Error-message", "");
+  testErrorCodeFormat("ERR", "");
+  testErrorCodeFormat(" ERR", "");
+  testErrorCodeFormat("", "");
+  return 0;
+}
+
+TEST_MAIN({
+  TESTFUNC(testErrorCodeLengthExtraction);
+  TESTFUNC(testErrorCodeFormatting);
+})
diff --git a/tests/pytests/test_config.py b/tests/pytests/test_config.py

Original file line number	Diff line number	Diff line change
`@@ -1988,7 +1988,7 @@ static void addIndexCursor(const IndexSpec *sp) {`
`1988`	`1988`	`char *end = strchr(s, '{');`
`1989`	`1989`	`if (end) {`
`1990`	`1990`	`*end = '\0';`
`1991`		`- CursorList_AddSpec(&RSCursorsCoord, s, RSCURSORS_DEFAULT_CAPACITY);`
	`1991`	`+ CursorList_AddSpec(&RSCursorsCoord, s);`
`1992`	`1992`	`}`
`1993`	`1993`	`rm_free(s);`
`1994`	`1994`	`}`
Original file line number	Diff line number	Diff line change
`@@ -82,12 +82,12 @@ size_t MRChannel_MaxSize(MRChannel *chan) {`
`82`	`82`	`return ret;`
`83`	`83`	`}`
`84`	`84`
`85`		`-int MRChannel_Push(MRChannel chan, void ptr) {`
	`85`	`+PushErrorMask MRChannel_Push(MRChannel chan, void ptr) {`
`86`	`86`
`87`	`87`	`pthread_mutex_lock(&chan->lock);`
`88`		`- int rc = 1;`
	`88`	`+ int rc = 0;`
`89`	`89`	`if (!chan->open \|\| (chan->maxSize > 0 && chan->size == chan->maxSize)) {`
`90`		`- rc = 0;`
	`90`	`+ rc = CHANNEL_CLOSED;`
`91`	`91`	`goto end;`
`92`	`92`	`}`
`93`	`93`
`@@ -104,7 +104,7 @@ int MRChannel_Push(MRChannel chan, void ptr) {`
`104`	`104`	`}`
`105`	`105`	`chan->size++;`
`106`	`106`	`end:`
`107`		`- if (pthread_cond_broadcast(&chan->cond)) rc = 0;`
	`107`	`+ if (pthread_cond_broadcast(&chan->cond)) rc \|= BROADCAST_FAILURE;`
`108`	`108`	`pthread_mutex_unlock(&chan->lock);`
`109`	`109`	`return rc;`
`110`	`110`	`}`