From ec71f363f7f9404cf0a82343e5c9970da42d9d6a Mon Sep 17 00:00:00 2001 From: Changming Sun Date: Wed, 12 Dec 2018 15:52:36 -0800 Subject: [PATCH] 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 --- .../core/framework/parallel_executor.cc | 7 +++-- .../core/framework/parallel_executor.h | 10 +++++-- tools/ci_build/build.py | 26 +++++++++---------- 3 files changed, 25 insertions(+), 18 deletions(-) diff --git a/onnxruntime/core/framework/parallel_executor.cc b/onnxruntime/core/framework/parallel_executor.cc index 99eb06a159..367056ac90 100644 --- a/onnxruntime/core/framework/parallel_executor.cc +++ b/onnxruntime/core/framework/parallel_executor.cc @@ -47,7 +47,7 @@ Status ParallelExecutor::Execute(const SessionState& session_state, // Wait for finish. { std::unique_lock 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 lock(complete_mutex_); + out_standings_++; + } //std::cout << "Enqueue async node: " << p_node_index << ", out_standings: " << out_standings_ << std::endl; std::packaged_task task{std::bind(&ParallelExecutor::RunNodeAsync, this, p_node_index, std::cref(session_state), std::cref(logger))}; session_state.GetThreadPool()->RunTask(std::move(task)); diff --git a/onnxruntime/core/framework/parallel_executor.h b/onnxruntime/core/framework/parallel_executor.h index 6acc748fac..5d4aa551a0 100644 --- a/onnxruntime/core/framework/parallel_executor.h +++ b/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 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 root_frame_; std::vector node_refs_; std::mutex ref_mutex_; - std::atomic out_standings_; + int out_standings_; //protected by complete_mutex_ std::mutex complete_mutex_; std::condition_variable complete_cv_; diff --git a/tools/ci_build/build.py b/tools/ci_build/build.py index 2dce493f2d..da3245e7d5 100755 --- a/tools/ci_build/build.py +++ b/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)