WIP - AI.DAGRUN, commands code refactoring and improvement (#322)

* [wip] refactored TENSORGET, TENSORSET, and MODELRUN  to be re-used by AI.DAGRUN

* [add] first version of dagrun with modelrun and persist working

* [wip] refactored non-command methods within redisai.c into dag, run_info, background_workers, and model_script_run_session

* [fix] fixed wrong includes

* [add] adding init methods to RAI_DagOp and RedisAI_RunInfo

* [wip] ai.tensorset, PERSIST and LOAD working as expected

* [add] dagrun's tensorset and tensorget working as expected

* [add] extended test for tensorset and tensorget

* [wip] wip on modelrun within dagrun

* [wip] fist version of tensorget |> modelrun |> tensorget working

* [add] refactor RunInfo, so that instead of a single mctx and sctx pointer, it holds an array of (mctx | sctx) pointers (within RAI_DagOp). To be gradually adopted on modelrun and scriptrun ( for now only on dagrun )

* [add] added redisai-py as a requirement for tests ( it helps testing for more complex patterns )

* [add] added test for semantics of values we reply from the local context (ensuring that write after read does not alter the tensorget read value)

* [wip] discover the DAGRUN device queue from the arguments of MODELRUN. If no MODELRUN default to CPU

* [fix] fxied wrong reference passing on RedisAI_Parse_ModelRun_RedisCommand

* [fix] fixed wrong reference passing on RedisAI_Parse_ModelRun_RedisCommand from RedisAI_DagRunSession

* [wip] wip on minor optimizations

* [add] exteded dag.h to have proper documentation

* [add] extended model_script_run_session header file with documentation to better describe the context in which RedisAI blocking commands MODELRUN and SCRIPTRUN operate

* [add] moved configuration properties and parsing out of redisai.c to config.h/c

* [add] backends_intra_op_parallelism and backends_inter_op_parallelism working as expected for TF

* [add] intra_op and inter_op parallelism working as expected for TF backend

* [add] exclude perf profile reports from git

* [add] wip on mem sanitizer

* [add] working on RAI_FreeRunInfo and RAI_FreeDagOp

* [add] using RAI_InitRunInfo on RedisAI_ScriptRun_RedisCommand

* [add] using array data type on RedisAI_RunInfo rinfo->outkeys

* [add] small leaks fix for dag

* [add] partial refactor of RedisAI_ScriptRun_RedisCommand to make usage of RedisAI_RunInfo helper methods ( consistent constructors and destructors among modelrun,scriptrun and dagrun)

* [add] kickoff negative testing of AI.DAGRUN

* [add] extended negative testing on dag and removed complexity of tensor datatype (removed possibility to retain RString)

* [add] extended AI.DAGRUN negative testing. fixed negative testing leaks

* [add] more extensive tests and severall touches on same keys on AI.DAGRUN ci

* Fixes for macOS and in general (#327)

* Prevent a DAG run info to be considered batchable

* Ensure sync on failing ONNX test

Co-authored-by: Luca Antiga <[email protected]>

Author: filipecosta90

.circleci/config.yml

@@ -37,6 +37,7 @@ commands:
           command: |
             mkdir -p ~/workspace/tests
             make -C opt test SHOW=1
+          no_output_timeout: 20m
       - run:
           name: Package
           command: make -C opt pack BRANCH="${CIRCLE_BRANCH//[^A-Za-z0-9._-]/_}" INTO=~/workspace/packages SHOW=1

.gitignore

@@ -3,6 +3,7 @@
 /build/
 /install*
 /test/venv/
+logs/
 /test/logs/
 .venv/
 venv*/
@@ -13,6 +14,9 @@ venv*/
 *.tar.gz
 /VARIANT
 
+### Cmake auto tools
+cmake-build-debug
+
 # Misc
 .DS_Store
 *.swp
@@ -73,6 +77,10 @@ __pycache__
 *.idb
 *.pdb
 
+# Debug/Profile files
+# ignore perf html reports
+*.html 
+
 # Kernel Module Compile Results
 *.mod*
 *.cmd
@@ -86,6 +94,10 @@ dkms.conf
 # Covers JetBrains IDEs: IntelliJ, RubyMine, PhpStorm, AppCode, PyCharm, CLion, Android Studio and Webstorm
 # Reference: https://intellij-support.jetbrains.com/hc/en-us/articles/206544839
 
+.idea
+.idea/
+.idea/*
+
 # User-specific stuff:
 .idea/workspace.xml
 .idea/tasks.xml

src/CMakeLists.txt

@@ -2,7 +2,13 @@ ADD_LIBRARY(redisai_obj OBJECT
         util/dict.c
         util/queue.c
         redisai.c
+        run_info.c
+        background_workers.c
+        model_script_run_session.c
+        config.c
+        dag.c
         backends.c
+        backends/util.c
         model.c
         err.c
         script.c
@@ -13,13 +19,14 @@ ADD_LIBRARY(redisai_obj OBJECT
         rmutil/args.c
         rmutil/heap.c
         rmutil/priority_queue.c
-        rmutil/vector.c)
+        rmutil/vector.c run_info.c)
 
 IF(BUILD_TF)
     ADD_LIBRARY(redisai_tensorflow_obj OBJECT
             backends/tensorflow.c
             backends/util.c
             err.c
+            util/dict.c
             tensor.c)
 ENDIF()
 
@@ -28,6 +35,7 @@ IF(BUILD_TFLITE)
             backends/tflite.c
             backends/util.c
             err.c
+            util/dict.c
             tensor.c)
 ENDIF()
 
@@ -36,6 +44,7 @@ IF(BUILD_TORCH)
             backends/torch.c
             backends/util.c
             err.c
+            util/dict.c
             tensor.c)
 ENDIF()
 
@@ -44,6 +53,7 @@ IF(BUILD_ORT)
             backends/onnxruntime.c
             backends/util.c
             err.c
+            util/dict.c
             tensor.c)
 ENDIF()
 

src/backends/tensorflow.c

@@ -2,6 +2,7 @@
 #include "backends/util.h"
 #include "tensor.h"
 #include "util/arr_rm_alloc.h"
+#include "model.h"
 
 #include "tensorflow/c/c_api.h"
 
@@ -292,29 +293,57 @@ RAI_Model *RAI_ModelCreateTF(RAI_Backend backend, const char* devicestr, RAI_Mod
 
   if (device == RAI_DEVICE_CPU) {
     // Set number of GPU to 0 with
-    // config.device_count = {'GPU': 0} 
-    uint8_t config[9] = {0x0a, 0x07, 0x0a, 0x03, 0x47, 0x50, 0x55, 0x10, 0x00};
-    TF_SetConfig(sessionOptions, (void *)config, 9, status);
-  }
-  else if (device == RAI_DEVICE_GPU) {
+    // config.device_count = {'GPU': 0}
+    uint8_t config[] = {0x0a, 0x07, 0x0a, 0x03, 0x47, 0x50, 0x55, 0x10, 0x00};
+    TF_SetConfig(sessionOptions, (void *)config, sizeof(config), optionsStatus);
+
+    if (TF_GetCode(optionsStatus) != TF_OK) {
+      RAI_SetError(error, RAI_EMODELCONFIGURE,
+                   RedisModule_Strdup(TF_Message(optionsStatus)));
+      // TODO: free memory
+      return NULL;
+    }
+
+    if (opts.backends_intra_op_parallelism > 0) {
+      uint8_t proto[] = {0x10, (uint8_t)opts.backends_intra_op_parallelism};
+      TF_SetConfig(sessionOptions, proto, sizeof(proto), optionsStatus);
+      if (TF_GetCode(optionsStatus) != TF_OK) {
+        RAI_SetError(error, RAI_EMODELCONFIGURE,
+                     RedisModule_Strdup(TF_Message(optionsStatus)));
+        // TODO: free memory
+        return NULL;
+      }
+    }
+
+    if (opts.backends_inter_op_parallelism > 0) {
+      uint8_t proto1[] = {0x28, (uint8_t)opts.backends_inter_op_parallelism};
+      TF_SetConfig(sessionOptions, proto1, sizeof(proto1), optionsStatus);
+      if (TF_GetCode(optionsStatus) != TF_OK) {
+        RAI_SetError(error, RAI_EMODELCONFIGURE,
+                     RedisModule_Strdup(TF_Message(optionsStatus)));
+        // TODO: free memory
+        return NULL;
+      }
+    }
+  } else if (device == RAI_DEVICE_GPU) {
     if (deviceid == -1) {
       // Set
       // config.gpu_options.allow_growth = True
       uint8_t config[4] = {0x32, 0x02, 0x20, 0x01};
-      TF_SetConfig(sessionOptions, (void *)config, 4, status);
+      TF_SetConfig(sessionOptions, (void *)config, 4, optionsStatus);
     }
     else {
       // Set
       // config.gpu_options.allow_growth = True
       // config.gpu_options.visible_device_list = '<deviceid>'
       uint8_t config[7] = {0x32, 0x05, 0x20, 0x01, 0x2a, 0x01, 0x30};
       config[6] += (uint8_t)deviceid;
-      TF_SetConfig(sessionOptions, (void *)config, 7, status);
+      TF_SetConfig(sessionOptions, (void *)config, 7, optionsStatus);
     }
   }
 
   if (TF_GetCode(optionsStatus) != TF_OK) {
-    RAI_SetError(error, RAI_EMODELCONFIGURE, RedisModule_Strdup(TF_Message(status)));
+    RAI_SetError(error, RAI_EMODELCONFIGURE, RedisModule_Strdup(TF_Message(optionsStatus)));
     // TODO: free memory
     return NULL;
   }
@@ -437,7 +466,7 @@ int RAI_ModelRunTF(RAI_ModelRunCtx** mctxs, RAI_Error *error) {
 
   size_t batch_sizes[nbatches];
   size_t batch_offsets[nbatches];
-  if (array_len(mctxs[0]->inputs) > 0) {
+  if (ninputs > 0) {
     for (size_t b=0; b<nbatches; ++b) {
       batch_sizes[b] = RAI_TensorDim(mctxs[b]->inputs[0].tensor, 0);
     }
@@ -453,7 +482,6 @@ int RAI_ModelRunTF(RAI_ModelRunCtx** mctxs, RAI_Error *error) {
     for (size_t b=0; b<nbatches; ++b) {
       batched_input_tensors[b] = mctxs[b]->inputs[i].tensor;
     }
-    // inputTensorsValues[i] = RAI_TFTensorFromTensor(mctx->inputs[i].tensor);
     inputTensorsValues[i] = RAI_TFTensorFromTensors(batched_input_tensors, nbatches);
     TF_Output port;
     port.oper = TF_GraphOperationByName(mctxs[0]->model->model, mctxs[0]->inputs[i].name);
@@ -495,21 +523,11 @@ int RAI_ModelRunTF(RAI_ModelRunCtx** mctxs, RAI_Error *error) {
 
   for(size_t i=0; i<noutputs; ++i) {
     for (size_t b=0; b<nbatches; b++) {
-      RAI_Tensor* output_tensor = RAI_TensorCreateFromTFTensor(outputTensorsValues[i], batch_offsets[b], batch_sizes[b]);
-      mctxs[b]->outputs[i].tensor = RAI_TensorGetShallowCopy(output_tensor);
-      RAI_TensorFree(output_tensor);
+      mctxs[b]->outputs[i].tensor = RAI_TensorCreateFromTFTensor(outputTensorsValues[i], batch_offsets[b], batch_sizes[b]);
     }
     TF_DeleteTensor(outputTensorsValues[i]);
   }
 
-  // TODO: add (make sure we deallocate once)
-  // for (size_t i=0 ; i<array_len(mctx->inputs); ++i) {
-  //   TF_DeleteTensor(inputTensorsValues[i]);
-  // }
-  // for (size_t i=0 ; i<array_len(mctx->outputs); ++i) {
-  //   TF_DeleteTensor(outputTensorsValues[i]);
-  // }
-
   TF_DeleteStatus(status);
 
   return 0;

src/backends/util.c

@@ -1,21 +1,20 @@
 #include "backends/util.h"
 
-int parseDeviceStr(const char* devicestr, RAI_Device* device, int64_t* deviceid) {
+int parseDeviceStr(const char* devicestr, RAI_Device* device,
+                   int64_t* deviceid) {
   if (strcasecmp(devicestr, "CPU") == 0) {
     *device = RAI_DEVICE_CPU;
     *deviceid = -1;
-  }
-  else if (strcasecmp(devicestr, "GPU") == 0) {
+  } else if (strcasecmp(devicestr, "GPU") == 0) {
     *device = RAI_DEVICE_GPU;
     *deviceid = -1;
-  }
-  else if (strncasecmp(devicestr, "GPU:", 4) == 0) {
+  } else if (strncasecmp(devicestr, "GPU:", 4) == 0) {
     *device = RAI_DEVICE_GPU;
     sscanf(devicestr, "GPU:%lld", deviceid);
-  }
-  else {
+  } else {
     return 0;
   }
 
   return 1;
 }
+

src/backends/util.h

@@ -1,11 +1,13 @@
 #ifndef SRC_BACKENDS_UTIL_H_
 #define SRC_BACKENDS_UTIL_H_
 
-#include "config.h"
 #include <stdint.h>
 #include <stdio.h>
 #include <strings.h>
 
-int parseDeviceStr(const char* devicestr, RAI_Device* device, int64_t* deviceid);
+#include "config.h"
+
+int parseDeviceStr(const char* devicestr, RAI_Device* device,
+                   int64_t* deviceid);
 
 #endif /* SRC_BACKENDS_UTIL_H_ */