Merge branch 'master' into snyk-fix-0967dab6c7294532dde05e178b572fee

Author: chayim

opt/build/docker/dockerfile.tmpl

@@ -24,6 +24,13 @@ RUN echo "Building for {{REDIS_OSNICK}} ({{REDIS_OS}}) for {{REDIS_ARCH}} [with
 {% include "templates/gpu.yml" %}
 {% endif %}
 
+# centos8 specific integration until a move to rocky or similar
+{% if REDIS_OSNICK == "centos8" %}
+RUN cd /etc/yum.repos.d/
+RUN sed -i 's/mirrorlist/#mirrorlist/g' /etc/yum.repos.d/CentOS-*
+RUN sed -i 's|#baseurl=http://mirror.centos.org|baseurl=http://vault.centos.org|g' /etc/yum.repos.d/CentOS-*
+{% endif %}
+
 WORKDIR /build
 COPY --from=redis /usr/local/ /usr/local/
 

opt/system-setup.py

@@ -71,7 +71,7 @@ def common_last(self):
         else:
             self.run("%s/bin/getcmake" % READIES)
 
-        self.pip_install("-r %s/tests/flow/tests_setup/test_requirements.txt" % ROOT)
+        self.pip_install("--ignore-installed PyYAML -r %s/tests/flow/tests_setup/test_requirements.txt" % ROOT)
 
         self.pip_install("awscli")
         self.pip_install("mkdocs mkdocs-material mkdocs-extensions")

src/backends/tensorflow.c

@@ -15,6 +15,10 @@ int RAI_InitBackendTF(int (*get_api_fn)(const char *, void *)) {
     get_api_fn("RedisModule_Realloc", ((void **)&RedisModule_Realloc));
     get_api_fn("RedisModule_Strdup", ((void **)&RedisModule_Strdup));
 
+    // Set min logging level to 3 (out of 5) - this is workaround since if TF is writing extensively
+    // log messages that to stderr, it may cause the system to be stuck.
+    RedisModule_Assert(putenv("TF_CPP_MIN_LOG_LEVEL=3") == 0);
+
     return REDISMODULE_OK;
 }
 

src/execution/DAG/dag.c

@@ -72,7 +72,7 @@ static void Dag_LoadInputsToCurrentOp(RedisAI_RunInfo *rinfo, RAI_DagOp *current
     }
 
     for (uint i = 0; i < n_outkeys; i++) {
-        RAI_ExecutionCtx_AddOuputPlaceholder(currentOp->ectx);
+        RAI_ExecutionCtx_AddOutputPlaceholder(currentOp->ectx);
     }
 }
 
@@ -260,7 +260,7 @@ void RedisAI_DagRunSession_ScriptRun_Step(RedisAI_RunInfo *rinfo, RAI_DagOp *cur
             RAI_ExecutionCtx_AddInput(currentOp->ectx, inputTensors[i]);
         }
         for (uint i = 0; i < n_outkeys; i++) {
-            RAI_ExecutionCtx_AddOuputPlaceholder(currentOp->ectx);
+            RAI_ExecutionCtx_AddOutputPlaceholder(currentOp->ectx);
         }
     }
 
@@ -574,44 +574,15 @@ int RedisAI_DagRun_Reply(RedisModuleCtx *ctx, RedisModuleString **argv, int argc
             break;
         }
 
-        case REDISAI_DAG_CMD_MODELRUN: {
-            rinfo->dagReplyLength++;
-            struct RedisAI_RunStats *rstats = NULL;
-            RAI_GetRunStats(currentOp->runkey, &rstats);
-            if (currentOp->result == REDISMODULE_ERR) {
-                RAI_SafeAddDataPoint(rstats, 0, 1, 1, 0);
-                RedisModule_ReplyWithError(ctx, currentOp->err->detail_oneline);
-                dag_error = 1;
-            } else if (currentOp->result == -1) {
-                RedisModule_ReplyWithSimpleString(ctx, "NA");
-            } else {
-                RAI_Tensor *t = NULL;
-                if (RAI_ExecutionCtx_NumOutputs(currentOp->ectx) > 0) {
-                    t = RAI_ExecutionCtx_GetOutput(currentOp->ectx, 0);
-                }
-                int batch_size = 0;
-                if (t) {
-                    batch_size = RAI_TensorDim(t, 0);
-                }
-                RAI_SafeAddDataPoint(rstats, currentOp->duration_us, 1, 0, batch_size);
-                RedisModule_ReplyWithSimpleString(ctx, "OK");
-            }
-            break;
-        }
-
+        case REDISAI_DAG_CMD_MODELRUN:
         case REDISAI_DAG_CMD_SCRIPTRUN: {
             rinfo->dagReplyLength++;
-            struct RedisAI_RunStats *rstats = NULL;
-            RAI_GetRunStats(currentOp->runkey, &rstats);
             if (currentOp->result == REDISMODULE_ERR) {
-                RAI_SafeAddDataPoint(rstats, 0, 1, 1, 0);
                 RedisModule_ReplyWithError(ctx, currentOp->err->detail_oneline);
                 dag_error = 1;
             } else if (currentOp->result == -1) {
                 RedisModule_ReplyWithSimpleString(ctx, "NA");
             } else {
-                int batch_size = 1;
-                RAI_SafeAddDataPoint(rstats, currentOp->duration_us, 1, 0, batch_size);
                 RedisModule_ReplyWithSimpleString(ctx, "OK");
             }
             break;

src/execution/DAG/dag_builder.c

@@ -61,7 +61,6 @@ RAI_DAGRunOp *RAI_DAGCreateModelRunOp(RAI_Model *model) {
     op->commandType = REDISAI_DAG_CMD_MODELRUN;
     op->ectx = (RAI_ExecutionCtx *)mctx;
     op->devicestr = model->devicestr;
-    op->runkey = RAI_HoldString((RedisModuleString *)model->infokey);
     return (RAI_DAGRunOp *)op;
 }
 
@@ -73,7 +72,6 @@ RAI_DAGRunOp *RAI_DAGCreateScriptRunOp(RAI_Script *script, const char *func_name
     op->commandType = REDISAI_DAG_CMD_SCRIPTRUN;
     op->ectx = (RAI_ExecutionCtx *)sctx;
     op->devicestr = script->devicestr;
-    op->runkey = RAI_HoldString((RedisModuleString *)script->infokey);
     return (RAI_DAGRunOp *)op;
 }
 

src/execution/DAG/dag_op.c

@@ -13,7 +13,6 @@ int RAI_InitDagOp(RAI_DagOp **result) {
     dagOp = (RAI_DagOp *)RedisModule_Calloc(1, sizeof(RAI_DagOp));
 
     dagOp->commandType = REDISAI_DAG_CMD_NONE;
-    dagOp->runkey = NULL;
     dagOp->inkeys = (RedisModuleString **)array_new(RedisModuleString *, 1);
     dagOp->outkeys = (RedisModuleString **)array_new(RedisModuleString *, 1);
     dagOp->inkeys_indices = array_new(size_t, 1);
@@ -31,13 +30,9 @@ int RAI_InitDagOp(RAI_DagOp **result) {
     return REDISMODULE_OK;
 }
 
-void RAI_DagOpSetRunKey(RAI_DagOp *dagOp, RedisModuleString *runkey) { dagOp->runkey = runkey; }
-
 void RAI_FreeDagOp(RAI_DagOp *dagOp) {
 
     RAI_FreeError(dagOp->err);
-    if (dagOp->runkey)
-        RedisModule_FreeString(NULL, dagOp->runkey);
 
     if (dagOp->outTensor)
         RAI_TensorFree(dagOp->outTensor);

src/execution/DAG/dag_op.h

@@ -18,7 +18,6 @@ typedef enum DAGCommand {
 
 typedef struct RAI_DagOp {
     DAGCommand commandType;
-    RedisModuleString *runkey;
     RedisModuleString **inkeys;
     RedisModuleString **outkeys;
     size_t *inkeys_indices;
@@ -48,12 +47,3 @@ int RAI_InitDagOp(RAI_DagOp **result);
  * @param RAI_DagOp context in which RedisAI command operates.
  */
 void RAI_FreeDagOp(RAI_DagOp *dagOp);
-
-/**
- * @brief Sets the key name of current dag op execution subject. The subject is either a model or a
- * script.
- *
- * @param dagOp Current op.
- * @param runkey Subject key name.
- */
-void RAI_DagOpSetRunKey(RAI_DagOp *dagOp, RedisModuleString *runkey);

src/execution/background_workers.c

@@ -54,10 +54,40 @@ static void _BGThread_Wait(RunQueueInfo *run_queue_info) {
                            &absTimeout);
 }
 
+static void _BGThread_SaveStats(RedisAI_RunInfo *rinfo) {
+    for (size_t i = 0; i < rinfo->dagOpCount; i++) {
+        RAI_DagOp *currentOp = rinfo->dagOps[i];
+
+        if (currentOp->commandType == REDISAI_DAG_CMD_MODELRUN ||
+            currentOp->commandType == REDISAI_DAG_CMD_SCRIPTRUN) {
+            if (currentOp->result == REDISMODULE_ERR) {
+                RAI_StatsAddDataPoint(RAI_ExecutionCtx_GetStats(currentOp->ectx), 0, 1, 1, 0);
+            } else if (currentOp->result == REDISMODULE_OK) {
+                unsigned long batch_size = 1;
+                if (currentOp->commandType == REDISAI_DAG_CMD_MODELRUN) {
+                    RAI_Tensor *t = NULL;
+                    if (RAI_ExecutionCtx_NumOutputs(currentOp->ectx) > 0) {
+                        t = RAI_ExecutionCtx_GetOutput(currentOp->ectx, 0);
+                    }
+                    if (t) {
+                        batch_size = RAI_TensorDim(t, 0);
+                    } else {
+                        batch_size = 0;
+                    }
+                }
+                RAI_StatsAddDataPoint(RAI_ExecutionCtx_GetStats(currentOp->ectx),
+                                      currentOp->duration_us, 1, 0, batch_size);
+            }
+        }
+    }
+}
+
 static void _BGThread_RinfoFinish(RedisAI_RunInfo *rinfo) {
     RedisAI_RunInfo *orig = rinfo->orig_copy;
     uint dagRefCount = RAI_DagRunInfoFreeShallowCopy(rinfo);
     if (dagRefCount == 0) {
+        // Save stats for every DAG execute operation.
+        _BGThread_SaveStats(orig);
         RedisAI_OnFinishCtx *finish_ctx = orig;
         orig->OnFinish(finish_ctx, orig->private_data);
     }

src/execution/execution_contexts/execution_ctx.c

@@ -2,9 +2,11 @@
 #include "redismodule.h"
 #include "util/arr.h"
 
-void RAI_ExecutionCtx_Init(RAI_ExecutionCtx *ctx, RAI_ExecutionCtx_Free_fn freeFn) {
+void RAI_ExecutionCtx_Init(RAI_ExecutionCtx *ctx, RAI_RunStats *run_stats,
+                           RAI_ExecutionCtx_Free_fn freeFn) {
     ctx->inputs = array_new(RAI_Tensor *, 10);
     ctx->outputs = array_new(RAI_Tensor *, 10);
+    ctx->runStats = run_stats;
     ctx->freeFn = freeFn;
 }
 void RAI_ExecutionCtx_Free(RAI_ExecutionCtx *ctx) {
@@ -21,6 +23,7 @@ void RAI_ExecutionCtx_Free(RAI_ExecutionCtx *ctx) {
 }
 
 inline size_t RAI_ExecutionCtx_NumInputs(RAI_ExecutionCtx *ctx) { return array_len(ctx->inputs); }
+
 inline void RAI_ExecutionCtx_AddInput(RAI_ExecutionCtx *ctx, RAI_Tensor *t) {
     if (t != NULL) {
         t = RAI_TensorGetShallowCopy(t);
@@ -35,7 +38,7 @@ inline RAI_Tensor *RAI_ExecutionCtx_GetInput(RAI_ExecutionCtx *ctx, size_t index
 
 inline size_t RAI_ExecutionCtx_NumOutputs(RAI_ExecutionCtx *ctx) { return array_len(ctx->outputs); }
 
-inline void RAI_ExecutionCtx_AddOuputPlaceholder(RAI_ExecutionCtx *ctx) {
+inline void RAI_ExecutionCtx_AddOutputPlaceholder(RAI_ExecutionCtx *ctx) {
     ctx->outputs = array_append(ctx->outputs, NULL);
 }
 
@@ -48,3 +51,5 @@ inline RAI_Tensor *RAI_ExecutionCtx_GetOutput(RAI_ExecutionCtx *ctx, size_t inde
     RedisModule_Assert(index < array_len(ctx->outputs));
     return ctx->outputs[index];
 }
+
+inline RAI_RunStats *RAI_ExecutionCtx_GetStats(RAI_ExecutionCtx *ctx) { return ctx->runStats; }

src/execution/execution_contexts/execution_ctx.h

@@ -1,5 +1,6 @@
 #pragma once
 
+#include <redis_ai_objects/stats.h>
 #include "redis_ai_objects/tensor.h"
 
 // Pre decleration
@@ -15,6 +16,7 @@ typedef void (*RAI_ExecutionCtx_Free_fn)(RAI_ExecutionCtx *ctx);
 typedef struct RAI_ExecutionCtx {
     RAI_Tensor **inputs;             // DAG op input tensors.
     RAI_Tensor **outputs;            // DAG op output tensors.
+    RAI_RunStats *runStats;          // The underline op's (Model/Script) stats entry.
     RAI_ExecutionCtx_Free_fn freeFn; // Inheriting execution context free function.
 } RAI_ExecutionCtx;
 
@@ -24,7 +26,8 @@ typedef struct RAI_ExecutionCtx {
  * @param ctx - Execution context to initialize.
  * @param freeFn - Specific free function for inheriting execution contexts (script or model)
  */
-void RAI_ExecutionCtx_Init(RAI_ExecutionCtx *ctx, RAI_ExecutionCtx_Free_fn freeFn);
+void RAI_ExecutionCtx_Init(RAI_ExecutionCtx *ctx, RAI_RunStats *run_stats,
+                           RAI_ExecutionCtx_Free_fn freeFn);
 
 /**
  * @brief Frees the execution context internal structures. To be used from an inhereting execution
@@ -72,7 +75,7 @@ size_t RAI_ExecutionCtx_NumOutputs(RAI_ExecutionCtx *ctx);
  *
  * @param ctx - Execution context.
  */
-void RAI_ExecutionCtx_AddOuputPlaceholder(RAI_ExecutionCtx *ctx);
+void RAI_ExecutionCtx_AddOutputPlaceholder(RAI_ExecutionCtx *ctx);
 
 /**
  * @brief Sets an output tensor in a specfic index, populated before by a placeholder.
@@ -91,3 +94,10 @@ void RAI_ExecutionCtx_SetOutput(RAI_ExecutionCtx *ctx, RAI_Tensor *t, size_t ind
  * @return RAI_Tensor* - Output tensor.
  */
 RAI_Tensor *RAI_ExecutionCtx_GetOutput(RAI_ExecutionCtx *ctx, size_t index);
+
+/**
+ * @brief Returns the RunStats object for underline object.
+ * @param ctx - Execution context.
+ * @return RAI_RunStats
+ */
+RAI_RunStats *RAI_ExecutionCtx_GetStats(RAI_ExecutionCtx *ctx);