Refactor - Create a function that parses all run commands in command parser file. The parsing is split according to the original command, and at the end we always create run info and sent it to queues.

Author: alonre24

src/CMakeLists.txt

@@ -6,13 +6,13 @@ ADD_LIBRARY(redisai_obj OBJECT
         util/dict.c
         util/queue.c
         redisai.c
+        command_parser.c
         run_info.c
         background_workers.c
         config.c
-        dag.c
-        dag_parser.c
+        DAG/dag.c
+        DAG/dag_parser.c
         modelRun_ctx.c
-        async_llapi.c
         backends.c
         backends/util.c
         model.c

src/dag.c renamed to src/DAG/dag.c

@@ -874,53 +874,21 @@ void DAG_ReplyAndUnblock(RedisAI_OnFinishCtx *ctx, void *private_data) {
         RedisModule_UnblockClient(rinfo->client, rinfo);
 }
 
-int RedisAI_ProcessDagRunCommand(RedisModuleCtx *ctx, RedisModuleString **argv, int argc,
-                                 int dagMode) {
-
-    int flags = RedisModule_GetContextFlags(ctx);
-    bool blocking_not_allowed = (flags & (REDISMODULE_CTX_FLAGS_MULTI | REDISMODULE_CTX_FLAGS_LUA));
-    if (blocking_not_allowed)
-        return RedisModule_ReplyWithError(
-            ctx, "ERR Cannot run RedisAI command within a transaction or a LUA script");
-    RedisAI_RunInfo *rinfo = NULL;
-    if (RAI_InitRunInfo(&rinfo) == REDISMODULE_ERR) {
-        RedisModule_ReplyWithError(
-            ctx, "ERR Unable to allocate the memory and initialise the RedisAI_RunInfo structure");
-        return REDISMODULE_ERR;
-    }
-    // Parse DAG string command and store the data in rinfo obj.
-    int status = DAG_CommandParser(ctx, argv, argc, dagMode, &rinfo);
-    if (status == REDISMODULE_ERR)
-        return REDISMODULE_OK;
-    // Block the client before adding rinfo to the run queues (sync call).
-    rinfo->client = RedisModule_BlockClient(ctx, RedisAI_DagRun_Reply, NULL, RunInfo_FreeData, 0);
-    RedisModule_SetDisconnectCallback(rinfo->client, RedisAI_Disconnected);
-    rinfo->OnFinish = DAG_ReplyAndUnblock;
-    return DAG_InsertDAGToQueue(rinfo);
-}
+int Dag_PopulateSingleModelRunOp(RedisAI_RunInfo *rinfo, RAI_ModelRunCtx *mctx,
+                                 RedisModuleString **inkeys, RedisModuleString **outkeys,
+                                 RedisModuleString *runkey, long long timeout) {
 
-RedisAI_RunInfo *Dag_CreateFromSingleModelRunOp(RAI_ModelRunCtx *mctx, RAI_Error *error,
-                                                RedisModuleString **inkeys,
-                                                RedisModuleString **outkeys,
-                                                RedisModuleString *runkey, long long timeout) {
-    RedisAI_RunInfo *rinfo = NULL;
-    if (RAI_InitRunInfo(&rinfo) == REDISMODULE_ERR) {
-        RAI_SetError(
-            error, RedisAI_ErrorCode_EMODELRUN,
-            "ERR Unable to allocate the memory and initialise the RedisAI_RunInfo structure");
-        return NULL;
-    }
+    assert(rinfo->single_op_dag);
     rinfo->single_device_dag = 1;
-    rinfo->single_op_dag = 1;
     rinfo->dagOpCount = 1;
     rinfo->timeout = timeout;
 
     RAI_DagOp *currentDagOp;
     if (RAI_InitDagOp(&currentDagOp) == REDISMODULE_ERR) {
-        RAI_SetError(error, RedisAI_ErrorCode_EMODELRUN,
+        RAI_SetError(rinfo->err, RedisAI_ErrorCode_EMODELRUN,
                      "ERR Unable to allocate the memory and initialise the RAI_dagOp structure");
-        return NULL;
-    };
+        return REDISMODULE_ERR;
+    }
     rinfo->dagOps = array_append(rinfo->dagOps, currentDagOp);
 
     RAI_DagOp *currentOp = rinfo->dagOps[0];
@@ -931,5 +899,5 @@ RedisAI_RunInfo *Dag_CreateFromSingleModelRunOp(RAI_ModelRunCtx *mctx, RAI_Error
     currentOp->outkeys = outkeys;
     currentOp->runkey = runkey;
 
-    return rinfo;
+    return REDISMODULE_OK;
 }

src/dag.h renamed to src/DAG/dag.h

@@ -162,18 +162,17 @@ void RunInfo_FreeData(RedisModuleCtx *ctx, void *rinfo);
 void RedisAI_Disconnected(RedisModuleCtx *ctx, RedisModuleBlockedClient *bc);
 
 /**
- * @brief Create a DAG with a single operation of MODELRUN.
+ * @brief Populate a DAG with a single operation of MODELRUN.
+ * @param rinfo An existing DAG to populate.
  * @param mctx ModelRunCtx that represents the single MODELRUN op.
- * @param error RAI_Error object to store error if one occurs.
  * @param inkeys The DAG operation inkeys (the input tensors).
  * @param outkeys The DAG operation outkeys (the output tensors).
  * @param runkey The model key.
  * @param timeout The operation timeout (if given, otherwise it is zero).
  */
 
-RedisAI_RunInfo *Dag_CreateFromSingleModelRunOp(RAI_ModelRunCtx *mctx, RAI_Error *error,
-                                                RedisModuleString **inkeys,
-                                                RedisModuleString **outkeys,
-                                                RedisModuleString *runkey, long long timeout);
+int Dag_PopulateSingleModelRunOp(RedisAI_RunInfo *rinfo, RAI_ModelRunCtx *mctx,
+                                 RedisModuleString **inkeys, RedisModuleString **outkeys,
+                                 RedisModuleString *runkey, long long timeout);
 
 #endif /* SRC_DAG_H_ */

src/dag_parser.c renamed to src/DAG/dag_parser.c

@@ -7,10 +7,21 @@
 
 /**
  * DAGRUN Building Block to parse [LOAD <nkeys> key1 key2... ]
+ *
+ * @param ctx Context in which Redis modules operate
+ * @param argv Redis command arguments, as an array of strings
+ * @param argc Redis command number of arguments
+ * @param loadedContextDict local non-blocking hash table containing key names
+ * loaded from the keyspace tensors
+ * @param localContextDict local non-blocking hash table containing DAG's
+ * tensors
+ * @param chaining_operator operator used to split operations. Any command
+ * argument after the chaining operator is not considered
+ * @return processed number of arguments on success, or -1 if the parsing failed
  */
-int RAI_parseDAGLoadArgs(RedisModuleCtx *ctx, RedisModuleString **argv, int argc,
-                         AI_dict **loadedContextDict, AI_dict **localContextDict,
-                         const char *chaining_operator) {
+static int DAG_ParseLoadArgs(RedisModuleCtx *ctx, RedisModuleString **argv, int argc,
+                             AI_dict **loadedContextDict, AI_dict **localContextDict,
+                             const char *chaining_operator) {
     if (argc < 3) {
         RedisModule_WrongArity(ctx);
         return -1;
@@ -60,9 +71,18 @@ int RAI_parseDAGLoadArgs(RedisModuleCtx *ctx, RedisModuleString **argv, int argc
 
 /**
  * DAGRUN Building Block to parse [PERSIST <nkeys> key1 key2... ]
+ *
+ * @param ctx Context in which Redis modules operate
+ * @param argv Redis command arguments, as an array of strings
+ * @param argc Redis command number of arguments
+ * @param localContextDict local non-blocking hash table containing DAG's
+ * keynames marked as persistent
+ * @param chaining_operator operator used to split operations. Any command
+ * argument after the chaining operator is not considered
+ * @return processed number of arguments on success, or -1 if the parsing failed
  */
-int RAI_parseDAGPersistArgs(RedisModuleCtx *ctx, RedisModuleString **argv, int argc,
-                            AI_dict **persistContextDict, const char *chaining_operator) {
+static int DAG_ParsePersistArgs(RedisModuleCtx *ctx, RedisModuleString **argv, int argc,
+                                AI_dict **persistContextDict, const char *chaining_operator) {
     if (argc < 3) {
         RedisModule_WrongArity(ctx);
         return -1;
@@ -129,9 +149,9 @@ int DAG_CommandParser(RedisModuleCtx *ctx, RedisModuleString **argv, int argc, i
         if (!strcasecmp(arg_string, "LOAD") && !load_complete) {
             /* Load the required tensors from key space and store them in both
                dagTensorsLoadedContext and dagTensorsContext dicts. */
-            const int parse_result = RAI_parseDAGLoadArgs(ctx, &argv[arg_pos], argc - arg_pos,
-                                                          &(rinfo->dagTensorsLoadedContext),
-                                                          &(rinfo->dagTensorsContext), "|>");
+            const int parse_result = DAG_ParseLoadArgs(ctx, &argv[arg_pos], argc - arg_pos,
+                                                       &(rinfo->dagTensorsLoadedContext),
+                                                       &(rinfo->dagTensorsContext), "|>");
             if (parse_result > 0) {
                 arg_pos += parse_result - 1;
                 load_complete = true;
@@ -148,7 +168,7 @@ int DAG_CommandParser(RedisModuleCtx *ctx, RedisModuleString **argv, int argc, i
             }
             /* Store the keys to persist in dagTensorsPersistedContext dict.
                These keys will be populated late on with actual tensors. */
-            const int parse_result = RAI_parseDAGPersistArgs(
+            const int parse_result = DAG_ParsePersistArgs(
                 ctx, &argv[arg_pos], argc - arg_pos, &(rinfo->dagTensorsPersistedContext), "|>");
             if (parse_result > 0) {
                 arg_pos += parse_result - 1;

src/DAG/dag_parser.h

@@ -0,0 +1,9 @@
+#ifndef REDISAI_DAG_PARSER_H
+#define REDISAI_DAG_PARSER_H
+
+#include "run_info.h"
+
+int DAG_CommandParser(RedisModuleCtx *ctx, RedisModuleString **argv, int argc, int dagMode,
+                      RedisAI_RunInfo **rinfo_ptr);
+
+#endif // REDISAI_DAG_PARSER_H

src/async_llapi.c

@@ -9,7 +9,7 @@
  */
 
 #include "background_workers.h"
-#include "dag.h"
+#include "DAG/dag.h"
 #include "model.h"
 #include "redisai.h"
 #include "rmutil/alloc.h"

src/async_llapi.h

@@ -19,7 +19,7 @@
 #include <pthread.h>
 
 #include "config.h"
-#include "dag.h"
+#include "DAG/dag.h"
 #include "model.h"
 #include "redisai.h"
 #include "rmutil/alloc.h"