Fix a bug in parallel executor and enable test for it in CI (#157)

* Fix a bug in parallel executor and enable test for it in CI

onnxruntime/core/framework/parallel_executor.cc

@@ -47,7 +47,7 @@ Status ParallelExecutor::Execute(const SessionState& session_state,
   // Wait for finish.
   {
     std::unique_lock<std::mutex> lock(complete_mutex_);
-    while (out_standings_.load() > 0) complete_cv_.wait(lock);
+    while (out_standings_ > 0) complete_cv_.wait(lock);
   }
 
   VLOGS(logger, 1) << "Fetching output.";
@@ -223,7 +223,10 @@ void ParallelExecutor::RunNodeAsyncInternal(size_t p_node_index,
 }
 
 void ParallelExecutor::EnqueueNode(size_t p_node_index, const SessionState& session_state, const logging::Logger& logger) {
-  out_standings_++;
+  {
+    std::unique_lock<std::mutex> lock(complete_mutex_);
+    out_standings_++;
+  }
   //std::cout << "Enqueue async node: " << p_node_index << ", out_standings: " << out_standings_ << std::endl;
   std::packaged_task<void()> task{std::bind(&ParallelExecutor::RunNodeAsync, this, p_node_index, std::cref(session_state), std::cref(logger))};
   session_state.GetThreadPool()->RunTask(std::move(task));

onnxruntime/core/framework/parallel_executor.h

@@ -44,7 +44,13 @@ class ParallelExecutor : public IExecutor {
                      const logging::Logger& logger);
 
   void FinishNodeRun() {
-    if (--out_standings_ == 0) {
+    bool finished = false;
+    {
+      //Because we have a mutex here, it's not possible another thread is doing the test("while (out_standings_ > 0)"
+      std::lock_guard<std::mutex> lock(complete_mutex_);
+      finished = --out_standings_ == 0;
+    }
+    if (finished) {
       //std::cout << "all out standing nodes are completed." << std::endl;
       complete_cv_.notify_all();
     }
@@ -53,7 +59,7 @@ class ParallelExecutor : public IExecutor {
   std::unique_ptr<ExecutionFrame> root_frame_;
   std::vector<size_t> node_refs_;
   std::mutex ref_mutex_;
-  std::atomic<int> out_standings_;
+  int out_standings_;  //protected by complete_mutex_
   std::mutex complete_mutex_;
   std::condition_variable complete_cv_;
 

tools/ci_build/build.py

@@ -205,16 +205,18 @@ def download_test_data(build_dir, src_url, expected_md5):
             log.info('deleting %s' % models_dir)
             shutil.rmtree(models_dir)
         run_subprocess(['unzip','-qd', models_dir, local_zip_file])
+        return True
+    return False
 
 
 def generate_build_tree(cmake_path, source_dir, build_dir, cuda_home, cudnn_home, pb_home, configs, cmake_extra_defines, args, cmake_extra_args):
-    download_test_data(build_dir,'https://onnxruntimetestdata.blob.core.windows.net/models/20181210.zip','a966def7447f4ff04f5665bca235b3f3')
+    has_test_data = download_test_data(build_dir,'https://onnxruntimetestdata.blob.core.windows.net/models/20181210.zip','a966def7447f4ff04f5665bca235b3f3')
     log.info("Generating CMake build tree")
     cmake_dir = os.path.join(source_dir, "cmake")
     # TODO: fix jemalloc build so it does not conflict with onnxruntime shared lib builds. (e.g. onnxuntime_pybind)
     # for now, disable jemalloc if pybind is also enabled.
     cmake_args = [cmake_path, cmake_dir,
-                 "-Donnxruntime_RUN_ONNX_TESTS=" + ("ON" if args.enable_onnx_tests else "OFF"),
+                 "-Donnxruntime_RUN_ONNX_TESTS=" + ("ON" if has_test_data else "OFF"),
                  "-Donnxruntime_GENERATE_TEST_REPORTS=ON",
                  "-Donnxruntime_DEV_MODE=ON",
                  "-DPYTHON_EXECUTABLE=" + sys.executable,
@@ -421,13 +423,7 @@ def run_onnxruntime_tests(args, source_dir, ctest_path, build_dir, configs, enab
             if onnxml_test:
                 run_subprocess([sys.executable, 'onnxruntime_test_python_keras.py'], cwd=cwd, dll_path=dll_path)
 
-        # shared lib tests - both simple + custom op
-        if args.build_shared_lib:
-            if is_ubuntu_1604():
-                run_subprocess([cwd+'/onnxruntime_shared_lib_test'], cwd=cwd, dll_path=dll_path)
-
-def run_onnx_tests(build_dir, configs, onnx_test_data_dir, provider):
-    #TODO: enable multiple threaded executor test
+def run_onnx_tests(build_dir, configs, onnx_test_data_dir, provider, enable_parallel_executor_test):
     for config in configs:
         cwd = get_config_build_dir(build_dir, config)
         if is_windows():
@@ -436,14 +432,16 @@ def run_onnx_tests(build_dir, configs, onnx_test_data_dir, provider):
         else:
            exe = os.path.join(cwd, 'onnx_test_runner')
            model_dir = os.path.join(build_dir, "models")
-        cmd = [exe]
+        cmd = []
         if provider:
           cmd += ["-e", provider]
         if config != 'Debug' and os.path.exists(model_dir):
           cmd.append(model_dir)
         if os.path.exists(onnx_test_data_dir):
           cmd.append(onnx_test_data_dir)
-        run_subprocess(cmd, cwd=cwd)
+        run_subprocess([exe] + cmd, cwd=cwd)
+        if enable_parallel_executor_test:
+          run_subprocess([exe,'-x'] + cmd, cwd=cwd)
 
 def build_python_wheel(source_dir, build_dir, configs, use_cuda):
     for config in configs:
@@ -531,11 +529,11 @@ def main():
     # run the onnx model tests if requested explicitly.
     if (args.enable_onnx_tests):
         if args.use_cuda:
-          run_onnx_tests(build_dir, configs, onnx_test_data_dir, 'cuda')
+          run_onnx_tests(build_dir, configs, onnx_test_data_dir, 'cuda', False)
         else:
-          run_onnx_tests(build_dir, configs, onnx_test_data_dir, None)
+          run_onnx_tests(build_dir, configs, onnx_test_data_dir, None, True)
           if args.use_mkldnn:
-            run_onnx_tests(build_dir, configs, onnx_test_data_dir, 'mkldnn')
+            run_onnx_tests(build_dir, configs, onnx_test_data_dir, 'mkldnn', True)
 
     if args.build_wheel:
         build_python_wheel(source_dir, build_dir, configs, args.use_cuda)