MOD-5948: Respect timeout policy P1 (single shard) (#4038) (#4133)

* wip * wip * fix resp3 response * fix resp2 section * fix total results counter report * fix loop * fix response condition * fix condition * fix leak * add timeout check after aggregation in strict timeout policy * fix coordinator to wait for shard-reply BEFORE polling for timeout * fix test * fix test * wait for reply after polling for timeout (so we don't loose data) * fix leak * fix leak * add test * non-related code touchups * reposition response section * move duplicated code to function * some fixes * address reveiw * fix nelem increment * fix use after free * address review * address comment
RediSearch · Dec 1, 2023 · 99d950c · 99d950c
1 parent 5fc89f4
commit 99d950c
Show file tree

Hide file tree

Showing 13 changed files with 290 additions and 159 deletions.
diff --git a/coord/src/dist_aggregate.c b/coord/src/dist_aggregate.c
@@ -358,7 +358,6 @@ static int rpnetNext(ResultProcessor *self, SearchResult *r) {
 
   // get the next reply from the channel
   while (!root || !rows || MRReply_Length(rows) == 0) {
-    // if(TimedOut(&self->parent->sctx->timeout)) {
     if(TimedOut(&self->parent->sctx->timeout)) {
       // Set the `timedOut` flag in the MRIteratorCtx, later to be read by the
       // callback so that a `CURSOR DEL` command will be dispatched instead of

diff --git a/src/aggregate/aggregate.h b/src/aggregate/aggregate.h
@@ -92,6 +92,8 @@ typedef enum {
 #define IsFormatExpand(r) ((r)->reqflags & QEXEC_FORMAT_EXPAND)
 #define IsWildcard(r) ((r)->ast.root->type == QN_WILDCARD)
 #define HasScorer(r) ((r)->optimizer->scorerType != SCORER_TYPE_NONE)
+// Get the index search context from the result processor
+#define RP_SCTX(rpctx) ((rpctx)->parent->sctx)
 
 #ifdef MT_BUILD
 // Indicates whether a query should run in the background. This