Revert "[2.10] [MOD-11011] Fix deadlock while RDB loading and RM_Yield (#6763) (#6785)"

nafraf · nafraf · commit 7edfb1eeb376 · 2025-09-27T11:52:51.000-05:00
This reverts commit b38f839.
diff --git a/src/debug_commands.c b/src/debug_commands.c
@@ -1595,36 +1595,6 @@ DEBUG_COMMAND(bgScanController) {
 
 }
 
-// Global variable for sleep time before yielding (in microseconds)
-static unsigned int g_indexerSleepBeforeYieldMicros = 0;
-
-unsigned int GetIndexerSleepBeforeYieldMicros(void) {
-  return g_indexerSleepBeforeYieldMicros;
-}
-
-/**
- * FT.DEBUG INDEXER_SLEEP_BEFORE_YIELD [<microseconds>]
- * Get or set the sleep time in microseconds before yielding during indexing while loading
- */
-DEBUG_COMMAND(IndexerSleepBeforeYieldMicros) {
-  if (argc > 3) {
-    return RedisModule_WrongArity(ctx);
-  }
-
-  // Set new sleep time
-  if (argc == 3) {
-    long long sleepMicros;
-    if (RedisModule_StringToLongLong(argv[2], &sleepMicros) != REDISMODULE_OK || sleepMicros < 0) {
-      return RedisModule_ReplyWithError(ctx, "Invalid sleep time. Must be a non-negative integer.");
-    }
-
-    g_indexerSleepBeforeYieldMicros = (unsigned int)sleepMicros;
-    return RedisModule_ReplyWithSimpleString(ctx, "OK");
-  }
-
-  return RedisModule_WrongArity(ctx);
-}
-
 DebugCommandType commands[] = {{"DUMP_INVIDX", DumpInvertedIndex}, // Print all the inverted index entries.
                                {"DUMP_NUMIDX", DumpNumericIndex}, // Print all the headers (optional) + entries of the numeric tree.
                                {"DUMP_NUMIDXTREE", DumpNumericIndexTree}, // Print tree general info, all leaves + nodes + stats
@@ -1659,7 +1629,6 @@ DebugCommandType commands[] = {{"DUMP_INVIDX", DumpInvertedIndex}, // Print all
                                {"GET_HIDE_USER_DATA_FROM_LOGS", getHideUserDataFromLogs},
                                {"YIELDS_ON_LOAD_COUNTER", YieldCounter},
                                {"BG_SCAN_CONTROLLER", bgScanController},
-                               {"INDEXER_SLEEP_BEFORE_YIELD_MICROS", IndexerSleepBeforeYieldMicros},
                                /**
                                 * The following commands are for debugging distributed search/aggregation.
                                 */
diff --git a/src/debug_commands.h b/src/debug_commands.h
@@ -38,9 +38,3 @@ typedef struct DebugCTX {
 
 // Should be called after each debug command that changes the debugCtx
 void validateDebugMode(DebugCTX *debugCtx);
-
-// Yield counter functions
-void IncrementYieldCounter(void);
-
-// Indexer sleep before yield functions
-unsigned int GetIndexerSleepBeforeYieldMicros(void);
diff --git a/src/indexer.c b/src/indexer.c
@@ -106,6 +106,7 @@ static void writeCurEntries(DocumentIndexer *indexer, RSAddDocumentCtx *aCtx, Re
     if (invidx) {
       entry->docId = aCtx->doc->docId;
       RS_LOG_ASSERT(entry->docId, "docId should not be 0");
+      IndexerYieldWhileLoading(ctx->redisCtx);
       writeIndexEntry(spec, invidx, encoder, entry);
       if (Index_StoreFieldMask(spec)) {
         invidx->fieldMask |= entry->fieldMask;
@@ -222,6 +223,8 @@ static void indexBulkFields(RSAddDocumentCtx *aCtx, RedisSearchCtx *sctx) {
       if (fs->types == INDEXFLD_T_FULLTEXT || !FieldSpec_IsIndexable(fs) || fdata->isNull) {
         continue;
       }
+
+      IndexerYieldWhileLoading(sctx->redisCtx);
       if (IndexerBulkAdd(cur, sctx, doc->fields + ii, fs, fdata, &cur->status) != 0) {
         IndexError_AddError(&cur->spec->stats.indexError, cur->status.message, cur->status.detail, doc->docKey);
         FieldSpec_AddError(&cur->spec->fields[fs->index], cur->status.message, cur->status.detail, doc->docKey);
@@ -396,21 +399,14 @@ bool g_isLoading = false;
  * Yield to Redis after a certain number of operations during indexing.
  * This helps keep Redis responsive during long indexing operations.
  * @param ctx The Redis context
- * @param numOps Tue number of operations to count in the counter before considering RSGlobalConfig.indexerYieldEveryOpsWhileLoading. These are related to the number of fields in the document
- * @param flags The flags to pass to RedisModule_Yield
  */
-void IndexerYieldWhileLoading(RedisModuleCtx *ctx, unsigned int numOps, int flags) {
+static void IndexerYieldWhileLoading(RedisModuleCtx *ctx) {
   static size_t opCounter = 0;
 
-  // If server is loading, Yield to Redis if the number of operations is greater than the yieldEveryOps
-  opCounter += numOps;
-  if (g_isLoading && opCounter >= RSGlobalConfig.indexerYieldEveryOpsWhileLoading) {
-    opCounter = opCounter % RSGlobalConfig.indexerYieldEveryOpsWhileLoading;
+  // If server is loading, Yield to Redis every RSGlobalConfig.indexerYieldEveryOps operations
+  if (g_isLoading && ++opCounter >= RSGlobalConfig.indexerYieldEveryOpsWhileLoading) {
+    opCounter = 0;
     IncrementYieldCounter(); // Track that we called yield
-    unsigned int sleepMicros = GetIndexerSleepBeforeYieldMicros();
-    if (sleepMicros > 0) {
-      usleep(sleepMicros);
-    }
-    RedisModule_Yield(ctx, flags, NULL);
+    RedisModule_Yield(ctx, REDISMODULE_YIELD_FLAG_CLIENTS, NULL);
   }
 }
diff --git a/src/indexer.h b/src/indexer.h
@@ -88,9 +88,7 @@ int IndexerBulkAdd(RSAddDocumentCtx *cur, RedisSearchCtx *sctx,
  * Yield to Redis after a certain number of operations during indexing while loading.
  * This helps keep Redis responsive during long indexing operations.
  * @param ctx The Redis context
- * @param numOps Tue number of operations to count in the counter before considering RSGlobalConfig.indexerYieldEveryOpsWhileLoading. These are related to the number of fields in the document
- * @param flags The flags to pass to RedisModule_Yield
  */
-void IndexerYieldWhileLoading(RedisModuleCtx *ctx, unsigned int numOps, int flags);
+static void IndexerYieldWhileLoading(RedisModuleCtx *ctx);
 
 #endif
diff --git a/src/info/indexes_info.c b/src/info/indexes_info.c
@@ -7,7 +7,6 @@
 #include "indexes_info.h"
 #include "util/dict.h"
 #include "spec.h"
-#include <string.h>  // Add this for strerror
 
 // Assuming the GIL is held by the caller
 TotalIndexesInfo IndexesInfo_TotalInfo() {
@@ -28,11 +27,7 @@ TotalIndexesInfo IndexesInfo_TotalInfo() {
       continue;
     }
     // Lock for read
-    int rc = pthread_rwlock_rdlock(&sp->rwlock);
-    if (rc != 0) {
-      RedisModule_Log(RSDummyContext, "warning", "Failed to acquire read lock on index: rc=%d (%s). Cannot continue getting Index info", rc, strerror(rc));
-      continue;
-    }
+    pthread_rwlock_rdlock(&sp->rwlock);
 
     // Vector index stats
     VectorIndexStats vec_info = IndexSpec_GetVectorIndexStats(sp);
diff --git a/src/spec.c b/src/spec.c
@@ -1812,21 +1812,6 @@ void IndexSpec_InitializeSynonym(IndexSpec *sp) {
 
 ///////////////////////////////////////////////////////////////////////////////////////////////
 
-static void IndexSpec_InitLock(IndexSpec *sp) {
-  int res = 0;
-  pthread_rwlockattr_t attr;
-  res = pthread_rwlockattr_init(&attr);
-  RS_ASSERT(res == 0);
-#if !defined(__APPLE__) && !defined(__FreeBSD__) && defined(__GLIBC__)
-  int pref = PTHREAD_RWLOCK_PREFER_WRITER_NONRECURSIVE_NP;
-  res = pthread_rwlockattr_setkind_np(&attr, pref);
-  RS_ASSERT(res == 0);
-#endif
-
-  pthread_rwlock_init(&sp->rwlock, &attr);
-}
-
-
 IndexSpec *NewIndexSpec(const char *name) {
   IndexSpec *sp = rm_calloc(1, sizeof(IndexSpec));
   sp->fields = rm_calloc(sizeof(FieldSpec), SPEC_MAX_FIELDS);
@@ -1854,7 +1839,17 @@ IndexSpec *NewIndexSpec(const char *name) {
   memset(&sp->stats, 0, sizeof(sp->stats));
   sp->stats.indexError = IndexError_Init();
 
-  IndexSpec_InitLock(sp);
+  int res = 0;
+  pthread_rwlockattr_t attr;
+  res = pthread_rwlockattr_init(&attr);
+  RS_ASSERT(res == 0);
+#if !defined(__APPLE__) && !defined(__FreeBSD__)
+  int pref = PTHREAD_RWLOCK_PREFER_WRITER_NONRECURSIVE_NP;
+  res = pthread_rwlockattr_setkind_np(&attr, pref);
+  RS_ASSERT(res == 0);
+#endif
+
+  pthread_rwlock_init(&sp->rwlock, &attr);
 
   return sp;
 }
@@ -2659,7 +2654,6 @@ int IndexSpec_CreateFromRdb(RedisModuleCtx *ctx, RedisModuleIO *rdb, int encver,
   RedisModule_Free(rawName);
 
   IndexSpec *sp = rm_calloc(1, sizeof(IndexSpec));
-  IndexSpec_InitLock(sp);
   StrongRef spec_ref = StrongRef_New(sp, (RefManager_Free)IndexSpec_Free);
   sp->own_ref = spec_ref;
   // setting isDuplicate to true will make sure index will not be removed from aliases container.
@@ -2792,7 +2786,6 @@ void *IndexSpec_LegacyRdbLoad(RedisModuleIO *rdb, int encver) {
 
   RedisModuleCtx *ctx = RedisModule_GetContextFromIO(rdb);
   IndexSpec *sp = rm_calloc(1, sizeof(IndexSpec));
-  IndexSpec_InitLock(sp);
   StrongRef spec_ref = StrongRef_New(sp, (RefManager_Free)IndexSpec_Free);
   sp->own_ref = spec_ref;
 
@@ -2890,7 +2883,6 @@ void *IndexSpec_LegacyRdbLoad(RedisModuleIO *rdb, int encver) {
   Cursors_initSpec(sp);
 
   dictAdd(legacySpecDict, sp->name, spec_ref.rm);
-
   return spec_ref.rm;
 }
 
@@ -3119,8 +3111,6 @@ int IndexSpec_UpdateDoc(IndexSpec *spec, RedisModuleCtx *ctx, RedisModuleString
     return REDISMODULE_ERR;
   }
 
-  unsigned int numOps = doc.numFields != 0 ? doc.numFields: 1;
-  IndexerYieldWhileLoading(ctx, numOps, REDISMODULE_YIELD_FLAG_CLIENTS);
   RedisSearchCtx_LockSpecWrite(&sctx);
   IndexSpec_IncrActiveWrites(spec);
 
diff --git a/tests/pytests/test_debug_commands.py b/tests/pytests/test_debug_commands.py
@@ -57,7 +57,6 @@ def testDebugHelp(self):
             'GET_HIDE_USER_DATA_FROM_LOGS',
             'YIELDS_ON_LOAD_COUNTER',
             'BG_SCAN_CONTROLLER',
-            'INDEXER_SLEEP_BEFORE_YIELD_MICROS',
             'FT.AGGREGATE',
             'FT.SEARCH',
         ]
diff --git a/tests/pytests/test_rdb_load.py b/tests/pytests/test_rdb_load.py

Original file line number	Diff line number	Diff line change
`@@ -57,7 +57,6 @@ def testDebugHelp(self):`
`57`	`57`	`'GET_HIDE_USER_DATA_FROM_LOGS',`
`58`	`58`	`'YIELDS_ON_LOAD_COUNTER',`
`59`	`59`	`'BG_SCAN_CONTROLLER',`
`60`		`- 'INDEXER_SLEEP_BEFORE_YIELD_MICROS',`
`61`	`60`	`'FT.AGGREGATE',`
`62`	`61`	`'FT.SEARCH',`
`63`	`62`	`]`