[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 <yuan2.hu@intel.com>
Co-authored-by: Wanglei Shen <wanglei.shen@intel.com>

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

@@ -290,6 +290,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

@@ -99,12 +99,10 @@ 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();

tests/stress_tests/common/ie_pipelines/pipelines.h

@@ -33,4 +33,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,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

@@ -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;

tests/stress_tests/memleaks_tests/tests.cpp

@@ -193,11 +193,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();
+        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

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

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