Small refactoring stage 2

Moved graph run into graph object, make use of refcount for tensors and for graph.
Added utils directory with arr implementation that provide dynamic
array object.

Author: MeirShpilraien

src/graph.c

@@ -1,7 +1,19 @@
 #include "graph.h"
+#include "utils/arr_rm_alloc.h"
 
 RedisModuleType *RedisDL_GraphType = NULL;
 
+typedef struct RDL_GraphCtxParam{
+  TF_Output name;
+  RDL_Tensor* tensor;
+}RDL_GraphCtxParam;
+
+typedef struct RDL_GraphRunCtx{
+  RDL_Graph* graph;
+  RDL_GraphCtxParam* inputs;
+  RDL_GraphCtxParam* outputs;
+}RDL_GraphRunCtx;
+
 static void* Graph_RdbLoad(struct RedisModuleIO *io, int encver){
   //todo
   return NULL;
@@ -75,6 +87,9 @@ RDL_Graph* Graph_Create(const char* prefix, const char* graphdef, size_t graphle
 }
 
 void Graph_Free(RDL_Graph* graph){
+  if(--graph->refCount > 0){
+    return;
+  }
   TF_Status *status = TF_NewStatus();
   TF_CloseSession(graph->session, status);
 
@@ -100,3 +115,102 @@ void Graph_Free(RDL_Graph* graph){
 
   RedisModule_Free(graph);
 }
+
+RDL_GraphRunCtx* Graph_RunCtxCreate(RDL_Graph* graph){
+#define PARAM_INITIAL_SIZE 10
+  RDL_GraphRunCtx* gctx = RedisModule_Alloc(sizeof(*gctx));
+  gctx->graph = Graph_GetShallowCopy(graph);
+  gctx->inputs = array_new(RDL_GraphCtxParam, PARAM_INITIAL_SIZE);
+  gctx->outputs = array_new(RDL_GraphCtxParam, PARAM_INITIAL_SIZE);
+  return gctx;
+}
+
+static int Graph_RunCtxAddParam(RDL_GraphRunCtx* gctx, RDL_GraphCtxParam* paramArr, const char* name, RDL_Tensor* tensor){
+  TF_Output port;
+  port.oper = TF_GraphOperationByName(gctx->graph->graph, name);
+  port.index = 0;
+  if(port.oper == NULL){
+    return 0;
+  }
+  RDL_GraphCtxParam param = {
+      .name = port,
+      .tensor = tensor ? Tensor_GetShallowCopy(tensor): NULL,
+  };
+  paramArr = array_append(paramArr, param);
+  return 1;
+}
+
+int Graph_RunCtxAddInput(RDL_GraphRunCtx* gctx, const char* inputName, RDL_Tensor* inputTensor){
+  return Graph_RunCtxAddParam(gctx, gctx->inputs, inputName, inputTensor);
+}
+
+int Graph_RunCtxAddOutput(RDL_GraphRunCtx* gctx, const char* outputName){
+  return Graph_RunCtxAddParam(gctx, gctx->outputs, outputName, NULL);
+}
+
+size_t Graph_RunCtxNumOutputs(RDL_GraphRunCtx* gctx){
+  return array_len(gctx->outputs);
+}
+
+RDL_Tensor* Graph_RunCtxOutputTensor(RDL_GraphRunCtx* gctx, size_t index){
+  assert(Graph_RunCtxNumOutputs(gctx) > index && index >= 0);
+  return gctx->outputs[index].tensor;
+}
+
+void Graph_RunCtxFreeInternals(RDL_GraphRunCtx* gctx){
+  for(size_t i = 0 ; i < array_len(gctx->inputs) ; ++i){
+    Tensor_Free(gctx->inputs[i].tensor);
+  }
+  array_free(gctx->inputs);
+
+  for(size_t i = 0 ; i < array_len(gctx->outputs) ; ++i){
+    if(gctx->outputs[i].tensor){
+      Tensor_Free(gctx->outputs[i].tensor);
+    }
+  }
+  array_free(gctx->outputs);
+
+  Graph_Free(gctx->graph);
+}
+
+int Graph_Run(RDL_GraphRunCtx* gctx){
+  TF_Status *status = TF_NewStatus();
+
+  TF_Tensor* inputTensorsValues[array_len(gctx->inputs)];
+  TF_Output inputs[array_len(gctx->inputs)];
+  TF_Tensor* outputTensorsValues[array_len(gctx->outputs)];
+  TF_Output outputs[array_len(gctx->outputs)];
+
+  for(size_t i = 0 ; i < array_len(gctx->inputs) ; ++i){
+    inputTensorsValues[i] = gctx->inputs[i].tensor->tensor;
+    inputs[i] = gctx->inputs[i].name;
+  }
+
+  for(size_t i = 0 ; i < array_len(gctx->outputs) ; ++i){
+    outputs[i] = gctx->outputs[i].name;
+  }
+
+  TF_SessionRun(gctx->graph->session, NULL /* run_options */,
+                inputs, inputTensorsValues, array_len(gctx->inputs),
+                outputs, outputTensorsValues, array_len(gctx->outputs),
+                NULL /* target_opers */, 0 /* ntargets */,
+                NULL /* run_Metadata */,
+                status);
+
+  if (TF_GetCode(status) != TF_OK) {
+    TF_DeleteStatus(status);
+    return 0;
+  }
+
+  for(size_t i = 0 ; i < array_len(gctx->outputs) ; ++i){
+    gctx->outputs[i].tensor = Tensor_CreateFromTensor(outputTensorsValues[i]);
+  }
+
+  TF_DeleteStatus(status);
+  return 1;
+}
+
+RDL_Graph* Graph_GetShallowCopy(RDL_Graph* graph){
+  ++graph->refCount;
+  return graph;
+}

src/graph.h

@@ -10,6 +10,7 @@
 
 #include "tensorflow/c/c_api.h"
 #include "redismodule.h"
+#include "tensor.h"
 
 typedef struct RDL_Graph{
   TF_Graph* graph;
@@ -20,11 +21,23 @@ typedef struct RDL_Graph{
   size_t refCount;
 }RDL_Graph;
 
+typedef struct RDL_GraphCtxParam RDL_GraphCtxParam;
+
+typedef struct RDL_GraphRunCtx RDL_GraphRunCtx;
+
 extern RedisModuleType *RedisDL_GraphType;
 
 int Graph_Init(RedisModuleCtx* ctx);
 RDL_Graph* Graph_Create(const char* prefix, const char* graphdef, size_t graphlen);
 void Graph_Free(RDL_Graph* graph);
+RDL_GraphRunCtx* Graph_RunCtxCreate(RDL_Graph* graph);
+int Graph_RunCtxAddInput(RDL_GraphRunCtx* gctx, const char* inputName, RDL_Tensor* inputTensor);
+int Graph_RunCtxAddOutput(RDL_GraphRunCtx* gctx, const char* outputName);
+size_t Graph_RunCtxNumOutputs(RDL_GraphRunCtx* gctx);
+RDL_Tensor* Graph_RunCtxOutputTensor(RDL_GraphRunCtx* gctx, size_t index);
+void Graph_RunCtxFreeInternals(RDL_GraphRunCtx* gctx);
+int Graph_Run(RDL_GraphRunCtx* gctx);
+RDL_Graph* Graph_GetShallowCopy(RDL_Graph* graph);
 
 
 

src/redisdl.c

@@ -93,20 +93,6 @@ mstime_t mstime(void) {
     return ustime()/1000;
 }
 
-TF_Tensor* RedisDL_clone(RedisModuleCtx *ctx, const TF_Tensor *tensor) {
-  int ndims = TF_NumDims(tensor);
-  long long *dims = RedisModule_PoolAlloc(ctx, ndims * sizeof(long long));
-  for (int j=0; j<ndims; j++) {
-    dims[j] = TF_Dim(tensor, j);
-  }
-  size_t len = TF_TensorByteSize(tensor);
-  void *data = TF_TensorData(tensor);
-  TF_Tensor *out = TF_AllocateTensor(TF_TensorType(tensor),
-                                     dims, ndims, len);
-  memcpy(TF_TensorData(out), data, len);
-  return out;
-}
-
 enum RedisDL_DataFmt {
   REDISDL_DATA_BLOB = 0,
   REDISDL_DATA_VALUES
@@ -493,28 +479,19 @@ int RedisDL_GSet_RedisCommand(RedisModuleCtx *ctx, RedisModuleString **argv, int
 
 struct RedisDL_RunInfo {
   RedisModuleBlockedClient *client;
-  TF_Session *session;
-  TF_Output *inputs;
-  TF_Tensor **input_values;
-  long long ninputs;
-  TF_Output *outputs;
-  TF_Tensor **output_values;
-  long long noutputs;
-  RedisModuleKey *graphkey;
   RedisModuleString **outkeys;
-  TF_Status *status;
+  RDL_GraphRunCtx* gctx;
+  int status;
 };
 
-void RedisDL_FreeRunInfo(struct RedisDL_RunInfo *rinfo) {
-  RedisModule_Free(rinfo->inputs);
-  for (int i=0; i<rinfo->ninputs; i++) {
-    TF_DeleteTensor(rinfo->input_values[i]);
+void RedisDL_FreeRunInfo(RedisModuleCtx *ctx, struct RedisDL_RunInfo *rinfo) {
+  for(int i = 0 ; i < Graph_RunCtxNumOutputs(rinfo->gctx) ; ++i){
+    RedisModule_FreeString(ctx, rinfo->outkeys[i]);
   }
-  RedisModule_Free(rinfo->input_values);
-  RedisModule_Free(rinfo->outputs);
-  RedisModule_Free(rinfo->output_values);
   RedisModule_Free(rinfo->outkeys);
-  TF_DeleteStatus(rinfo->status);
+
+  Graph_RunCtxFreeInternals(rinfo->gctx);
+
   RedisModule_Free(rinfo);
 }
 
@@ -525,24 +502,19 @@ void *RedisDL_RunSession(void *arg) {
 
   mstime_t start = mstime();
 
-  TF_Status *status = TF_NewStatus();
 
-  TF_SessionRun(rinfo->session, NULL /* run_options */,
-                rinfo->inputs, rinfo->input_values, rinfo->ninputs,
-                rinfo->outputs, rinfo->output_values, rinfo->noutputs,
-                NULL /* target_opers */, 0 /* ntargets */,
-                NULL /* run_Metadata */,
-                rinfo->status);
+  rinfo->status = Graph_Run(rinfo->gctx);
 
   mstime_t end = mstime();
 
   RedisModule_ThreadSafeContextLock(ctx);
   RedisModule_Log(ctx, "notice", "TF_SessionRun took %fs", (end - start) / 1000.0);
   RedisModule_ThreadSafeContextUnlock(ctx);
 
-  RedisModule_FreeThreadSafeContext(ctx);
   RedisModule_UnblockClient(rinfo->client, rinfo);
 
+  RedisModule_FreeThreadSafeContext(ctx);
+
   return NULL;
 }
 
@@ -557,31 +529,33 @@ int RedisDL_Run_Reply(RedisModuleCtx *ctx, RedisModuleString **argv, int argc) {
   REDISMODULE_NOT_USED(argc);
   struct RedisDL_RunInfo *rinfo = RedisModule_GetBlockedClientPrivateData(ctx);
 
-  if (TF_GetCode(rinfo->status) != TF_OK) {
-    int ret = RedisModule_ReplyWithError(ctx, TF_Message(rinfo->status));
-    RedisDL_FreeRunInfo(rinfo);
+  if (!rinfo->status) {
+    int ret = RedisModule_ReplyWithError(ctx, "graph run failed");
+    RedisDL_FreeRunInfo(ctx, rinfo);
     return ret;
   }
 
-  for (int i=0; i<rinfo->noutputs; i++) {
+  for (size_t i=0; i<Graph_RunCtxNumOutputs(rinfo->gctx); ++i){
     RedisModuleKey *outkey = RedisModule_OpenKey(ctx, rinfo->outkeys[i],
                                                  REDISMODULE_READ|REDISMODULE_WRITE);
     int type = RedisModule_KeyType(outkey);
     if (type != REDISMODULE_KEYTYPE_EMPTY &&
         RedisModule_ModuleTypeGetType(outkey) != RedisDL_TensorType) {
       RedisModule_CloseKey(outkey);
-      RedisDL_FreeRunInfo(rinfo);
+      RedisDL_FreeRunInfo(ctx, rinfo);
       return RedisModule_ReplyWithError(ctx, REDISMODULE_ERRORMSG_WRONGTYPE);
     }
-    RDL_Tensor *t = Tensor_CreateFromTensor(rinfo->output_values[i]);
-    RedisModule_ModuleTypeSetValue(outkey, RedisDL_TensorType, t);
+    RDL_Tensor *t = Graph_RunCtxOutputTensor(rinfo->gctx, i);
+    if(t){
+      RedisModule_ModuleTypeSetValue(outkey, RedisDL_TensorType, Tensor_GetShallowCopy(t));
+    }
     RedisModule_CloseKey(outkey);
   }
 
   // FIXME This crashes Redis, we need to investigate.
   //RedisModule_CloseKey(rinfo->graphkey);
 
-  RedisDL_FreeRunInfo(rinfo);
+  RedisDL_FreeRunInfo(ctx, rinfo);
 
   return RedisModule_ReplyWithSimpleString(ctx, "OK");
 }
@@ -610,9 +584,6 @@ int RedisDL_GRun_RedisCommand(RedisModuleCtx *ctx, RedisModuleString **argv, int
 
   RDL_Graph *gto = RedisModule_ModuleTypeGetValue(key);
 
-  TF_Graph* graph = gto->graph;
-  TF_Session* session = gto->session;
-
   long long ninputs;
   if ((RedisModule_StringToLongLong(argv[2], &ninputs) != REDISMODULE_OK)
       || ninputs < 0) {
@@ -634,64 +605,42 @@ int RedisDL_GRun_RedisCommand(RedisModuleCtx *ctx, RedisModuleString **argv, int
     return RedisModule_ReplyWithError(ctx, "ERR odd key/name pairs");
   }
 
-  TF_Output *inputs = RedisModule_Alloc(ninputs*sizeof(TF_Output));
-  TF_Output *outputs = RedisModule_Alloc(noutputs*sizeof(TF_Output));
-  TF_Tensor **input_values = RedisModule_Alloc(ninputs*sizeof(TF_Tensor*));
-  TF_Tensor **output_values = RedisModule_Alloc(noutputs*sizeof(TF_Tensor*));
+  struct RedisDL_RunInfo *rinfo = RedisModule_Alloc(sizeof(struct RedisDL_RunInfo));
+  rinfo->gctx = Graph_RunCtxCreate(gto);
 
-  RedisModuleString **outkeys = RedisModule_Alloc(noutputs*sizeof(RedisModuleString*));
+  rinfo->outkeys = RedisModule_Alloc(noutputs*sizeof(RedisModuleString*));
 
   for (int i=pairoffset; i<argc; i+=2) {
     int isinput = i < pairoffset + 2 * ninputs;
 
-    size_t namelen;
     RedisModuleString* argname = argv[i+1];
 
     if (isinput) {
       RedisModuleKey *argkey = RedisModule_OpenKey(ctx, argv[i], REDISMODULE_READ);
       if (RedisModule_ModuleTypeGetType(argkey) != RedisDL_TensorType) {
+        // todo free rinfo
         RedisModule_CloseKey(argkey);
         return RedisModule_ReplyWithError(ctx, REDISMODULE_ERRORMSG_WRONGTYPE);
       }
       RDL_Tensor *t = RedisModule_ModuleTypeGetValue(argkey);
-      input_values[(i-pairoffset)/2] = RedisDL_clone(ctx, t->tensor);
       RedisModule_CloseKey(argkey);
-      const char* opname = RedisModule_StringPtrLen(argname, &namelen);
-      // RedisModule_Log(ctx, "warning", "%s", opname);
-      TF_Output port;
-      port.oper = TF_GraphOperationByName(graph, opname);
-      port.index = 0;
-      if (port.oper == NULL) {
+      const char* opname = RedisModule_StringPtrLen(argname, NULL);
+      if(!Graph_RunCtxAddInput(rinfo->gctx, opname, t)){
+        // todo free rinfo
         return RedisModule_ReplyWithError(ctx, "Input key not found.");
       }
-      inputs[(i-pairoffset)/2] = port;
     } else {
-      const char* opname = RedisModule_StringPtrLen(argname, &namelen);
-      TF_Output port;
-      port.oper = TF_GraphOperationByName(graph, opname);
-      port.index = 0;
-      if (port.oper == NULL) {
+      const char* opname = RedisModule_StringPtrLen(argname, NULL);
+      if(!Graph_RunCtxAddOutput(rinfo->gctx, opname)){
+        // todo free rinfo
         return RedisModule_ReplyWithError(ctx, "Output key not found.");
       }
-      outputs[(i-pairoffset)/2-ninputs] = port;
-      outkeys[(i-pairoffset)/2-ninputs] = argv[i];
+      RedisModule_RetainString(ctx, argv[i]);
+      rinfo->outkeys[(i-pairoffset)/2-ninputs] = argv[i];
     }
   }
 
-  RedisModuleBlockedClient *bc = RedisModule_BlockClient(ctx, RedisDL_Run_Reply, NULL, NULL, 0);
-
-  struct RedisDL_RunInfo *rinfo = RedisModule_Alloc(sizeof(struct RedisDL_RunInfo));
-  rinfo->client = bc;
-  rinfo->session = session;
-  rinfo->inputs = inputs;
-  rinfo->input_values = input_values;
-  rinfo->ninputs = ninputs;
-  rinfo->outputs = outputs;
-  rinfo->output_values = output_values;
-  rinfo->noutputs = noutputs;
-  rinfo->graphkey = key;
-  rinfo->outkeys = outkeys;
-  rinfo->status = TF_NewStatus();
+  rinfo->client = RedisModule_BlockClient(ctx, RedisDL_Run_Reply, NULL, NULL, 0);
 
   //  RedisModule_AbortBlock(bc);
   //  return RedisModule_ReplyWithError(ctx, "-ERR Can't start thread");

src/tensor.c

@@ -215,3 +215,8 @@ int Tensor_GetValueAsLongLong(RDL_Tensor* t, long long i, long long* val) {
   }
   return 1;
 }
+
+RDL_Tensor* Tensor_GetShallowCopy(RDL_Tensor* t){
+  ++t->refCount;
+  return t;
+}

src/tensor.h

@@ -29,6 +29,7 @@ int Tensor_SetValueFromLongLong(RDL_Tensor* tensor, long long i, long long val);
 int Tensor_SetValueFromDouble(RDL_Tensor* tensor, long long i, double val);
 int Tensor_GetValueAsDouble(RDL_Tensor* t, long long i, double* val);
 int Tensor_GetValueAsLongLong(RDL_Tensor* t, long long i, long long* val);
+RDL_Tensor* Tensor_GetShallowCopy(RDL_Tensor* t);