exteded testsuite with test that deterministically crashes redisai (issue #323) (#324)

* [add] added test that deterministically crashes redisai (issue #323)

* Handle the case of batch size 1 minimally

Co-authored-by: Luca Antiga <luca.antiga@orobix.com>

Author: filipecosta90

src/redisai.c

@@ -205,7 +205,10 @@ void RedisAI_FreeRunStats(RedisModuleCtx *ctx, struct RedisAI_RunStats *rstats)
 }
 
 void *RedisAI_RunSession(struct RedisAI_RunInfo **batch_rinfo) {
-  if (array_len(batch_rinfo) == 0) {
+
+  const long long batch_size = array_len(batch_rinfo);
+
+  if (batch_size == 0) {
     return NULL;
   }
 
@@ -215,10 +218,15 @@ void *RedisAI_RunSession(struct RedisAI_RunInfo **batch_rinfo) {
   RAI_ModelRunCtx* mctx = NULL;
   RAI_ScriptRunCtx* sctx = NULL;
   if (batch_rinfo[0]->mctx) {
-    mctx = RAI_ModelRunCtxCreate(batch_rinfo[0]->mctx->model);
-    for (long long i=0; i<array_len(batch_rinfo); i++) {
-      int id = RAI_ModelRunCtxAddBatch(mctx);
-      RAI_ModelRunCtxCopyBatch(mctx, id, batch_rinfo[i]->mctx, 0);
+    if (batch_size > 1) {
+      mctx = RAI_ModelRunCtxCreate(batch_rinfo[0]->mctx->model);
+      for (long long i=0; i<batch_size; i++) {
+        int id = RAI_ModelRunCtxAddBatch(mctx);
+        RAI_ModelRunCtxCopyBatch(mctx, id, batch_rinfo[i]->mctx, 0);
+      }
+    }
+    else {
+      mctx = batch_rinfo[0]->mctx;
     }
   }
   else if (batch_rinfo[0]->sctx) {
@@ -235,19 +243,24 @@ void *RedisAI_RunSession(struct RedisAI_RunInfo **batch_rinfo) {
   }
   rtime = ustime() - start;
 
-  for (long long i=0; i<array_len(batch_rinfo); i++) {
+  for (long long i=0; i<batch_size; i++) {
     struct RedisAI_RunInfo *rinfo = batch_rinfo[i];
     if (mctx) {
-      size_t noutputs = RAI_ModelRunCtxNumOutputs(mctx);
-      for (long long o=0; o<noutputs; o++) {
-        RAI_Tensor* tensor = mctx->batches[i].outputs[o].tensor;
-        if (tensor) {
-          rinfo->mctx->batches[0].outputs[o].tensor = RAI_TensorGetShallowCopy(tensor);
-        }
-        else {
-          rinfo->mctx->batches[0].outputs[o].tensor = NULL;
+      if (batch_size > 1) {
+        size_t noutputs = RAI_ModelRunCtxNumOutputs(mctx);
+        for (long long o=0; o<noutputs; o++) {
+          RAI_Tensor* tensor = mctx->batches[i].outputs[o].tensor;
+          if (tensor) {
+            rinfo->mctx->batches[0].outputs[o].tensor = RAI_TensorGetShallowCopy(tensor);
+          }
+          else {
+            rinfo->mctx->batches[0].outputs[o].tensor = NULL;
+          }
         }
       }
+      else {
+        // Do nothing if no batching
+      }
     }
     else if (sctx) {
       // No batching for scripts for now
@@ -270,7 +283,12 @@ void *RedisAI_RunSession(struct RedisAI_RunInfo **batch_rinfo) {
   }
 
   if (mctx) {
-    RAI_ModelRunCtxFree(mctx);
+    if (batch_size > 1) {
+      RAI_ModelRunCtxFree(mctx);
+    }
+    else {
+      // Do nothing if no batching
+    }
   }
   else if (sctx) {
     // No batching for scripts for now

test/tests_tensorflow.py

@@ -708,3 +708,45 @@ def test_tensorflow_modelrun_financialNet(env):
                                       'referenceTensor:{}'.format(tensor_number), 'OUTPUTS',
                                       'classificationTensor:{}_{}'.format(tensor_number, repetition))
             env.assertEqual(ret, b'OK')
+
+
+def test_tensorflow_modelrun_financialNet_multiproc(env):
+    con = env.getConnection()
+
+    model_pb, creditcard_transactions, creditcard_referencedata = load_creditcardfraud_data(env)
+
+    tensor_number = 1
+    for transaction_tensor in creditcard_transactions:
+        ret = con.execute_command('AI.TENSORSET', 'transactionTensor:{0}'.format(tensor_number),
+                                  'FLOAT', 1, 30,
+                                  'BLOB', transaction_tensor.tobytes())
+        env.assertEqual(ret, b'OK')
+        tensor_number = tensor_number + 1
+
+    tensor_number = 1
+    for reference_tensor in creditcard_referencedata:
+        ret = con.execute_command('AI.TENSORSET', 'referenceTensor:{0}'.format(tensor_number),
+                                  'FLOAT', 1, 256,
+                                  'BLOB', reference_tensor.tobytes())
+        env.assertEqual(ret, b'OK')
+        tensor_number = tensor_number + 1
+
+    ret = con.execute_command('AI.MODELSET', 'financialNet', 'TF', "CPU",
+                              'INPUTS', 'transaction', 'reference', 'OUTPUTS', 'output', model_pb)
+    env.assertEqual(ret, b'OK')
+
+    def functor_financialNet(env, key_max, repetitions):
+        for tensor_number in range(1, key_max):
+            for repetition in range(1, repetitions):
+                ret = env.execute_command('AI.MODELRUN', 'financialNet', 'INPUTS',
+                                        'transactionTensor:{}'.format(tensor_number),
+                                        'referenceTensor:{}'.format(tensor_number), 'OUTPUTS',
+                                        'classificationTensor:{}_{}'.format(tensor_number, repetition))
+    
+    t = time.time()
+    run_test_multiproc(env, 10,
+                       lambda env: functor_financialNet(env,len(transaction_tensor),100) )
+    elapsed_time = time.time() - t
+    total_ops = len(transaction_tensor)*100
+    avg_ops_sec = total_ops/elapsed_time
+    env.debugPrint("AI.MODELRUN elapsed time(sec) {:6.2f}\tTotal ops  {:10.2f}\tAvg. ops/sec {:10.2f}".format(elapsed_time, total_ops, avg_ops_sec), True)