[Feat] add connection manager (#978)

## Purpose
Add connection manager to kv asu transport
## Modifications 
- **ConnectionManager**: connection pool, which manages the three-layer
connection model: Pool (NPU->ASU:config) → Group (IP:Endpoint) → Channel
(QP)
1. As a member of AsuTransportImpl, the ConnectionManager registers the
link creation/deletion interface and link parameter information with
AsuTransportImpl. There is no derived class, and it is applicable to
different AsuTransportImpls.
2. Provides the following external interfaces: initialization,
destruction, route selection (RoundRobin/LeastLoaded), and exception
reporting (marking the channel status and adding the channel to the
exception list).
3. The background thread automatically re-establishes faulty channels.
(It checks the exception list, checks the inflight count of the channel,
and determines whether the channel can be destroyed and re-established.)
## Test

connection_concurrent_test.cc、connection_manager_test.cc、connection_transport_test.cc
<img width="1081" height="925"
alt="img_v3_02125_ae50282f-bb62-4c6e-810b-99f2a459ce5g"
src="https://github.com/user-attachments/assets/f8542fb1-4909-43d7-a08a-e51443bcc536"
/>

Co-authored-by: Frankie_lx <Frankie_lx@noreply.gitcode.com>

Author: Frankielx

ucm/transport/kv/asu/CMakeLists.txt

@@ -8,7 +8,7 @@ target_include_directories(asu_transport
         ${CMAKE_CURRENT_SOURCE_DIR}/trans/src
         ${UCM_ROOT_DIR}/ucm/shared/infra
 )
-target_link_libraries(asu_transport PUBLIC pthread)
+target_link_libraries(asu_transport PUBLIC infra_logger pthread)
 
 file(GLOB ASU_CLIENT_SOURCES CONFIGURE_DEPENDS client/src/*.cpp)
 add_library(asu_client SHARED ${ASU_CLIENT_SOURCES})
@@ -25,15 +25,18 @@ target_include_directories(asu_client
 target_link_libraries(asu_client PUBLIC asu_transport kv_common pthread)
 
 if(BUILD_UNIT_TESTS)
+    target_compile_definitions(asu_transport PRIVATE ASU_BUILD_TESTS)
     include(GoogleTest)
     file(GLOB_RECURSE ASU_TEST_SOURCE_FILES CONFIGURE_DEPENDS test/case/*.cc)
     file(GLOB_RECURSE ASU_CLIENT_TEST_SOURCE_FILES CONFIGURE_DEPENDS client/test/*.cpp)
     list(APPEND ASU_TEST_SOURCE_FILES ${ASU_CLIENT_TEST_SOURCE_FILES})
     add_executable(asu.test ${ASU_TEST_SOURCE_FILES})
+    target_compile_definitions(asu.test PRIVATE ASU_BUILD_TESTS)
     target_include_directories(asu.test PRIVATE
         ${CMAKE_CURRENT_SOURCE_DIR}/common
         ${CMAKE_CURRENT_SOURCE_DIR}/client/src
         ${CMAKE_CURRENT_SOURCE_DIR}/trans/src
+        ${UCM_ROOT_DIR}/ucm/shared/infra
     )
     target_link_libraries(asu.test PRIVATE
         asu_client

ucm/transport/kv/asu/client/test/asu_client_impl_test.cpp

@@ -195,6 +195,13 @@ class FakeTransport : public AsuTransport {
         return Check(taskId, result);
     }
 
+    Status StubCheck(TaskId taskId, TaskResult& result) override { return Check(taskId, result); }
+
+    Status StubWait(TaskId taskId, std::uint64_t timeoutMs, TaskResult& result) override
+    {
+        return Wait(taskId, timeoutMs, result);
+    }
+
     Status RegisterRegions(const std::vector<MemoryRegion>& regions,
                            std::vector<RegisterResult>& results) override
     {

ucm/transport/kv/asu/common/task_manager_base.h

@@ -88,7 +88,7 @@ class TaskManagerBase {
         return Status::OK();
     }
 
-private:
+protected:
     State initialState_;
     std::string taskName_;
     std::atomic<TaskId> nextTaskId_{1};

ucm/transport/kv/asu/test/case/asu_client_e2e_metrics_test.cc

@@ -451,6 +451,13 @@ class InMemoryAsuTransport final : public AsuTransport {
         return Check(taskId, result);
     }
 
+    Status StubCheck(TaskId taskId, TaskResult& result) override { return Check(taskId, result); }
+
+    Status StubWait(TaskId taskId, std::uint64_t timeoutMs, TaskResult& result) override
+    {
+        return Wait(taskId, timeoutMs, result);
+    }
+
     Status RegisterRegions(const std::vector<MemoryRegion>& regions,
                            std::vector<RegisterResult>& results) override
     {