Added another test - multidevice resnet + documentation of LLAPI

alonre24 · alonre24 · commit ff2754b1d7f2 · 2021-01-11T13:10:45.000+02:00
diff --git a/src/DAG/dag_builder.h b/src/DAG/dag_builder.h
@@ -2,28 +2,89 @@
 
 #include "redisai.h"
 
+/**
+ * @brief Create a new empty DAG runInfo object.
+ */
 RAI_DAGRunCtx *RAI_DAGRunCtxCreate(void);
 
+/**
+ * @brief Create a new MODELRUN op for a DAG.
+ * @param model The model to run.
+ */
 RAI_DAGRunOp *RAI_DAGCreateModelRunOp(RAI_Model *model);
 
+/**
+ * @brief Create a new SCRIPTRUN op for a DAG.
+ * @param script The script to run.
+ * @param func_name The specific function to run in the given script.
+ */
 RAI_DAGRunOp *RAI_DAGCreateScriptRunOp(RAI_Script *script, const char *func_name);
 
+/**
+ * @brief Add an input key to a DAG run op (before inserting it to the DAG).
+ * @param DAGop The DAG run op (MODELRUN / SCRIPTRUN).
+ * @param input The tensor input name (this name should appear in a previous op of the DAG).
+ */
 int RAI_DAGRunOpAddInput(RAI_DAGRunOp *DAGOp, const char *input);
 
+/**
+ * @brief Add an output key to a DAG run op (before inserting it to the DAG).
+ * @param DAGop The DAG run op (MODELRUN / SCRIPTRUN).
+ * @param output The tensor output name (this name may appear in one of the following ops of the
+ * DAG).
+ */
 int RAI_DAGRunOpAddOutput(RAI_DAGRunOp *DAGOp, const char *output);
 
+/**
+ * @brief Add a run op (MODELRUN/SCRIPTRUN) to a DAG.
+ * @param runInfo The DAG to insert the op to.
+ * @param DAGop The DAG run op (MODELRUN / SCRIPTRUN).
+ * @param err Error is returned in case of a MODELRUN op if the number of inputs and outputs
+ * given to the op does not match to the number of inputs and outputs in the model definition.
+ */
 int RAI_DAGAddRunOp(RAI_DAGRunCtx *run_info, RAI_DAGRunOp *DAGop, RAI_Error *err);
 
+/**
+ * @brief Load a tensor from keyspace to the DAG local context.
+ * @param runInfo The DAG to load the tensor into.
+ * @param tname The tensor key.
+ * @param err Error is returned in case that the key does not exist, or not holding a tensor type.
+ */
 int RAI_DAGLoadTensor(RAI_DAGRunCtx *run_info, const char *t_name, RAI_Error *err);
 
+/**
+ * @brief Load a tensor from keyspace to the DAG local context.
+ * @param runInfo The DAG to load the tensor into.
+ * @param tname The tensor key (can hold any binary string).
+ * @param err Error is returned in case that the key does not exist, or not holding a tensor type.
+ */
 int RAI_DAGLoadTensorRS(RAI_DAGRunCtx *run_info, RedisModuleString *t_name, RAI_Error *err);
 
+/**
+ * @brief Append a TENSORSET op to a DAG (can use to load an intermediate tensors)
+ * @param runInfo The DAG to append this op into.
+ * @param tensor The tensor to set.
+ */
 int RAI_DAGAddTensorSet(RAI_DAGRunCtx *run_info, const char *t_name, RAI_Tensor *tensor);
 
+/**
+ * @brief Append a TENSORGET op to a DAG (can use to output intermediate and final tensors)
+ * @param runInfo The DAG to append this op into.
+ * @param tensor The tensor to set.
+ */
 int RAI_DAGAddTensorGet(RAI_DAGRunCtx *run_info, const char *t_name, RAI_Error *err);
 
+/**
+ * @brief Returns the number of ops in a DAG.
+ */
 size_t RAI_DAGNumOps(RAI_DAGRunCtx *run_info);
 
+/**
+ * @brief Free DAG's runInfo and all its internal ops.
+ */
 void RAI_DAGFree(RAI_DAGRunCtx *run_info);
 
-void RAI_DAGRunOpFree(RAI_DAGRunOp *dagOp);
+/**
+ * @brief Free a specific DAG run op (MODELRUN/SCRIPTRUN).
+ */
+void RAI_DAGRunOpFree(RAI_DAGRunOp *dagOp);
diff --git a/src/DAG/dag_execute.h b/src/DAG/dag_execute.h
@@ -16,13 +16,44 @@
  demangle the keys in order to persist them.*/
 int MangleTensorsNames(RedisAI_RunInfo *rinfo);
 
+/**
+ * @brief Run asynchronously a DAG. This will validate that the sequence of DAG ops
+ * is valid and generate a unique key to the tensor that flow in the DAG (mangleTensorsNames)
+ * Then, DAG is sent to the devices' run queues and will be execute by a workung thread.
+ * @param DAGAsyncFinish This is a callback that will be called after the whole DAG finish its run.
+ * @param private_data This is an input to the DAGAsyncFinish callback. Can be used to save the
+ * results and errors
+ * @param err Error is returned in case that the validation failed, and the DAG wasn't inserted to
+ * the queues.
+ */
 int RAI_DAGRun(RAI_DAGRunCtx *run_info, RAI_OnFinishCB DAGAsyncFinish, void *private_data,
                RAI_Error *err);
 
+/**
+ * @brief This can be called in the finish CB, returns the number of outputs (TENSORGET ops).
+ * @param finish_ctx This represents the DAG runInfo at the end of the run.
+ */
 size_t RAI_DAGNumOutputs(RAI_OnFinishCtx *finish_ctx);
 
+/**
+ * @brief This can be called in the finish CB, returns a specific output tensor (result of a
+ * TENSORGET op).
+ * @param finish_ctx This represents the DAG runInfo at the end of the run.
+ * @param index The index of the TENSORGET op in the DAG.
+ * @retval returns the tensor that the i'th TENSORGET op outputs.
+ */
 RAI_Tensor *RAI_DAGOutputTensor(RAI_OnFinishCtx *finish_ctx, size_t index);
 
+/**
+ * @brief Returns 1 if (at least) one of the DAG ops encountered an error.
+ */
 int RAI_DAGRunError(RAI_OnFinishCtx *finish_ctx);
 
+/**
+ * @brief This can be called in the finish CB, returns the status of a certain in a DAG.
+ * @param finish_ctx This represents the DAG runInfo at the end of the run.
+ * @param index Index of a specific op in the DAG.
+ * @retval returns an object that represents the i'th op status, from which a user can
+ * obtain the error code (error code is "OK" if no error has occurred) and error details.
+ */
 RAI_Error *RAI_DAGCopyOpStatus(RAI_OnFinishCtx *finish_ctx, size_t index);
diff --git a/src/tensor.c b/src/tensor.c
@@ -486,12 +486,12 @@ int RAI_TensorGetValueAsDouble(RAI_Tensor *t, long long i, double *val) {
             *val = ((double *)data)[i];
             break;
         default:
-            return 1;
+            return REDISMODULE_ERR;
         }
     } else {
-        return 1;
+        return REDISMODULE_ERR;
     }
-    return 0;
+    return REDISMODULE_OK;
 }
 
 int RAI_TensorGetValueAsLongLong(RAI_Tensor *t, long long i, long long *val) {
@@ -515,7 +515,7 @@ int RAI_TensorGetValueAsLongLong(RAI_Tensor *t, long long i, long long *val) {
             *val = ((int64_t *)data)[i];
             break;
         default:
-            return 0;
+            return REDISMODULE_ERR;
         }
     } else if (dtype.code == kDLUInt) {
         switch (dtype.bits) {
@@ -532,12 +532,12 @@ int RAI_TensorGetValueAsLongLong(RAI_Tensor *t, long long i, long long *val) {
             *val = ((uint64_t *)data)[i];
             break;
         default:
-            return 0;
+            return REDISMODULE_ERR;
         }
     } else {
-        return 0;
+        return REDISMODULE_ERR;
     }
-    return 1;
+    return REDISMODULE_OK;
 }
 
 RAI_Tensor *RAI_TensorGetShallowCopy(RAI_Tensor *t) {
@@ -819,7 +819,7 @@ int RAI_TensorReplyWithValues(RedisModuleCtx *ctx, RAI_Tensor *t) {
         double val;
         for (i = 0; i < len; i++) {
             int ret = RAI_TensorGetValueAsDouble(t, i, &val);
-            if (ret == 1) {
+            if (ret == REDISMODULE_ERR) {
                 RedisModule_ReplyWithError(ctx, "ERR cannot get values for this datatype");
                 return -1;
             }
@@ -829,7 +829,7 @@ int RAI_TensorReplyWithValues(RedisModuleCtx *ctx, RAI_Tensor *t) {
         long long val;
         for (i = 0; i < len; i++) {
             int ret = RAI_TensorGetValueAsLongLong(t, i, &val);
-            if (!ret) {
+            if (ret == REDISMODULE_ERR) {
                 RedisModule_ReplyWithError(ctx, "ERR cannot get values for this datatype");
                 return -1;
             }
diff --git a/tests/flow/tests_llapi.py b/tests/flow/tests_llapi.py
@@ -103,3 +103,47 @@ def test_dag_build_and_run(env):
 
     ret = con.execute_command("RAI_llapi.DAGrun")
     env.assertEqual(ret, b'DAG run success')
+
+
+@ensure_test_module_loaded
+def test_llapi_dagrun_multidevice_resnet(env):
+    con = env.getConnection()
+
+    model_name_0 = 'imagenet_model1:{{1}}'
+    model_name_1 = 'imagenet_model2:{{1}}'
+    script_name_0 = 'imagenet_script1:{{1}}'
+    script_name_1 = 'imagenet_script2:{{1}}'
+    inputvar = 'images'
+    outputvar = 'output'
+    image_key = 'image:{{1}}'
+    temp_key1 = 'temp_key1:{{1}}'
+    temp_key2_0 = 'temp_key2_0'
+    temp_key2_1 = 'temp_key2_1'
+    class_key_0 = 'output0:{{1}}'
+    class_key_1 = 'output1:{{1}}'
+
+    model_pb, script, labels, img = load_resnet_test_data()
+
+    device_0 = 'CPU:1'
+    device_1 = DEVICE
+
+    ret = con.execute_command('AI.MODELSET', model_name_0, 'TF', device_0,
+                              'INPUTS', inputvar,
+                              'OUTPUTS', outputvar,
+                              'BLOB', model_pb)
+    env.assertEqual(ret, b'OK')
+
+    ret = con.execute_command('AI.MODELSET', model_name_1, 'TF', device_1,
+                              'INPUTS', inputvar,
+                              'OUTPUTS', outputvar,
+                              'BLOB', model_pb)
+    env.assertEqual(ret, b'OK')
+    ret = con.execute_command('AI.SCRIPTSET', script_name_0, device_0, 'SOURCE', script)
+    env.assertEqual(ret, b'OK')
+    ret = con.execute_command('AI.SCRIPTSET', script_name_1, device_1, 'SOURCE', script)
+    env.assertEqual(ret, b'OK')
+    ret = con.execute_command('AI.TENSORSET', image_key, 'UINT8', img.shape[1], img.shape[0], 3, 'BLOB', img.tobytes())
+    env.assertEqual(ret, b'OK')
+
+    ret = con.execute_command("RAI_llapi.DAG_resnet")
+    env.assertEqual(ret, b'DAG resnet success')
diff --git a/tests/module/LLAPI.c b/tests/module/LLAPI.c
@@ -264,5 +264,10 @@ int RedisModule_OnLoad(RedisModuleCtx *ctx, RedisModuleString **argv, int argc)
         return REDISMODULE_ERR;
     }
 
+    if(RedisModule_CreateCommand(ctx, "RAI_llapi.DAG_resnet", RAI_llapi_DAG_resnet, "",
+      0, 0, 0) == REDISMODULE_ERR) {
+        return REDISMODULE_ERR;
+    }
+
 	return REDISMODULE_OK;
 }
diff --git a/tests/module/LLAPI_DAG.c b/tests/module/LLAPI_DAG.c
@@ -286,10 +286,9 @@ static int _testSimpleDAGRun2Error(RedisModuleCtx *ctx, RAI_DAGRunCtx *run_info)
     int res = LLAPIMODULE_ERR;
 
     RedisModuleKey *key;
-    RAI_Tensor *tensor = _getFromKeySpace(ctx, "a{1}", &key);
+    RAI_Tensor *tensor = (RAI_Tensor*) _getFromKeySpace(ctx, "a{1}", &key);
     RedisModule_CloseKey(key);
     RedisAI_DAGAddTensorSet(run_info, "input1", tensor);
-    //RedisAI_DAGLoadTensor(run_info, "a{1}", err);
 
     // The script myscript{1} should exist in key space.
     RAI_Script *script = (RAI_Script*)_getFromKeySpace(ctx, "myscript{1}", &key);
@@ -383,3 +382,86 @@ int RAI_llapi_DAGRun(RedisModuleCtx *ctx, RedisModuleString **argv, int argc) {
     }
     return RedisModule_ReplyWithSimpleString(ctx, "DAG run success");
 }
+
+int RAI_llapi_DAG_resnet(RedisModuleCtx *ctx, RedisModuleString **argv, int argc) {
+    REDISMODULE_NOT_USED(argv);
+
+    if(argc > 1) {
+        RedisModule_WrongArity(ctx);
+        return REDISMODULE_OK;
+    }
+    RAI_Error* err;
+    RedisAI_InitError(&err);
+    RAI_Tensor** outputs = array_new(RAI_Tensor *, 1);
+    RAI_Error** opsStatus = array_new(RAI_Error *, 1);
+    DAGRunResults results = {.outputs = outputs, .opsStatus = opsStatus};
+    const char *test_res = "DAG resnet failed";
+
+    // Build the DAG with LOAD->SCRIPTRUN->MODELRUN->MODELRUN-SCRIPTRUN->SCRIPTRUN->TENSORGET
+    RAI_DAGRunCtx *run_info = RedisAI_DAGRunCtxCreate();
+    int status = RedisAI_DAGLoadTensor(run_info, "image:{{1}}", err);
+    if (status != REDISMODULE_OK) goto cleanup;
+
+    RedisModuleKey *key;
+    RAI_Script *script = (RAI_Script *)_getFromKeySpace(ctx,  "imagenet_script1:{{1}}", &key);
+    RedisModule_CloseKey(key);
+    RAI_DAGRunOp *script_op = RedisAI_DAGCreateScriptRunOp(script, "pre_process_3ch");
+    RedisAI_DAGRunOpAddInput(script_op, "image:{{1}}");
+    RedisAI_DAGRunOpAddOutput(script_op, "tmp_key:{{1}}");
+    RedisAI_DAGAddRunOp(run_info, script_op, err);
+
+    RAI_Model *model = (RAI_Model *)_getFromKeySpace(ctx,  "imagenet_model1:{{1}}", &key);
+    RedisModule_CloseKey(key);
+    RAI_DAGRunOp *model_op = RedisAI_DAGCreateModelRunOp(model);
+    RedisAI_DAGRunOpAddInput(model_op, "tmp_key:{{1}}");
+    RedisAI_DAGRunOpAddOutput(model_op, "tmp_key2_0");
+    status = RedisAI_DAGAddRunOp(run_info, model_op, err);
+    if (status != REDISMODULE_OK) goto cleanup;
+
+    model = (RAI_Model *)_getFromKeySpace(ctx,  "imagenet_model2:{{1}}", &key);
+    RedisModule_CloseKey(key);
+    model_op = RedisAI_DAGCreateModelRunOp(model);
+    RedisAI_DAGRunOpAddInput(model_op, "tmp_key:{{1}}");
+    RedisAI_DAGRunOpAddOutput(model_op, "tmp_key2_1");
+    status = RedisAI_DAGAddRunOp(run_info, model_op, err);
+    if (status != REDISMODULE_OK) goto cleanup;
+
+    script_op = RedisAI_DAGCreateScriptRunOp(script, "ensemble");
+    RedisAI_DAGRunOpAddInput(script_op, "tmp_key2_0");
+    RedisAI_DAGRunOpAddInput(script_op, "tmp_key2_1");
+    RedisAI_DAGRunOpAddOutput(script_op, "tmp_key_1:{{1}}");
+    RedisAI_DAGAddRunOp(run_info, script_op, err);
+
+    script_op = RedisAI_DAGCreateScriptRunOp(script, "post_process");
+    RedisAI_DAGRunOpAddInput(script_op, "tmp_key_1:{{1}}");
+    RedisAI_DAGRunOpAddOutput(script_op, "output:{{1}}");
+    RedisAI_DAGAddRunOp(run_info, script_op, err);
+
+    RedisAI_DAGAddTensorGet(run_info, "output:{{1}}", err);
+
+    pthread_mutex_lock(&global_lock);
+    if (RedisAI_DAGRun(run_info, _DAGFinishFunc, &results, err) != REDISMODULE_OK) {
+        pthread_mutex_unlock(&global_lock);
+        goto cleanup;
+    }
+    // Wait until the onFinish callback returns.
+    pthread_cond_wait(&global_cond, &global_lock);
+    pthread_mutex_unlock(&global_lock);
+
+    // Verify that we received the expected output.
+    RedisModule_Assert(array_len(results.outputs) == 1);
+    RAI_Tensor *out_tensor = outputs[0];
+    long long val;
+    if(RedisAI_TensorGetValueAsLongLong(out_tensor, 0, &val) != 0) goto cleanup;
+    RedisAI_TensorFree(out_tensor);
+    if (0<= val && val <= 1000) {
+        test_res = "DAG resnet success";
+    }
+
+    cleanup:
+    RedisAI_FreeError(err);
+    array_free(results.outputs);
+    array_free(results.opsStatus);
+    RedisAI_DAGFree(run_info);
+    return RedisModule_ReplyWithSimpleString(ctx, test_res);
+}

Original file line number	Diff line number	Diff line change
`@@ -264,5 +264,10 @@ int RedisModule_OnLoad(RedisModuleCtx ctx, RedisModuleString *argv, int argc)`
`264`	`264`	`return REDISMODULE_ERR;`
`265`	`265`	`}`
`266`	`266`
	`267`	`+ if(RedisModule_CreateCommand(ctx, "RAI_llapi.DAG_resnet", RAI_llapi_DAG_resnet, "",`
	`268`	`+ 0, 0, 0) == REDISMODULE_ERR) {`
	`269`	`+ return REDISMODULE_ERR;`
	`270`	`+ }`
	`271`	`+`
`267`	`272`	`return REDISMODULE_OK;`
`268`	`273`	`}`