diff --git a/src/inference/dev_api/openvino/runtime/threading/cpu_message.hpp b/src/inference/dev_api/openvino/runtime/threading/cpu_message.hpp
new file mode 100644
index 00000000000..abfff73efd7
--- /dev/null
+++ b/src/inference/dev_api/openvino/runtime/threading/cpu_message.hpp
@@ -0,0 +1,87 @@
+// Copyright (C) 2018-2024 Intel Corporation
+// SPDX-License-Identifier: Apache-2.0
+//
+
+/**
+ * @brief OpenVINO Runtime Executor Manager
+ * @file openvino/runtime/threading/executor_manager.hpp
+ */
+
+#pragma once
+
+#include <atomic>
+#include <condition_variable>
+#include <memory>
+#include <mutex>
+#include <queue>
+#include <set>
+#include <thread>
+#include <vector>
+
+#include "openvino/runtime/common.hpp"
+#include "openvino/runtime/threading/istreams_executor.hpp"
+#include "openvino/runtime/threading/itask_executor.hpp"
+#include "openvino/runtime/compiled_model.hpp"
+#include "openvino/runtime/iasync_infer_request.hpp"
+
+namespace ov {
+
+namespace threading {
+enum MsgType { START_INFER, TENSOR_PARALLEL, CALL_BACK, REDUCE, QUIT };
+
+struct MessageInfo {
+    MsgType msg_type;
+    std::vector<int> rank;
+    void* buf;
+    Task task;
+};
+class OPENVINO_RUNTIME_API MessageManager {
+public:
+    MessageManager();
+
+    void send_message(const MessageInfo& msg_info);
+
+    std::vector<MessageInfo> wait_message(int stream_id);
+
+    void infer_wait();
+
+    void reduce_wait(int stream_id);
+
+    void server_wait();
+
+    void stop_server_thread();
+
+    void clear();
+
+    ~MessageManager();
+
+    void set_sub_compiled_models(std::vector<std::shared_ptr<ov::ICompiledModel>> models);
+    std::vector<std::shared_ptr<ov::ICompiledModel>> get_sub_compiled_models();
+
+    void set_sub_infer_requests(std::vector<std::shared_ptr<ov::IAsyncInferRequest>> requests);
+    std::vector<std::shared_ptr<ov::IAsyncInferRequest>> get_sub_infer_requests();
+
+    int get_num_sub_streams();
+
+private:
+    int _num_sub_streams;
+    std::vector<std::shared_ptr<ov::ICompiledModel>> _sub_compiled_models;
+    std::vector<std::shared_ptr<ov::IAsyncInferRequest>> _sub_infer_requests;
+    std::thread _serverThread;
+    bool _isServerStopped = false;
+    std::vector<MessageInfo> _messageQueue;
+    std::vector<std::vector<MessageInfo>> _readQueue;
+    std::vector<int> _reduceQueue;
+    std::mutex _msgMutex;
+    std::mutex _readMutex;
+    std::mutex _inferMutex;
+    std::mutex _reduceMutex;
+    std::condition_variable _msgCondVar;
+    std::condition_variable _readCondVar;
+    std::condition_variable _inferCondVar;
+    std::condition_variable _reduceCondVar;
+};
+
+OPENVINO_RUNTIME_API std::shared_ptr<MessageManager> message_manager();
+}  // namespace threading
+}  // namespace ov
\ No newline at end of file
diff --git a/src/inference/dev_api/openvino/runtime/threading/cpu_streams_executor.hpp b/src/inference/dev_api/openvino/runtime/threading/cpu_streams_executor.hpp
index 1638ada4282..1f694fe9b70 100644
--- a/src/inference/dev_api/openvino/runtime/threading/cpu_streams_executor.hpp
+++ b/src/inference/dev_api/openvino/runtime/threading/cpu_streams_executor.hpp
@@ -53,6 +53,8 @@ public:
 
     int get_socket_id() override;
 
+    std::vector<int> get_rank() override;
+
     void run_sub_stream(Task task, int id) override;
 
 private:
diff --git a/src/inference/dev_api/openvino/runtime/threading/istreams_executor.hpp b/src/inference/dev_api/openvino/runtime/threading/istreams_executor.hpp
index d98419c568d..180a2b77a48 100644
--- a/src/inference/dev_api/openvino/runtime/threading/istreams_executor.hpp
+++ b/src/inference/dev_api/openvino/runtime/threading/istreams_executor.hpp
@@ -95,6 +95,7 @@ public:
         std::vector<std::vector<int>> _streams_info_table = {};
         std::vector<std::vector<int>> _stream_processor_ids;
         int _sub_streams = 0;
+        std::vector<int> _rank = {};
 
         /**
          * @brief Get and reserve cpu ids based on configuration and hardware information,
@@ -124,6 +125,7 @@ public:
          * @param[in]  cpu_reservation              @copybrief Config::_cpu_reservation
          * @param[in]  cpu_pinning                  @copybrief Config::_cpu_pinning
          * @param[in]  streams_info_table           @copybrief Config::_streams_info_table
+         * @param[in]  rank                         @copybrief Config::_rank
          */
         Config(std::string name = "StreamsExecutor",
                int streams = 1,
@@ -131,14 +133,16 @@ public:
                ov::hint::SchedulingCoreType thread_preferred_core_type = ov::hint::SchedulingCoreType::ANY_CORE,
                bool cpu_reservation = false,
                bool cpu_pinning = false,
-               std::vector<std::vector<int>> streams_info_table = {})
+               std::vector<std::vector<int>> streams_info_table = {},
+               std::vector<int> rank = {})
             : _name{name},
               _streams{streams},
               _threads_per_stream{threads_per_stream},
               _thread_preferred_core_type(thread_preferred_core_type),
               _cpu_reservation{cpu_reservation},
               _cpu_pinning{cpu_pinning},
-              _streams_info_table{streams_info_table} {
+              _streams_info_table{streams_info_table},
+              _rank{rank} {
             update_executor_config();
         }
 
@@ -197,6 +201,9 @@ public:
         int get_sub_streams() const {
             return _sub_streams;
         }
+        std::vector<int> get_rank() const {
+            return _rank;
+        }
         StreamsMode get_sub_stream_mode() const {
             const auto proc_type_table = get_proc_type_table();
             int sockets = proc_type_table.size() > 1 ? static_cast<int>(proc_type_table.size()) - 1 : 1;
@@ -249,6 +256,13 @@ public:
      */
     virtual int get_socket_id() = 0;
 
+    /**
+     * @brief Return the rank of current stream
+     *        Return {} when current stream has no rank
+     * @return Rank array, or throws exceptions if called not from stream thread
+     */
+    virtual std::vector<int> get_rank() = 0;
+
     /**
      * @brief Execute the task in the current thread using streams executor configuration and constraints
      * @param task A task to start
diff --git a/src/inference/src/dev/threading/cpu_message.cpp b/src/inference/src/dev/threading/cpu_message.cpp
new file mode 100644
index 00000000000..02722b2c342
--- /dev/null
+++ b/src/inference/src/dev/threading/cpu_message.cpp
@@ -0,0 +1,198 @@
+// Copyright (C) 2018-2024 Intel Corporation
+// SPDX-License-Identifier: Apache-2.0
+//
+#include "openvino/runtime/threading/cpu_message.hpp"
+
+#include <memory>
+#include <mutex>
+#include <queue>
+#include <set>
+#include <thread>
+#include <vector>
+
+namespace ov {
+namespace threading {
+
+MessageManager::MessageManager() {}
+
+MessageManager::~MessageManager() {}
+
+void MessageManager::send_message(const MessageInfo& msg_info) {
+    {
+        std::lock_guard<std::mutex> lock(_msgMutex);
+        _messageQueue.push_back(msg_info);
+        // std::cout << "send : " << msg_info.msg_type << ", " << (msg_info.rank.size() > 0 ? msg_info.rank[0] : -1)
+        //           << "\n";
+    }
+    _msgCondVar.notify_all();
+}
+
+std::vector<MessageInfo> MessageManager::wait_message(int stream_id) {
+    std::vector<MessageInfo> messages_total;
+    std::unique_lock<std::mutex> lock(_readMutex);
+    _readCondVar.wait(lock, [&] {
+        // std::cout << "wait_" << stream_id << " : " << _readQueue[stream_id].size() << " / " << _num_sub_streams <<
+        // "\n";
+        return static_cast<int>(_readQueue[stream_id].size()) >= _num_sub_streams;
+    });
+    std::swap(_readQueue[stream_id], messages_total);
+    // std::cout << "wait_" << stream_id << " " << _readQueue[stream_id].size() << " end\n";
+    return messages_total;
+}
+
+void MessageManager::infer_wait() {
+    std::unique_lock<std::mutex> lock(_inferMutex);
+    _inferCondVar.wait(lock);
+}
+
+void MessageManager::reduce_wait(int stream_id) {
+    std::unique_lock<std::mutex> lock(_reduceMutex);
+    while (_reduceQueue[stream_id] < _num_sub_streams) {
+        // std::cout << "reduce_wait_" << stream_id << " " << _reduceQueue[stream_id] << " end\n";
+        _reduceCondVar.wait(lock);
+    }
+    _reduceQueue[stream_id] = 0;
+}
+
+void MessageManager::server_wait() {
+    if (!_serverThread.joinable()) {
+        assert(_num_sub_streams);
+        _readQueue.assign(_num_sub_streams, std::vector<MessageInfo>());
+        _reduceQueue.assign(_num_sub_streams, 0);
+        MsgType msg_type;
+        _serverThread = std::thread([&]() {
+            int count = 0;
+            int reduce_count = 0;
+            while (!_isServerStopped) {
+                std::vector<MessageInfo> msgQueue;
+                {
+                    // std::cout << "server_wait ........" << _isServerStopped << "\n";
+                    std::unique_lock<std::mutex> lock(_msgMutex);
+                    _msgCondVar.wait(lock, [&] {
+                        return !_messageQueue.empty();
+                    });
+                    std::swap(_messageQueue, msgQueue);
+                    // std::cout << "server_wait receive: " << msgQueue[0].msg_type << " rank:" <<
+                    // msgQueue[0].rank.size()
+                    //           << " / " << msgQueue.size() << "\n";
+                }
+
+                for (auto rec_info : msgQueue) {
+                    msg_type = rec_info.msg_type;
+                    if (msg_type == START_INFER) {
+                        Task task = std::move(rec_info.task);
+                        task();
+                    } else if (msg_type == TENSOR_PARALLEL) {
+                        // Resend _readQueue that failed last time
+                        bool stop = false;
+                        while (!stop) {
+                            stop = true;
+                            for (int i = 0; i < _num_sub_streams; i++) {
+                                if (static_cast<int>(_readQueue[i].size()) == _num_sub_streams) {
+                                    stop = false;
+                                }
+                            }
+                            if (!stop) {
+                                _readCondVar.notify_all();
+                            }
+                        }
+                        for (int i = 0; i < _num_sub_streams; i++) {
+                            std::lock_guard<std::mutex> lock(_readMutex);
+                            _readQueue[i].push_back(rec_info);
+                        }
+                        _readCondVar.notify_all();
+                    } else if (msg_type == CALL_BACK) {  // CALL_BACK
+                        count++;
+                        // std::cout << "server_wait CALL_BACK: " << count << "/" << _num_sub_streams << "\n";
+                        if (count == _num_sub_streams) {
+                            _inferCondVar.notify_one();
+                            count = 0;
+                        }
+                    } else if (msg_type == REDUCE) {  // REDUCE
+                        reduce_count++;
+                        // std::cout << "server_wait REDUCE: " << reduce_count << "/" << _num_sub_streams << "\n";
+                        if (reduce_count == _num_sub_streams) {
+                            {
+                                std::lock_guard<std::mutex> lock(_reduceMutex);
+                                _reduceQueue.assign(_num_sub_streams, reduce_count);
+                            }
+                            _reduceCondVar.notify_all();
+                            reduce_count = 0;
+                        }
+                    } else if (msg_type == QUIT) {
+                        _isServerStopped = true;
+                    }
+                }
+            }
+            // std::cout << "-------- server_wait end ---------\n";
+        });
+    }
+}
+
+void MessageManager::set_sub_compiled_models(std::vector<std::shared_ptr<ov::ICompiledModel>> models) {
+    _sub_compiled_models = models;
+    _num_sub_streams = static_cast<int>(_sub_compiled_models.size());
+    assert(_num_sub_streams);
+}
+
+std::vector<std::shared_ptr<ov::ICompiledModel>> MessageManager::get_sub_compiled_models() {
+    return _sub_compiled_models;
+}
+
+void MessageManager::set_sub_infer_requests(std::vector<std::shared_ptr<IAsyncInferRequest>> requests) {
+    _sub_infer_requests = requests;
+}
+
+std::vector<std::shared_ptr<ov::IAsyncInferRequest>> MessageManager::get_sub_infer_requests() {
+    return _sub_infer_requests;
+}
+
+int MessageManager::get_num_sub_streams() {
+    return _num_sub_streams;
+}
+
+void MessageManager::stop_server_thread() {
+    MessageInfo msg_info;
+    msg_info.msg_type = ov::threading::MsgType::QUIT;
+    send_message(msg_info);
+    if (_serverThread.joinable()) {
+        _serverThread.join();
+    }
+}
+
+void MessageManager::clear() {
+    _sub_infer_requests.clear();
+    _sub_compiled_models.clear();
+}
+
+namespace {
+
+class MessageManageHolder {
+    std::mutex _mutex;
+    std::weak_ptr<MessageManager> _manager;
+
+public:
+    MessageManageHolder(const MessageManageHolder&) = delete;
+    MessageManageHolder& operator=(const MessageManageHolder&) = delete;
+
+    MessageManageHolder() = default;
+
+    std::shared_ptr<ov::threading::MessageManager> get() {
+        std::lock_guard<std::mutex> lock(_mutex);
+        auto manager = _manager.lock();
+        if (!manager) {
+            _manager = manager = std::make_shared<MessageManager>();
+        }
+        return manager;
+    }
+};
+
+}  // namespace
+
+std::shared_ptr<MessageManager> message_manager() {
+    static MessageManageHolder message_manage;
+    return message_manage.get();
+}
+
+}  // namespace threading
+}  // namespace ov
\ No newline at end of file
diff --git a/src/inference/src/dev/threading/cpu_streams_executor.cpp b/src/inference/src/dev/threading/cpu_streams_executor.cpp
index 5faaaf45878..ebfe445e1d0 100644
--- a/src/inference/src/dev/threading/cpu_streams_executor.cpp
+++ b/src/inference/src/dev/threading/cpu_streams_executor.cpp
@@ -189,6 +189,7 @@ struct CPUStreamsExecutor::Impl {
             int streams_num = _impl->_config.get_streams();
             const auto stream_id =
                 streams_num == 0 ? 0 : (_sub_stream_id >= 0 ? streams_num + _sub_stream_id : _streamId % streams_num);
+            _rank = _impl->_config.get_rank();
             get_cur_stream_info(stream_id,
                                 _impl->_config.get_cpu_pinning(),
                                 org_proc_type_table,
@@ -216,6 +217,7 @@ struct CPUStreamsExecutor::Impl {
         int _socketId = 0;
         bool _execute = false;
         int _sub_stream_id = -1;
+        std::vector<int> _rank;
         std::queue<Task> _taskQueue;
 #if OV_THREAD == OV_THREAD_TBB || OV_THREAD == OV_THREAD_TBB_AUTO
         std::unique_ptr<custom::task_arena> _taskArena;
@@ -344,7 +346,7 @@ struct CPUStreamsExecutor::Impl {
         _exectorMgr = executor_manager();
         auto numaNodes = get_available_numa_nodes();
         int streams_num = _config.get_streams();
-        int sub_streams_num = _config.get_sub_streams();
+        int sub_streams_num = 0;
         if (streams_num != 0) {
             std::copy_n(std::begin(numaNodes),
                         std::min<std::size_t>(streams_num, numaNodes.size()),
@@ -363,6 +365,7 @@ struct CPUStreamsExecutor::Impl {
                     {
                         std::unique_lock<std::mutex> lock(_mutex);
                         _queueCondVar.wait(lock, [&] {
+                            // std::cout << _config.get_name() << " addr: " << this << " in : " << streamId << "\n";
                             return !_taskQueue.empty() || (stopped = _isStopped);
                         });
                         if (!_taskQueue.empty()) {
@@ -491,6 +494,7 @@ struct CPUStreamsExecutor::Impl {
     std::vector<int> _usedNumaNodes;
     CustomThreadLocal _streams;
     std::shared_ptr<ExecutorManager> _exectorMgr;
+    bool _isExit = false;
 };
 
 int CPUStreamsExecutor::get_stream_id() {
@@ -508,6 +512,11 @@ int CPUStreamsExecutor::get_socket_id() {
     return stream->_socketId;
 }
 
+std::vector<int> CPUStreamsExecutor::get_rank() {
+    auto stream = _impl->_streams.local();
+    return stream->_rank;
+}
+
 CPUStreamsExecutor::CPUStreamsExecutor(const IStreamsExecutor::Config& config) : _impl{new Impl{config}} {}
 
 CPUStreamsExecutor::~CPUStreamsExecutor() {
diff --git a/src/inference/src/dev/threading/cpu_streams_executor_internal.cpp b/src/inference/src/dev/threading/cpu_streams_executor_internal.cpp
index 02c9b940cff..54d02cfe87f 100644
--- a/src/inference/src/dev/threading/cpu_streams_executor_internal.cpp
+++ b/src/inference/src/dev/threading/cpu_streams_executor_internal.cpp
@@ -92,8 +92,15 @@ void reserve_cpu_by_streams_info(const std::vector<std::vector<int>> _streams_in
     int num_conditions = 0;
     int condition_idx = 0;
     bool last_all_proc = false;
+    bool sub_stream_enable = false;
 
     for (size_t i = 0; i < _streams_info_table.size(); i++) {
+        if (i > 0 && _streams_info_table[i][NUMBER_OF_STREAMS] < 0 &&
+            _streams_info_table[i - 1][NUMBER_OF_STREAMS] > 0) {
+            stream_pos.clear();
+            num_streams = 0;
+            sub_stream_enable = true;
+        }
         if (_streams_info_table[i][NUMBER_OF_STREAMS] != 0) {
             stream_pos.push_back(num_streams);
         }
@@ -109,11 +116,15 @@ void reserve_cpu_by_streams_info(const std::vector<std::vector<int>> _streams_in
         std::string proc_type = "";
         std::string numa_node = "";
         std::string socket = "";
-        if (_streams_info_table[i][NUMBER_OF_STREAMS] != 0) {
+        if ((_streams_info_table[i][NUMBER_OF_STREAMS] > 0 && !sub_stream_enable) ||
+            (_streams_info_table[i][NUMBER_OF_STREAMS] < 0 && sub_stream_enable)) {
             streams_table.push_back(_streams_info_table[i]);
             if (_streams_info_table[i][NUMBER_OF_STREAMS] < 0) {
-                streams_table[streams_table.size() - 1][NUMBER_OF_STREAMS] = 1;
+                streams_table[streams_table.size() - 1][NUMBER_OF_STREAMS] =
+                    std::abs(_streams_info_table[i][NUMBER_OF_STREAMS]);
             }
+        } else {
+            continue;
         }
         if (last_all_proc && _streams_info_table[i][NUMBER_OF_STREAMS] != 0) {
             last_all_proc = false;
@@ -135,7 +146,7 @@ void reserve_cpu_by_streams_info(const std::vector<std::vector<int>> _streams_in
                 std::make_pair(proc_type + numa_node + socket, _streams_info_table[i][THREADS_PER_STREAM]));
             threads_status[condition_idx].push_back(0);
         }
-        if (_streams_info_table[i][PROC_TYPE] > ALL_PROC && _streams_info_table[i][NUMBER_OF_STREAMS] > 0) {
+        if (_streams_info_table[i][PROC_TYPE] > ALL_PROC && _streams_info_table[i][NUMBER_OF_STREAMS] != 0) {
             condition_idx++;
         }
     }
diff --git a/src/plugins/intel_cpu/src/async_infer_request.cpp b/src/plugins/intel_cpu/src/async_infer_request.cpp
index b77d0059db7..383aa7dc460 100644
--- a/src/plugins/intel_cpu/src/async_infer_request.cpp
+++ b/src/plugins/intel_cpu/src/async_infer_request.cpp
@@ -4,6 +4,8 @@
 
 #include "async_infer_request.h"
 
+#include "openvino/runtime/threading/cpu_message.hpp"
+
 ov::intel_cpu::AsyncInferRequest::AsyncInferRequest(
     const std::shared_ptr<IInferRequest>& request,
     const std::shared_ptr<ov::threading::ITaskExecutor>& task_executor,
@@ -13,6 +15,11 @@ ov::intel_cpu::AsyncInferRequest::AsyncInferRequest(
 }
 
 ov::intel_cpu::AsyncInferRequest::~AsyncInferRequest() {
+    if (m_has_sub_infers) {
+        auto message = ov::threading::message_manager();
+        message->stop_server_thread();
+        message->clear();
+    }
     stop_and_wait();
 }
 
diff --git a/src/plugins/intel_cpu/src/async_infer_request.h b/src/plugins/intel_cpu/src/async_infer_request.h
index ebf1a210f30..c7b6b5d0898 100644
--- a/src/plugins/intel_cpu/src/async_infer_request.h
+++ b/src/plugins/intel_cpu/src/async_infer_request.h
@@ -17,7 +17,12 @@ public:
                       const std::shared_ptr<ov::threading::ITaskExecutor>& callback_executor);
     ~AsyncInferRequest();
 
+    void setSubInfer(bool has_sub_infer) {
+        m_has_sub_infers = has_sub_infer;
+    }
+
     void throw_if_canceled() const;
+    bool m_has_sub_infers = false;
 };
 
 }  // namespace intel_cpu
diff --git a/src/plugins/intel_cpu/src/compiled_model.cpp b/src/plugins/intel_cpu/src/compiled_model.cpp
index a695bae170b..00c9c7975fd 100644
--- a/src/plugins/intel_cpu/src/compiled_model.cpp
+++ b/src/plugins/intel_cpu/src/compiled_model.cpp
@@ -17,6 +17,8 @@
 #include "openvino/util/common_util.hpp"
 #include "openvino/runtime/threading/cpu_streams_executor.hpp"
 #include "transformations/utils/utils.hpp"
+#include "openvino/runtime/threading/cpu_streams_info.hpp"
+#include "openvino/runtime/threading/cpu_message.hpp"
 
 #include "cpu/x64/cpu_isa_traits.hpp"
 #include <cstring>
@@ -54,13 +56,19 @@ CompiledModel::CompiledModel(const std::shared_ptr<ov::Model>& model,
     if (!core)
         OPENVINO_THROW("Unable to get API version. Core is unavailable");
 
-    ov::threading::IStreamsExecutor::Ptr stream_executor = nullptr;
+    IStreamsExecutor::Config executor_confg;
     if (cfg.exclusiveAsyncRequests) {
         // special case when all InferRequests are muxed into a single queue
         m_task_executor = m_plugin->get_executor_manager()->get_executor("CPU");
     } else {
-        stream_executor = m_plugin->get_executor_manager()->get_idle_cpu_streams_executor(m_cfg.streamExecutorConfig);
-        m_task_executor = stream_executor;
+        executor_confg = m_cfg.enableSubStreams ? IStreamsExecutor::Config{"CPUMainStreamExecutor",
+                                                                           1,
+                                                                           1,
+                                                                           ov::hint::SchedulingCoreType::ANY_CORE,
+                                                                           false,
+                                                                           true}
+                                                : m_cfg.streamExecutorConfig;
+        m_task_executor = m_plugin->get_executor_manager()->get_idle_cpu_streams_executor(executor_confg);
     }
     if (0 != m_cfg.streamExecutorConfig.get_streams()) {
         m_callback_executor = m_plugin->get_executor_manager()->get_idle_cpu_streams_executor(
@@ -74,11 +82,11 @@ CompiledModel::CompiledModel(const std::shared_ptr<ov::Model>& model,
     if (m_callback_executor)
         set_callback_executor(m_callback_executor);
 
-    int streams = std::max(1, m_cfg.streamExecutorConfig.get_streams());
+    int streams = std::max(1, executor_confg.get_streams());
     std::vector<Task> tasks;
     tasks.resize(streams);
     m_graphs.resize(streams);
-    if (m_cfg.streamExecutorConfig.get_streams() != 0) {
+    if (executor_confg.get_streams() != 0) {
         auto all_graphs_ready = [&] {
             return std::all_of(m_graphs.begin(), m_graphs.end(), [&](Graph& graph) {
                 return graph.IsReady();
@@ -102,16 +110,40 @@ CompiledModel::CompiledModel(const std::shared_ptr<ov::Model>& model,
     } else {
         CompiledModel::get_graph();
     }
-    // init sub stream threads of executor
-    int sub_streams = m_cfg.streamExecutorConfig.get_sub_streams();
-    if (sub_streams > 0 && stream_executor != nullptr) {
-        std::vector<Task> tasks;
-        tasks.resize(sub_streams);
-        for (auto&& task : tasks) {
-            task = [] {};
+    // std::cout << "m_has_sub_compiled_models: " << m_cfg.enableSubStreams << ", " << m_cfg.streamExecutorConfig.get_sub_streams() << "\n";
+    if (m_cfg.enableSubStreams) {
+        m_cfg.enableSubStreams = false;
+        m_has_sub_compiled_models = true;
+        auto sub_cfg = m_cfg;
+        std::vector<std::shared_ptr<ov::ICompiledModel>> sub_models;
+        auto streams_info_table = m_cfg.streamExecutorConfig.get_streams_info_table();
+        auto message = message_manager();
+        for (int i = 0; i < m_cfg.streamExecutorConfig.get_sub_streams(); i++) {
+            std::vector<std::vector<int>> sub_streams_table;
+            sub_streams_table.push_back(streams_info_table[i + 1]);
+            sub_streams_table[0][NUMBER_OF_STREAMS] = 1;
+            sub_cfg.streamExecutorConfig = IStreamsExecutor::Config{"CPUStreamsExecutor",
+                                                                    1,
+                                                                    1,
+                                                                    ov::hint::SchedulingCoreType::ANY_CORE,
+                                                                    false,
+                                                                    true,
+                                                                    sub_streams_table,
+                                                                    sub_cfg.streamsRankTable[i]};
+            sub_models.push_back(std::make_shared<CompiledModel>(model, plugin, sub_cfg, loaded_from_cache));
         }
-        stream_executor->run_sub_stream_and_wait(tasks);
+        message->set_sub_compiled_models(sub_models);
     }
+    // init sub stream threads of executor
+    // int sub_streams = m_cfg.streamExecutorConfig.get_sub_streams();
+    // if (sub_streams > 0 && stream_executor != nullptr) {
+    //     std::vector<Task> tasks;
+    //     tasks.resize(sub_streams);
+    //     for (auto&& task : tasks) {
+    //         task = [] {};
+    //     }
+    //     stream_executor->run_sub_stream_and_wait(tasks);
+    // }
 }
 
 CompiledModel::GraphGuard::Lock CompiledModel::get_graph() const {
@@ -167,6 +199,16 @@ std::shared_ptr<ov::IAsyncInferRequest> CompiledModel::create_infer_request() co
         std::make_shared<AsyncInferRequest>(std::static_pointer_cast<SyncInferRequest>(internal_request),
                                             get_task_executor(),
                                             get_callback_executor());
+    if (m_has_sub_compiled_models) {
+        auto message = message_manager();
+        auto sub_models = message->get_sub_compiled_models();
+        std::vector<std::shared_ptr<IAsyncInferRequest>> requests;
+        for (auto model : sub_models) {
+            requests.push_back(model->create_infer_request());
+        }
+        message->set_sub_infer_requests(requests);
+        async_infer_request->setSubInfer(true);
+    }
     return async_infer_request;
 }
 
diff --git a/src/plugins/intel_cpu/src/compiled_model.h b/src/plugins/intel_cpu/src/compiled_model.h
index 64caee60aeb..75a9d047db2 100644
--- a/src/plugins/intel_cpu/src/compiled_model.h
+++ b/src/plugins/intel_cpu/src/compiled_model.h
@@ -73,6 +73,8 @@ private:
      *       even from main thread
      */
     GraphGuard::Lock get_graph() const;
+
+    bool m_has_sub_compiled_models = false;
 };
 
 }   // namespace intel_cpu
diff --git a/src/plugins/intel_cpu/src/config.h b/src/plugins/intel_cpu/src/config.h
index 7580d0bb61b..8f97183a326 100644
--- a/src/plugins/intel_cpu/src/config.h
+++ b/src/plugins/intel_cpu/src/config.h
@@ -65,13 +65,17 @@ struct Config {
     int threadsPerStream = 0;
     ov::threading::IStreamsExecutor::ThreadBindingType threadBindingType = ov::threading::IStreamsExecutor::ThreadBindingType::NONE;
     ov::hint::PerformanceMode hintPerfMode = ov::hint::PerformanceMode::LATENCY;
+    std::vector<std::vector<int>> streamsRankTable;
     bool changedHintPerfMode = false;
     ov::log::Level logLevel = ov::log::Level::NO;
     uint32_t hintNumRequests = 0;
     bool enableCpuPinning = true;
     bool changedCpuPinning = false;
     ov::hint::SchedulingCoreType schedulingCoreType = ov::hint::SchedulingCoreType::ANY_CORE;
-    std::set<ov::hint::ModelDistributionPolicy> modelDistributionPolicy = {};
+    std::set<ov::hint::ModelDistributionPolicy> modelDistributionPolicy = {ov::hint::ModelDistributionPolicy::TENSOR_PARALLEL};
+    int streamsRankLevel = 1;
+    int numSubStreams = 0;
+    bool enableSubStreams = false;
     bool enableHyperThreading = true;
     bool changedHyperThreading = false;
 #if defined(OPENVINO_ARCH_X86) || defined(OPENVINO_ARCH_X86_64)
diff --git a/src/plugins/intel_cpu/src/cpu_streams_calculation.cpp b/src/plugins/intel_cpu/src/cpu_streams_calculation.cpp
index 08ba43f9eff..20af0f2fa84 100644
--- a/src/plugins/intel_cpu/src/cpu_streams_calculation.cpp
+++ b/src/plugins/intel_cpu/src/cpu_streams_calculation.cpp
@@ -55,9 +55,9 @@ std::vector<std::vector<int>> get_streams_info_table(const int input_streams,
                                       const IStreamsExecutor::Config::StreamsMode sub_streams_model,
                                       const int& target_proc) {
         stream_info[PROC_TYPE] = ALL_PROC;
+        stream_info[NUMBER_OF_STREAMS] = 1;
         stream_info[NUMBER_OF_STREAMS] =
             sub_streams_model == IStreamsExecutor::Config::StreamsMode::SUB_STREAMS_NULL ? 1 : -1;
-        stream_info[THREADS_PER_STREAM] = num_threads;
         update_ids_method(one_proc_info);
         streams_info_table.push_back(stream_info);
         stream_info[NUMBER_OF_STREAMS] = 0;
@@ -371,7 +371,7 @@ std::vector<std::vector<int>> get_streams_info_table(const int input_streams,
             create_one_stream(proc_socket_table[current_socket_id],
                               proc_type_table,
                               proc_socket_table[current_socket_id][ALL_PROC],
-                              IStreamsExecutor::Config::StreamsMode::SUB_STREAMS_NULL);
+                              IStreamsExecutor::Config::StreamsMode::SUB_STREAMS_FOR_SOCKET);
             for (size_t n_node = 0; n_node < proc_socket_table.size(); n_node++) {
                 if (n_node != size_t(current_socket_id)) {
                     create_one_stream(proc_socket_table[n_node],
@@ -380,6 +380,23 @@ std::vector<std::vector<int>> get_streams_info_table(const int input_streams,
                                       IStreamsExecutor::Config::StreamsMode::SUB_STREAMS_FOR_SOCKET);
                 }
             }
+            stream_info = streams_info_table[0];
+            stream_info[NUMBER_OF_STREAMS] = 1;
+            for (size_t n = 1; n < streams_info_table.size(); n++) {
+                if (streams_info_table[n][NUMBER_OF_STREAMS] == -1) {
+                    if (stream_info[PROC_TYPE] != streams_info_table[n][PROC_TYPE]) {
+                        stream_info[PROC_TYPE] = ALL_PROC;
+                    }
+                    stream_info[THREADS_PER_STREAM] += streams_info_table[n][THREADS_PER_STREAM];
+                    if (stream_info[STREAM_NUMA_NODE_ID] != streams_info_table[n][STREAM_NUMA_NODE_ID]) {
+                        stream_info[STREAM_NUMA_NODE_ID] = -1;
+                    }
+                    if (stream_info[STREAM_SOCKET_ID] != streams_info_table[n][STREAM_SOCKET_ID]) {
+                        stream_info[STREAM_SOCKET_ID] = -1;
+                    }
+                }
+            }
+            streams_info_table.insert(streams_info_table.begin(), stream_info);
             n_streams--;
         }
 
@@ -462,6 +479,33 @@ std::vector<std::vector<int>> get_streams_info_table(const int input_streams,
     return streams_info_table;
 }
 
+std::vector<std::vector<int>> get_streams_rank_table(const std::vector<std::vector<int>>& streams_info_table,
+                                                     const int input_rank_level,
+                                                     int& num_sub_streams) {
+    std::vector<std::vector<int>> rank_table = {};
+    num_sub_streams = 0;
+    std::vector<int> init_rank = {};
+    int rank_level = input_rank_level == 0 ? 1 : input_rank_level;
+    init_rank.resize(rank_level, 0);
+
+    for (auto& row : streams_info_table) {
+        if (row[NUMBER_OF_STREAMS] < 0) {
+            for (int i = 0; i < abs(row[NUMBER_OF_STREAMS]); i++) {
+                init_rank[rank_level - 1] = num_sub_streams + i;
+                rank_table.push_back(init_rank);
+            }
+            num_sub_streams -= row[NUMBER_OF_STREAMS];
+        }
+    }
+    if (rank_level == 2) {
+        for (int i = num_sub_streams / 2; i < num_sub_streams; i++) {
+            rank_table[i][0] = 1;
+            rank_table[i][1] -= num_sub_streams / 2;
+        }
+    }
+    return rank_table;
+}
+
 int get_model_prefer_threads(const int num_streams,
                              const std::vector<std::vector<int>> proc_type_table,
                              const std::shared_ptr<ov::Model>& model,
@@ -608,6 +652,13 @@ std::vector<std::vector<int>> generate_stream_info(const int streams,
                                                      ov::util::to_string(config.hintPerfMode),
                                                      config.modelDistributionPolicy,
                                                      proc_type_table);
+    // streams_info_table = {{1, 1, 56, 1, 1}, {-1, 1, 28, 1, 1}, {-1, 1, 28, 0, 0}};
+    if (config.modelDistributionPolicy.find(ov::hint::ModelDistributionPolicy::TENSOR_PARALLEL) !=
+        config.modelDistributionPolicy.end()) {
+        config.streamsRankTable =
+            get_streams_rank_table(streams_info_table, config.streamsRankLevel, config.numSubStreams);
+        config.enableSubStreams = true;
+    }
 
     auto cpu_pinning =
         get_cpu_pinning(config.enableCpuPinning, config.changedCpuPinning, proc_type_table, streams_info_table);
diff --git a/src/plugins/intel_cpu/src/cpu_streams_calculation.hpp b/src/plugins/intel_cpu/src/cpu_streams_calculation.hpp
index 584f9a2d9a0..e362c0373d8 100644
--- a/src/plugins/intel_cpu/src/cpu_streams_calculation.hpp
+++ b/src/plugins/intel_cpu/src/cpu_streams_calculation.hpp
@@ -53,6 +53,18 @@ std::vector<std::vector<int>> get_streams_info_table(const int input_streams,
                                                      const std::string input_perf_hint,
                                                      const std::set<ov::hint::ModelDistributionPolicy> hint_llm_distribution_policy,
                                                      const std::vector<std::vector<int>>& proc_type_table);
+
+/**
+ * @brief      Generate streams rank table for tensor parallel according to streams info table.
+ * @param[in]  streams_info_table is streams information table for tensor parallel.
+ * @param[in]  input_rank_level is depth of rank nesting.
+ * @param[out] num_sub_streams is number of sub streams for tensor parallel.
+ * @return     streams rank table which will be used by StreamsExecutor.
+ */
+std::vector<std::vector<int>> get_streams_rank_table(const std::vector<std::vector<int>>& streams_info_table,
+                                                     const int input_rank_level,
+                                                     int& num_sub_streams);
+
 /**
  * @brief      Get model_prefer_threads
  * @param[in]  num_streams is target streams set by user via NUM_STREAMS or hints.
diff --git a/src/plugins/intel_cpu/src/infer_request.cpp b/src/plugins/intel_cpu/src/infer_request.cpp
index 82b72137b6b..2c06e8de5cc 100644
--- a/src/plugins/intel_cpu/src/infer_request.cpp
+++ b/src/plugins/intel_cpu/src/infer_request.cpp
@@ -18,6 +18,7 @@
 #include "proxy_mem_mgr.h"
 #include "utils/general_utils.h"
 #include "utils/ngraph_utils.hpp"
+#include "openvino/runtime/threading/cpu_message.hpp"
 
 using OvString = ov::element_type_traits<ov::element::string>::value_type;
 
@@ -104,8 +105,24 @@ void SyncInferRequest::infer() {
     OV_ITT_SCOPED_TASK(itt::domains::intel_cpu, m_profiling_task);
     auto graphLock = m_compiled_model->get_graph();
     m_graph = &(graphLock._graph);
+    auto message = ov::threading::message_manager();
 
     throw_if_canceled();
+    // std::cout << "[ infer ] " << m_asyncRequest->m_has_sub_infers << "\n";
+    if (m_asyncRequest->m_has_sub_infers) {
+        message->server_wait();
+        ov::threading::MessageInfo msg_info;
+        msg_info.msg_type = ov::threading::MsgType::START_INFER;
+        ov::threading::Task task = [&] {
+            SyncInferRequest::sub_streams_infer();
+        };
+        msg_info.task = std::move(task);
+        message->send_message(msg_info);
+        message->infer_wait();
+        // std::cout << "------ infer end -----\n";
+        return;
+    }
+
     convert_batched_tensors();
     if (m_batched_tensors.size() > 0) {
         // batched_tensors will be updated for each infer, external_ptr should be update together
@@ -303,6 +320,15 @@ void SyncInferRequest::change_default_ptr() {
 }
 
 std::vector<ov::SoPtr<ov::IVariableState>> SyncInferRequest::query_state() const {
+    if (m_asyncRequest->m_has_sub_infers) {
+        auto message = ov::threading::message_manager();
+        std::vector<ov::SoPtr<ov::IVariableState>> states;
+        for (auto request : message->get_sub_infer_requests()) {
+            auto cur = request->query_state();
+            states.insert(states.end(), cur.begin(), cur.end());
+        }
+        return states;
+    }
     return {m_memory_states.begin(), m_memory_states.end()};
 }
 
@@ -600,6 +626,41 @@ SyncInferRequest::OutputControlBlock::OutputControlBlock(const ov::element::Type
     m_tensor = std::make_shared<Tensor>(memory);
 }
 
+void SyncInferRequest::sub_streams_infer() {
+    std::map<ov::Output<const ov::Node>, ov::SoPtr<ov::ITensor>> input_tensors;
+    auto message = ov::threading::message_manager();
+    auto requests = message->get_sub_infer_requests();
+    auto inputs = get_inputs();
+    auto outputs = get_outputs();
+
+    size_t requests_num = requests.size();
+    // std::cout << "[ sub_streams_infer ] inputs: " << inputs.size() << " requests: " << requests_num << "\n";
+
+    if (requests.size() > 0) {
+        for (const auto& output : outputs) {
+            auto tensor = requests[0]->get_tensor(output);
+            set_tensor(output, tensor);
+        }
+        for (size_t i = 0; i < requests_num; i++) {
+            for (auto& input : inputs) {
+                auto tensor = get_tensor(input);
+                requests[i]->set_tensor(input, tensor);
+            }
+
+            requests[i]->set_callback([message](const std::exception_ptr& ptr) {
+                // std::cout << "set_callback------ " << i << "\n";
+                ov::threading::MessageInfo msg_info;
+                msg_info.msg_type = ov::threading::MsgType::CALL_BACK;
+                message->send_message(msg_info);
+            });
+        }
+        for (size_t i = 0; i < requests_num; i++) {
+            // std::cout << "start_async : " << i << "\n";
+            requests[i]->start_async();
+        }
+    }
+}
+
 }   // namespace intel_cpu
 }   // namespace ov
 
diff --git a/src/plugins/intel_cpu/src/infer_request.h b/src/plugins/intel_cpu/src/infer_request.h
index e3839466f0c..2e097c87673 100644
--- a/src/plugins/intel_cpu/src/infer_request.h
+++ b/src/plugins/intel_cpu/src/infer_request.h
@@ -104,6 +104,8 @@ private:
 
     const ov::Output<const ov::Node>& get_internal_port(const ov::Output<const ov::Node>& port) const;
 
+    void sub_streams_infer();
+
 private:
     std::unordered_map<std::size_t, OutputControlBlock> m_outputControlBlocks;
 
diff --git a/src/plugins/intel_cpu/src/plugin.cpp b/src/plugins/intel_cpu/src/plugin.cpp
index 06e09029ead..0acddb55bd9 100644
--- a/src/plugins/intel_cpu/src/plugin.cpp
+++ b/src/plugins/intel_cpu/src/plugin.cpp
@@ -131,11 +131,13 @@ Plugin::Plugin() : deviceFullName(getDeviceFullName()), specialSetup(new CPUSpec
     });
     auto& ov_version = ov::get_openvino_version();
     m_compiled_model_runtime_properties["OV_VERSION"] = std::string(ov_version.buildNumber);
+    m_msg_manager = ov::threading::message_manager();
 }
 
 Plugin::~Plugin() {
     executor_manager()->clear("CPU");
     executor_manager()->clear("CPUStreamsExecutor");
+    executor_manager()->clear("CPUMainStreamExecutor");
     executor_manager()->clear("CPUCallbackExecutor");
 }
 
@@ -288,11 +290,11 @@ std::shared_ptr<ov::ICompiledModel> Plugin::compile_model(const std::shared_ptr<
     conf.readProperties(config, modelType);
     calculate_streams(conf, cloned_model);
 
-    if (conf.streamExecutorConfig.get_sub_stream_mode() ==
-        IStreamsExecutor::Config::StreamsMode::SUB_STREAMS_FOR_SOCKET) {
-        int num_sub_streams = conf.streamExecutorConfig.get_sub_streams();
-        transformations.SetSubStreamNum(num_sub_streams);
-    }
+    // if (conf.streamExecutorConfig.get_sub_stream_mode() ==
+    //     IStreamsExecutor::Config::StreamsMode::SUB_STREAMS_FOR_SOCKET) {
+    //     int num_sub_streams = conf.streamExecutorConfig.get_sub_streams();
+    //     transformations.SetSubStreamNum(num_sub_streams);
+    // }
 
     transformations.PostLpt();
     transformations.Snippets();
diff --git a/src/plugins/intel_cpu/src/plugin.h b/src/plugins/intel_cpu/src/plugin.h
index 3332d873fdc..c2d24e98ee6 100644
--- a/src/plugins/intel_cpu/src/plugin.h
+++ b/src/plugins/intel_cpu/src/plugin.h
@@ -6,6 +6,7 @@
 
 #include "compiled_model.h"
 #include "cpu_streams_calculation.hpp"
+#include "openvino/runtime/threading/cpu_message.hpp"
 
 namespace ov {
 namespace intel_cpu {
@@ -43,6 +44,8 @@ public:
         OPENVINO_THROW_NOT_IMPLEMENTED("Not Implemented get_default_context  is not supported by CPU plugin!");
     };
 
+    std::shared_ptr<ov::threading::MessageManager> m_msg_manager;
+
 private:
     ov::Any get_ro_property(const std::string& name, const ov::AnyMap& options) const;