port PR 24562 to release/2023/3 (#24670)

### Details:
 - *port PR #24562 to release 2023.3*
 - *add support for InferAPI1 and api_version*

### Tickets:
-
*[issue-24509](https://github.com/openvinotoolkit/openvino/issues/24509)*

---------

Signed-off-by: HU Yuan2 <yuan2.hu@intel.com>
Co-authored-by: Wanglei Shen <wanglei.shen@intel.com>

src/inference/src/dev/threading/cpu_streams_executor.cpp

@@ -397,6 +397,13 @@ struct CPUStreamsExecutor::Impl {
             std::lock_guard<std::mutex> 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;
                 }
             }

tests/stress_tests/common/ie_pipelines/pipelines.cpp

@@ -100,15 +100,13 @@ std::function<void()> reinfer_request_inference(std::shared_ptr<InferApiBase> &i
     };
 }
 
-std::function<void()> recreate_and_infer_in_thread(std::shared_ptr<InferApiBase> &ie_wrapper) {
-    return [=] {
-        auto func = [=] {
-            ie_wrapper->create_infer_request();
-            ie_wrapper->prepare_input();
-            ie_wrapper->infer();
+std::function<void()> recreate_and_infer_in_thread(std::shared_ptr<InferApiBase> &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();
     };
 }
 

tests/stress_tests/common/ie_pipelines/pipelines.h

@@ -36,4 +36,4 @@ std::function<void()> recreate_infer_request(std::shared_ptr<InferApiBase> &ie_w
 
 std::function<void()> reinfer_request_inference(std::shared_ptr<InferApiBase> &ie_wrapper);
 
-std::function<void()> recreate_and_infer_in_thread(std::shared_ptr<InferApiBase> &ie_wrapper);
+std::function<void()> recreate_and_infer_in_thread(std::shared_ptr<InferApiBase> &ie_wrapper, const bool async = false);

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);
 }

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;

tests/stress_tests/memleaks_tests/tests.cpp

@@ -198,11 +198,13 @@ TEST_P(MemLeaksTestSuite, recreate_and_infer_in_thread) {
     std::vector<std::function<void()>> pipeline;
     size_t n_models = test_params.models.size();
 
+    std::vector<std::shared_ptr<InferApiBase>> 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 = [&] {

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,

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<std::thread> 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<std::shared_ptr<InferApiBase>> 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<std::thread> 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);
+}
+

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