From 1a4ae5e7a35f5fc054e704e62ff9e7255a4da945 Mon Sep 17 00:00:00 2001 From: Yuan Hu Date: Thu, 23 May 2024 19:47:57 +0800 Subject: [PATCH] [core] fix thread safe issue in #19832 (#24562) ### Details: - *I lost code `t_stream_count_map[(void*)this] = item.first;` in #19832* - *the thread safe issue happen in below workflow* - create thread A call CustomThreadLocal:local() in thread A -> create stream A (the count of stream A is 2) destory thread A (the count of stream A is 1) create thread B (same thread id with thread A) call CustomThreadLocal:local() in thread B -> use stream A(the count of stream A is 1, so it's broken) - *add testcase, also fix https://github.com/openvinotoolkit/openvino/pull/19986/files#r1332774754* ### Tickets: - Closes https://github.com/openvinotoolkit/openvino/issues/24509 --------- Signed-off-by: HU Yuan2 Co-authored-by: Wanglei Shen --- .../dev/threading/cpu_streams_executor.cpp | 7 ++ .../common/ie_pipelines/pipelines.cpp | 12 ++-- .../common/ie_pipelines/pipelines.h | 2 +- .../common/infer_api/infer_api.cpp | 14 ++++ .../stress_tests/common/infer_api/infer_api.h | 4 ++ tests/stress_tests/memleaks_tests/tests.cpp | 4 +- tests/stress_tests/unittests/tests.cpp | 6 ++ .../tests_pipelines/tests_pipelines.cpp | 65 +++++++++++++++++++ .../tests_pipelines/tests_pipelines.h | 1 + 9 files changed, 106 insertions(+), 9 deletions(-) diff --git a/src/inference/src/dev/threading/cpu_streams_executor.cpp b/src/inference/src/dev/threading/cpu_streams_executor.cpp index 6c54025a0d5..5faaaf45878 100644 --- a/src/inference/src/dev/threading/cpu_streams_executor.cpp +++ b/src/inference/src/dev/threading/cpu_streams_executor.cpp @@ -290,6 +290,13 @@ struct CPUStreamsExecutor::Impl { std::lock_guard guard(_stream_map_mutex); for (auto& item : _stream_map) { if (item.first->get_id() == id) { + // check if the ThreadTracker of this stream is already in t_stream_count_map + // if not, then create ThreadTracker for it + auto iter = t_stream_count_map.find((void*)this); + if (iter == t_stream_count_map.end()) { + auto new_tracker_ptr = item.first->fetch(); + t_stream_count_map[(void*)this] = new_tracker_ptr; + } return item.second; } } diff --git a/tests/stress_tests/common/ie_pipelines/pipelines.cpp b/tests/stress_tests/common/ie_pipelines/pipelines.cpp index 29026347eec..965ce9b4f93 100644 --- a/tests/stress_tests/common/ie_pipelines/pipelines.cpp +++ b/tests/stress_tests/common/ie_pipelines/pipelines.cpp @@ -99,15 +99,13 @@ std::function reinfer_request_inference(std::shared_ptr &i }; } -std::function recreate_and_infer_in_thread(std::shared_ptr &ie_wrapper) { - return [=] { - auto func = [=] { - ie_wrapper->create_infer_request(); - ie_wrapper->prepare_input(); - ie_wrapper->infer(); +std::function recreate_and_infer_in_thread(std::shared_ptr &ie_wrapper, const bool async) { + return [async, &ie_wrapper] { + auto func = [&ie_wrapper, &async] { + ie_wrapper->create_and_infer(async); }; std::thread t(func); - t.join(); + t.join(); }; } diff --git a/tests/stress_tests/common/ie_pipelines/pipelines.h b/tests/stress_tests/common/ie_pipelines/pipelines.h index c9a6912c36f..23a9a693a6c 100644 --- a/tests/stress_tests/common/ie_pipelines/pipelines.h +++ b/tests/stress_tests/common/ie_pipelines/pipelines.h @@ -33,4 +33,4 @@ std::function recreate_infer_request(std::shared_ptr &ie_w std::function reinfer_request_inference(std::shared_ptr &ie_wrapper); -std::function recreate_and_infer_in_thread(std::shared_ptr &ie_wrapper); +std::function recreate_and_infer_in_thread(std::shared_ptr &ie_wrapper, const bool async = false); diff --git a/tests/stress_tests/common/infer_api/infer_api.cpp b/tests/stress_tests/common/infer_api/infer_api.cpp index a777dcbacbe..e1fa37c400a 100644 --- a/tests/stress_tests/common/infer_api/infer_api.cpp +++ b/tests/stress_tests/common/infer_api/infer_api.cpp @@ -35,6 +35,20 @@ void InferAPI2::create_infer_request() { infer_request = compiled_model.create_infer_request(); } +void InferAPI2::create_and_infer(const bool &async) { + auto new_infer_request = compiled_model.create_infer_request(); + fillTensors(new_infer_request, inputs); + if (async) { + new_infer_request.start_async(); + new_infer_request.wait(); + } else { + new_infer_request.infer(); + } + for (size_t i = 0; i < outputs.size(); ++i) { + const auto &output_tensor = new_infer_request.get_output_tensor(i); + } +} + void InferAPI2::prepare_input() { fillTensors(infer_request, inputs); } diff --git a/tests/stress_tests/common/infer_api/infer_api.h b/tests/stress_tests/common/infer_api/infer_api.h index b36fe241651..b18de15415f 100644 --- a/tests/stress_tests/common/infer_api/infer_api.h +++ b/tests/stress_tests/common/infer_api/infer_api.h @@ -20,6 +20,8 @@ public: virtual void create_infer_request() = 0; + virtual void create_and_infer(const bool &aysnc) = 0; + virtual void infer() = 0; virtual void prepare_input() = 0; @@ -47,6 +49,8 @@ public: void create_infer_request() override; + void create_and_infer(const bool &aysnc) override; + void prepare_input() override; void infer() override; diff --git a/tests/stress_tests/memleaks_tests/tests.cpp b/tests/stress_tests/memleaks_tests/tests.cpp index e984d1c0b8d..8337594c79e 100644 --- a/tests/stress_tests/memleaks_tests/tests.cpp +++ b/tests/stress_tests/memleaks_tests/tests.cpp @@ -193,11 +193,13 @@ TEST_P(MemLeaksTestSuite, recreate_and_infer_in_thread) { std::vector> pipeline; size_t n_models = test_params.models.size(); + std::vector> ie_wrapper_vector; for (int i = 0; i < n_models; i++) { auto ie_wrapper = create_infer_api_wrapper(); + ie_wrapper_vector.push_back(ie_wrapper); ie_wrapper->read_network(test_params.models[i]["full_path"]); ie_wrapper->load_network(test_params.device); - pipeline.push_back(recreate_and_infer_in_thread(ie_wrapper)); + pipeline.push_back(recreate_and_infer_in_thread(ie_wrapper_vector[i], false)); } auto test = [&] { diff --git a/tests/stress_tests/unittests/tests.cpp b/tests/stress_tests/unittests/tests.cpp index 3d793bee409..115e90da5a4 100644 --- a/tests/stress_tests/unittests/tests.cpp +++ b/tests/stress_tests/unittests/tests.cpp @@ -75,6 +75,12 @@ TEST_P(UnitTestSuite, create_infer_request_full_pipeline) { TEST_P(UnitTestSuite, infer_request_inference_full_pipeline) { runTest(test_infer_request_inference_full_pipeline, GetParam()); } + +TEST_P(UnitTestSuite, recreate_and_infer_in_thread) { + runTest(test_recreate_and_infer_in_thread, GetParam()); +} + + // tests_pipelines/tests_pipelines_full_pipeline.cpp INSTANTIATE_TEST_SUITE_P(StressUnitTests, UnitTestSuiteNoModel, diff --git a/tests/stress_tests/unittests/tests_pipelines/tests_pipelines.cpp b/tests/stress_tests/unittests/tests_pipelines/tests_pipelines.cpp index 8d0574086ba..324ff29b58e 100644 --- a/tests/stress_tests/unittests/tests_pipelines/tests_pipelines.cpp +++ b/tests/stress_tests/unittests/tests_pipelines/tests_pipelines.cpp @@ -81,3 +81,68 @@ void test_infer_request_inference(const std::string &model, const std::string &t infer_request_inference(model, target_device)(); } } + +static void test_recreate_and_infer_in_thread_one_model(const std::string &model, const std::string &target_device, const int &n, const bool &async) { + auto ie_wrapper = create_infer_api_wrapper(); + ie_wrapper->read_network(model); + ie_wrapper->set_config(target_device, ov::AnyMap{ov::inference_num_threads(2)}); + ie_wrapper->load_network(target_device); + auto fun = recreate_and_infer_in_thread(ie_wrapper, async); + for(int y = 0; y < n; y++) { + std::vector threads; + for (int i = 0; i < 4; i++) { + ie_wrapper->create_and_infer(async); + threads.emplace_back(fun); + } + for (int i = 0; i < 4; i++) { + threads[i].join(); + } + } +} + +static void test_recreate_and_infer_in_thread_two_model(const std::string &model, const std::string &target_device, const int &n) { + std::vector> ie_wrapper_vector; + for(int i = 0; i < 2; i++) { + auto ie_wrapper = create_infer_api_wrapper(); + ie_wrapper->read_network(model); + ie_wrapper->set_config(target_device, ov::AnyMap{ov::inference_num_threads(2)}); + ie_wrapper->load_network(target_device); + ie_wrapper_vector.push_back(ie_wrapper); + } + auto async_func0 = recreate_and_infer_in_thread(ie_wrapper_vector[0], true); + auto async_func1 = recreate_and_infer_in_thread(ie_wrapper_vector[1], true); + auto sync_func0 = recreate_and_infer_in_thread(ie_wrapper_vector[0], false); + auto sync_func1 = recreate_and_infer_in_thread(ie_wrapper_vector[1], false); + + for(int y = 0; y < n; y++) { + std::vector threads; + for (int i = 0; i < 4; i++) { + if ( i % 2 == 0) { + ie_wrapper_vector[0]->create_and_infer(true); + threads.emplace_back(async_func0); + } else { + ie_wrapper_vector[0]->create_and_infer(false); + threads.emplace_back(sync_func0); + } + } + for (int i = 0; i < 4; i++) { + if ( i % 2 == 0) { + ie_wrapper_vector[1]->create_and_infer(true); + threads.emplace_back(async_func1); + } else { + ie_wrapper_vector[1]->create_and_infer(false); + threads.emplace_back(sync_func1); + } + } + for (int i = 0; i < threads.size(); i++) { + threads[i].join(); + } + } +} + +void test_recreate_and_infer_in_thread(const std::string &model, const std::string &target_device, const int &n) { + test_recreate_and_infer_in_thread_one_model(model, target_device, n, false); + test_recreate_and_infer_in_thread_one_model(model, target_device, n, true); + test_recreate_and_infer_in_thread_two_model(model, target_device, n); +} + diff --git a/tests/stress_tests/unittests/tests_pipelines/tests_pipelines.h b/tests/stress_tests/unittests/tests_pipelines/tests_pipelines.h index e9bdc099700..547a22c5c9d 100644 --- a/tests/stress_tests/unittests/tests_pipelines/tests_pipelines.h +++ b/tests/stress_tests/unittests/tests_pipelines/tests_pipelines.h @@ -23,6 +23,7 @@ void test_create_compiled_model(const std::string &model, const std::string &tar void test_create_infer_request(const std::string &model, const std::string &target_device, const int &n); void test_infer_request_inference(const std::string &model, const std::string &target_device, const int &n); +void test_recreate_and_infer_in_thread(const std::string &model, const std::string &target_device, const int &n); // tests_pipelines/tests_pipelines.cpp // tests_pipelines/tests_pipelines_full_pipeline.cpp