[CPU] FC node tensor parallel (openvinotoolkit#25088)

### Details:
 - *Enable sub stream to compile model on each socket*
 - *Limit CPU cores to 32 when TP is enabled*
 - *Split fc into two sub-fcs at node level instead of graph level*
- *Concat fc output in the horizon direction. Drop allgatherv and
allreduce for performance reason*
 - *Supported llm fc int8 and 4bit mode*
 - *Remove previous version fc TP logic*

### Tickets:
 - *ticket-id*

---------

Co-authored-by: sunxiaoxia2022 <xiaoxia.sun@intel.com>
Co-authored-by: Shen, Wanglei <wanglei.shen@intel.com>

Author: 3 people

src/inference/dev_api/openvino/runtime/threading/cpu_message.hpp

@@ -0,0 +1,58 @@
+// 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 <assert.h>
+
+#include "openvino/runtime/common.hpp"
+#include "openvino/runtime/threading/istreams_executor.hpp"
+#include "openvino/runtime/threading/itask_executor.hpp"
+
+
+namespace ov {
+
+namespace threading {
+enum MsgType {CALL_BACK};
+
+struct MessageInfo {
+    MsgType msg_type;
+};
+
+class OPENVINO_RUNTIME_API MessageManager {
+public:
+    MessageManager();
+
+    void send_message(const MessageInfo& msg_info);
+
+    void server_wait();
+
+    ~MessageManager();
+
+    void set_num_sub_streams(int num_sub_streams);
+
+    int get_num_sub_streams();
+private:
+    int _num_sub_streams = 0;
+    std::vector<MessageInfo> _messageQueue;
+    std::mutex _msgMutex;
+    std::condition_variable _msgCondVar;
+};
+
+OPENVINO_RUNTIME_API std::shared_ptr<MessageManager> message_manager();
+}  // namespace threading
+}  // namespace ov

src/inference/dev_api/openvino/runtime/threading/cpu_streams_executor.hpp

@@ -55,7 +55,7 @@ class OPENVINO_RUNTIME_API CPUStreamsExecutor : public IStreamsExecutor {
 
     int get_socket_id() override;
 
-    void run_sub_stream(Task task, int id) override;
+    std::vector<int> get_rank() override;
 
 private:
     struct Impl;

src/inference/dev_api/openvino/runtime/threading/istreams_executor.hpp

@@ -38,6 +38,19 @@ class OPENVINO_RUNTIME_API IStreamsExecutor : virtual public ITaskExecutor {
      */
     using Ptr = std::shared_ptr<IStreamsExecutor>;
 
+    enum MsgType{
+        TP,
+        START_INFER,
+        CALL_BACK
+    };
+
+    struct MessageInfo{
+        MsgType msg_type;
+        std::vector<int> rank;
+        void* buf;
+        Task task;
+    };
+
     /**
      * @brief Defines inference thread binding type
      */
@@ -95,6 +108,7 @@ class OPENVINO_RUNTIME_API IStreamsExecutor : virtual public ITaskExecutor {
         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,21 +138,24 @@ class OPENVINO_RUNTIME_API IStreamsExecutor : virtual public ITaskExecutor {
          * @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,
                int threads_per_stream = 0,
                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{std::move(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{std::move(streams_info_table)} {
+              _streams_info_table{std::move(streams_info_table)},
+              _rank{rank} {
             update_executor_config();
         }
 
@@ -197,6 +214,9 @@ class OPENVINO_RUNTIME_API IStreamsExecutor : virtual public ITaskExecutor {
         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;
@@ -250,31 +270,17 @@ class OPENVINO_RUNTIME_API IStreamsExecutor : virtual public ITaskExecutor {
     virtual int get_socket_id() = 0;
 
     /**
-     * @brief Execute the task in the current thread using streams executor configuration and constraints
-     * @param task A task to start
+     * @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 void execute(Task task) = 0;
+    virtual std::vector<int> get_rank() = 0;
 
     /**
-     * @brief Execute ov::Task inside sub stream of task executor context
+     * @brief Execute the task in the current thread using streams executor configuration and constraints
      * @param task A task to start
-     * @param id Sub stream id
-     */
-    virtual void run_sub_stream(Task task, int id) = 0;
-
-    /**
-     * @brief Execute all of the tasks and waits for its completion.
-     *        Default run_sub_stream_and_wait() method implementation uses run_sub_stream() pure virtual method
-     *        and higher level synchronization primitives from STL.
-     *        The task is wrapped into std::packaged_task which returns std::future.
-     *        std::packaged_task will call the task and signal to std::future that the task is finished
-     *        or the exception is thrown from task
-     *        Than std::future is used to wait for task execution completion and
-     *        task exception extraction
-     * @note run_sub_stream_and_wait() does not copy or capture tasks!
-     * @param tasks A vector of tasks to execute
      */
-    void run_sub_stream_and_wait(const std::vector<Task>& tasks);
+    virtual void execute(Task task) = 0;
 };
 
 }  // namespace threading

src/inference/src/dev/threading/cpu_message.cpp

@@ -0,0 +1,96 @@
+// 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() {
+    _num_sub_streams = 0;
+}
+
+MessageManager::~MessageManager() {}
+
+void MessageManager::send_message(const MessageInfo& msg_info) {
+    {
+        std::lock_guard<std::mutex> lock(_msgMutex);
+        _messageQueue.push_back(msg_info);
+    }
+    _msgCondVar.notify_all();
+}
+
+void MessageManager::server_wait() {
+    assert(_num_sub_streams);
+    MsgType msg_type;
+    int count = 0;
+    bool isStopped = false;
+    while (!isStopped) {
+        std::vector<MessageInfo> msgQueue;
+        {
+            std::unique_lock<std::mutex> lock(_msgMutex);
+            _msgCondVar.wait(lock, [&] {
+                return !_messageQueue.empty();
+            });
+            std::swap(_messageQueue, msgQueue);
+        }
+
+        for (auto rec_info : msgQueue) {
+            msg_type = rec_info.msg_type;
+            if (msg_type == CALL_BACK) {  // CALL_BACK
+                count++;
+                if (count == _num_sub_streams) {
+                    count = 0;
+                    isStopped = true;
+                }
+            }
+        }
+    };
+}
+
+void MessageManager::set_num_sub_streams(int num_sub_streams) {
+    _num_sub_streams = num_sub_streams;
+}
+
+int MessageManager::get_num_sub_streams() {
+    return _num_sub_streams;
+}
+
+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