diff --git a/src/inference/src/dev/threading/cpu_streams_executor.cpp b/src/inference/src/dev/threading/cpu_streams_executor.cpp index eb706a37192..c3cf96ecb84 100644 --- a/src/inference/src/dev/threading/cpu_streams_executor.cpp +++ b/src/inference/src/dev/threading/cpu_streams_executor.cpp @@ -397,6 +397,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 deb655f4ab5..6aa3b6a2d07 100644 --- a/tests/stress_tests/common/ie_pipelines/pipelines.cpp +++ b/tests/stress_tests/common/ie_pipelines/pipelines.cpp @@ -100,15 +100,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 8b5bf905e12..43277053c8c 100644 --- a/tests/stress_tests/common/ie_pipelines/pipelines.h +++ b/tests/stress_tests/common/ie_pipelines/pipelines.h @@ -36,4 +36,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 6af09e8bc7d..05cb57211e4 100644 --- a/tests/stress_tests/common/infer_api/infer_api.cpp +++ b/tests/stress_tests/common/infer_api/infer_api.cpp @@ -35,6 +35,22 @@ void InferAPI1::create_infer_request() { inferRequest = exeNetwork.CreateInferRequest(); } +void InferAPI1::create_and_infer(const bool &async) { + auto newInferRequest = exeNetwork.CreateInferRequest(); + auto batchSize = cnnNetwork.getBatchSize(); + batchSize = batchSize != 0 ? batchSize : 1; + fillBlobs(newInferRequest, exeNetwork.GetInputsInfo(), batchSize); + if (async) { + newInferRequest.StartAsync(); + newInferRequest.Wait(); + } else { + newInferRequest.Infer(); + } + for (auto &output: outputInfo) { + InferenceEngine::Blob::Ptr outputBlob = newInferRequest.GetBlob(output.first); + } +} + void InferAPI1::prepare_input() { auto batchSize = cnnNetwork.getBatchSize(); batchSize = batchSize != 0 ? batchSize : 1; @@ -127,6 +143,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 7694eec0d4c..4c94dde461e 100644 --- a/tests/stress_tests/common/infer_api/infer_api.h +++ b/tests/stress_tests/common/infer_api/infer_api.h @@ -21,6 +21,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; @@ -48,6 +50,8 @@ public: void create_infer_request() override; + void create_and_infer(const bool &aysnc) override; + void prepare_input() override; void infer() override; @@ -85,6 +89,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 71f06bb08fd..1c7de64b7f0 100644 --- a/tests/stress_tests/memleaks_tests/tests.cpp +++ b/tests/stress_tests/memleaks_tests/tests.cpp @@ -198,11 +198,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(test_params.api_version); + 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 bad1b991ba0..d632bfc1658 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 de4bbcce570..2a264baa435 100644 --- a/tests/stress_tests/unittests/tests_pipelines/tests_pipelines.cpp +++ b/tests/stress_tests/unittests/tests_pipelines/tests_pipelines.cpp @@ -92,3 +92,71 @@ void test_infer_request_inference(const std::string &model, const std::string &t infer_request_inference(model, target_device, api_version)(); } } + +static void test_recreate_and_infer_in_thread_one_model(const std::string &model, const std::string &target_device, + const int &n, const int &api_version, const bool &async) { + auto ie_wrapper = create_infer_api_wrapper(api_version); + ie_wrapper->read_network(model); + ie_wrapper->set_config(target_device, "THROUGHPUT_STREAMS", 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, const int &api_version) { + std::vector> ie_wrapper_vector; + for(int i = 0; i < 2; i++) { + auto ie_wrapper = create_infer_api_wrapper(api_version); + ie_wrapper->read_network(model); + ie_wrapper->set_config(target_device, "THROUGHPUT_STREAMS", 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, + const int &api_version) { + test_recreate_and_infer_in_thread_one_model(model, target_device, n, api_version, false); + test_recreate_and_infer_in_thread_one_model(model, target_device, n, api_version, true); + test_recreate_and_infer_in_thread_two_model(model, target_device, n, api_version); +} + diff --git a/tests/stress_tests/unittests/tests_pipelines/tests_pipelines.h b/tests/stress_tests/unittests/tests_pipelines/tests_pipelines.h index 36adf585986..bb5dc169c7c 100644 --- a/tests/stress_tests/unittests/tests_pipelines/tests_pipelines.h +++ b/tests/stress_tests/unittests/tests_pipelines/tests_pipelines.h @@ -30,6 +30,9 @@ void test_create_infer_request(const std::string &model, const std::string &targ void test_infer_request_inference(const std::string &model, const std::string &target_device, const int &n, const int &api_version); + +void test_recreate_and_infer_in_thread(const std::string &model, const std::string &target_device, const int &n, + const int &api_version); // tests_pipelines/tests_pipelines.cpp // tests_pipelines/tests_pipelines_full_pipeline.cpp