04 RouDi 守护进程与服务发现

RouDi：iceoryx_posh/source/roudi/；运行时：iceoryx_posh/source/runtime/；CaPro：iceoryx_posh/source/capro/
源码根：ros2_humble/src/eclipse-iceoryx/iceoryx（版本 2.0.6）

1. RouDi 的职责与定位

RouDi（Routing and Discovery）是 iceoryx 的中心守护进程，一台机器（或一个 domain）上只运行一个。它负责：

职责	实现
创建/销毁共享内存段	RouDiMemoryManager + MemoryProvider
进程注册与生命周期监控	ProcessManager（心跳超时清理）
端口（Publisher/Subscriber/Client/Server）的分配与销毁	PortManager + PortPool
服务发现路由（CaPro 协议状态机的”邮差”）	PortManager::doDiscovery()
内省（introspection）数据发布	内置 topic

关键设计：RouDi 不在数据路径上。数据 chunk 从发布者到订阅者的传递完全发生在应用进程之间的共享内存里（见 05 篇）；RouDi 只在”控制面”工作——建立/拆除连接、分配资源。RouDi 崩溃不会中断已建立连接上的数据传输（但无法再建新连接、新进程无法注册）。

2. 启动流程

2.1 入口：iox-roudi 可执行文件

roudi_main.cpp 是 iox-roudi 的 main：解析命令行 → 解析 TOML 配置 → 构造 IceOryxRouDiApp 并 run()：

int main(int argc, char* argv[]) noexcept
{
    using iox::roudi::IceOryxRouDiApp;

    iox::config::CmdLineParserConfigFileOption cmdLineParser;
    auto cmdLineArgs = cmdLineParser.parse(argc, argv);
    ...
    iox::config::TomlRouDiConfigFileProvider configFileProvider(cmdLineArgs.value());

    auto roudiConfig = configFileProvider.parse();
    ...
    IceOryxRouDiApp roudi(cmdLineArgs.value(), roudiConfig.value());
    return roudi.run();
}

2.2 RouDiApp / IceOryxRouDiApp

基类 RouDiApp（application/roudi_app.cpp）处理信号（SIGINT/SIGTERM 通过信号量唤醒主线程退出）、日志级别、monitoring 模式、配置校验（段/mempool 不能为空）。派生类 IceOryxRouDiApp::run() 创建两个核心对象后阻塞等待退出信号：

uint8_t IceOryxRouDiApp::run() noexcept
{
    if (m_run)
    {
        static cxx::optional<IceOryxRouDiComponents> m_rouDiComponents;
        auto componentsScopeGuard = cxx::makeScopedStatic(m_rouDiComponents, m_config);

        static cxx::optional<RouDi> roudi;
        auto roudiScopeGuard =
            cxx::makeScopedStatic(roudi,
                                  m_rouDiComponents.value().rouDiMemoryManager,
                                  m_rouDiComponents.value().portManager,
                                  RouDi::RoudiStartupParameters{m_monitoringMode,
                                                                true,
                                                                RouDi::RuntimeMessagesThreadStart::IMMEDIATE,
                                                                m_compatibilityCheckLevel,
                                                                m_processKillDelay});
        waitForSignal();
    }
    return EXIT_SUCCESS;
}

IceOryxRouDiComponents 的构造顺序值得注意：先清理遗留的 RouDi IPC channel，再创建并”宣布”共享内存，最后才构造 PortManager：

IceOryxRouDiComponents::IceOryxRouDiComponents(const RouDiConfig_t& roudiConfig) noexcept
    : rouDiMemoryManager(roudiConfig)
    , portManager([&]() -> IceOryxRouDiMemoryManager* {
        // this temporary object will create a roudi IPC channel
        // and close it immediatelly
        // if there was an outdated roudi IPC channel, it will be cleaned up
        // if there is an outdated IPC channel, the start of the apps will be terminated
        runtime::IpcInterfaceBase::cleanupOutdatedIpcChannel(roudi::IPC_CHANNEL_ROUDI_NAME);

        rouDiMemoryManager.createAndAnnounceMemory().or_else([](RouDiMemoryManagerError error) {
            LogFatal() << "Could not create SharedMemory! Error: " << error;
            errorHandler(Error::kROUDI_COMPONENTS__SHARED_MEMORY_UNAVAILABLE, nullptr, iox::ErrorLevel::FATAL);
        });
        return &rouDiMemoryManager;
    }())
{
}

2.3 内存体系：MemoryProvider / MemoryBlock / RouDiMemoryManager

RouDi 的内存创建采用两级抽象（roudi/memory/ 目录）：

• MemoryBlock：一段有 size/alignment 需求的逻辑内存块，如 PortPoolMemoryBlock（端口池）、MemPoolCollectionMemoryBlock（introspection mempool）、MemPoolSegmentManagerMemoryBlock（SegmentManager，管理所有用户数据段）。
• MemoryProvider：把若干 MemoryBlock 汇总，计算总大小并实际创建内存。POSIX 实现是 PosixShmMemoryProvider，基于 shm_open + mmap：

cxx::expected<void*, MemoryProviderError> PosixShmMemoryProvider::createMemory(const uint64_t size,
                                                                               const uint64_t alignment) noexcept
{
    if (alignment > posix::pageSize())
    {
        return cxx::error<MemoryProviderError>(MemoryProviderError::MEMORY_ALIGNMENT_EXCEEDS_PAGE_SIZE);
    }

    posix::SharedMemoryObject::create(m_shmName, size, m_accessMode, m_openMode, nullptr)
        .and_then([this](auto& sharedMemoryObject) {
            sharedMemoryObject.finalizeAllocation();
            m_shmObject.emplace(std::move(sharedMemoryObject));
        });
    ...
    return cxx::success<void*>(baseAddress);
}

DefaultRouDiMemory 组装出管理段（名字为 iceoryx_mgmt，见 iceoryx_posh_types.hpp 中 SHM_NAME），它容纳 introspection mempool 与 SegmentManager：

DefaultRouDiMemory::DefaultRouDiMemory(const RouDiConfig_t& roudiConfig) noexcept
    : m_introspectionMemPoolBlock(introspectionMemPoolConfig())
    , m_segmentManagerBlock(roudiConfig)
    , m_managementShm(SHM_NAME, posix::AccessMode::READ_WRITE, posix::OpenMode::PURGE_AND_CREATE)
{
    m_managementShm.addMemoryBlock(&m_introspectionMemPoolBlock).or_else([](auto) {
        errorHandler(
            Error::kROUDI__DEFAULT_ROUDI_MEMORY_FAILED_TO_ADD_INTROSPECTION_MEMORY_BLOCK, nullptr, ErrorLevel::FATAL);
    });
    m_managementShm.addMemoryBlock(&m_segmentManagerBlock).or_else([](auto) {
        errorHandler(
            Error::kROUDI__DEFAULT_ROUDI_MEMORY_FAILED_TO_ADD_SEGMENT_MANAGER_MEMORY_BLOCK, nullptr, ErrorLevel::FATAL);
    });
}

RouDiMemoryManager::createAndAnnounceMemory() 逐个调用 provider 的 create()，全部成功后 announceMemoryAvailable()（roudi/memory/roudi_memory_manager.cpp 第 65~89 行）。用户数据段（payload segment）由 SegmentManager/MePooSegment 按 TOML 配置逐段创建，每段有 reader/writer POSIX 用户组权限。

所以磁盘上可见的共享内存文件通常是：/dev/shm/iceoryx_mgmt + 每个用户组一个数据段（如 /dev/shm/<group>）。

2.4 RouDi 对象：两个线程

RouDi 构造函数启动两个线程：

// run the threads
m_monitoringAndDiscoveryThread = std::thread(&RouDi::monitorAndDiscoveryUpdate, this);
posix::setThreadName(m_monitoringAndDiscoveryThread.native_handle(), "Mon+Discover");

if (roudiStartupParameters.m_runtimesMessagesThreadStart == RuntimeMessagesThreadStart::IMMEDIATE)
{
    startProcessRuntimeMessagesThread();
}

线程	周期	职责
Mon+Discover	DISCOVERY_INTERVAL = 100 ms	ProcessManager::run() → 心跳监控 + PortManager::doDiscovery()
IPC-msg-process	阻塞收消息	处理应用发来的 IPC 请求（注册、建端口等）

IPC 消息处理线程创建名为 roudi 的 IPC channel 并打印著名的 “RouDi is ready for clients”：

void RouDi::processRuntimeMessages() noexcept
{
    runtime::IpcInterfaceCreator roudiIpcInterface{IPC_CHANNEL_ROUDI_NAME};

    // the logger is intentionally not used, to ensure that this message is always printed
    std::cout << "RouDi is ready for clients" << std::endl;

    while (m_runHandleRuntimeMessageThread)
    {
        // read RouDi's IPC channel
        runtime::IpcMessage message;
        if (roudiIpcInterface.timedReceive(m_runtimeMessagesThreadTimeout, message))
        {
            auto cmd = runtime::stringToIpcMessageType(message.getElementAtIndex(0).c_str());
            std::string runtimeName = message.getElementAtIndex(1);

            processMessage(message, cmd, RuntimeName_t(cxx::TruncateToCapacity, runtimeName));
        }
    }
}

3. 进程注册：PoshRuntime 与 RouDi 的 IPC channel

3.1 IPC channel 的实体

应用与 RouDi 之间的控制通道是按平台选择的（iceoryx_hoofs/platform/*/platform_settings.hpp）：Linux/QNX/macOS 上是 Unix Domain Socket（using IoxIpcChannelType = iox::posix::UnixDomainSocket;，Linux 见 iceoryx_hoofs/platform/linux/include/iceoryx_hoofs/platform/platform_settings.hpp 第 39 行），Windows 上是 NamedPipe。历史版本用的 POSIX message queue 在 2.0 中仍保留实现（iceoryx_hoofs/internal/posix_wrapper/message_queue.hpp），但默认平台配置不再使用。

消息本身是 ASCII 文本、以分隔符拼接的字段序列（IpcMessage），收发封装在 IpcInterfaceBase（source/runtime/ipc_interface_base.cpp）中。通道拓扑是每进程一条：RouDi 有名为 roudi 的接收通道；每个应用按自己的 runtime 名字创建一条接收应答的通道。

3.2 应用侧：PoshRuntime::initRuntime()

应用第一行代码通常是 iox::runtime::PoshRuntime::initRuntime("app_name")，它构造单例 PoshRuntimeImpl：

PoshRuntimeImpl::PoshRuntimeImpl(cxx::optional<const RuntimeName_t*> name, const RuntimeLocation location) noexcept
    : PoshRuntime(name)
    , m_ipcChannelInterface(roudi::IPC_CHANNEL_ROUDI_NAME, *name.value(), runtime::PROCESS_WAITING_FOR_ROUDI_TIMEOUT)
    , m_ShmInterface([&] {
        // in case the runtime is located in the same process like RouDi the shm is already opened;
        ...
        return location == RuntimeLocation::SAME_PROCESS_LIKE_ROUDI
                   ? cxx::nullopt
                   : cxx::optional<SharedMemoryUser>({m_ipcChannelInterface.getShmTopicSize(),
                                                      m_ipcChannelInterface.getSegmentId(),
                                                      m_ipcChannelInterface.getSegmentManagerAddressOffset()});
    }())
{
}

两步：① IpcRuntimeInterface 完成注册握手；② SharedMemoryUser 用握手拿到的信息 mmap 共享内存。

3.3 注册握手（REG / REG_ACK）

IpcRuntimeInterface 构造函数是一个状态机：WAIT_FOR_ROUDI → SEND_REGISTER_REQUEST → WAIT_FOR_REGISTER_ACK → FINISHED，默认最多等待 RouDi PROCESS_WAITING_FOR_ROUDI_TIMEOUT = 60 s。注册请求携带进程名、pid、uid、时间戳与版本信息：

// send IpcMessageType::REG to RouDi

IpcMessage sendBuffer;
int pid = getpid();
cxx::Expects(pid >= 0);
sendBuffer << IpcMessageTypeToString(IpcMessageType::REG) << m_runtimeName << cxx::convert::toString(pid)
           << cxx::convert::toString(posix::PosixUser::getUserOfCurrentProcess().getID())
           << cxx::convert::toString(transmissionTimestamp)
           << static_cast<cxx::Serialization>(version::VersionInfo::getCurrentVersion()).toString();

bool successfullySent = m_RoudiIpcInterface.timedSend(sendBuffer, 100_ms);

RouDi 侧 ProcessManager::addProcess() 做版本兼容性检查、把进程加入 m_processList（上限 MAX_PROCESS_NUMBER），并回 REG_ACK，内容是管理段大小 + SegmentManager 在管理段内的偏移 + 时间戳 + segment id：

// send REG_ACK and BaseAddrString
runtime::IpcMessage sendBuffer;

auto offset = rp::BaseRelativePointer::getOffset(m_mgmtSegmentId, m_segmentManager);
sendBuffer << runtime::IpcMessageTypeToString(runtime::IpcMessageType::REG_ACK)
           << m_roudiMemoryInterface.mgmtMemoryProvider()->size() << offset << transmissionTimestamp
           << m_mgmtSegmentId;

m_processList.back().sendViaIpcChannel(sendBuffer);

应用拿到 ACK 后，SharedMemoryUser 以 OPEN_EXISTING 方式 mmap iceoryx_mgmt，用相对指针机制 registerPtr(segmentId, baseAddress, size) 注册地址映射，再顺着 SegmentManager 打开自己有权限的所有数据段（source/runtime/shared_memory_user.cpp 第 29~85 行）。每个进程 mmap 的虚拟地址不同，这正是所有共享内存内部结构必须使用相对指针（RelativePointer）的原因。

3.4 心跳 keep-alive 与进程监控

• 应用侧：PoshRuntimeImpl 的定时器周期性调用 sendKeepAliveAndHandleShutdownPreparation() → 发 KEEPALIVE 消息（间隔 PROCESS_KEEP_ALIVE_INTERVAL = 3 × 100 ms = 300 ms）。
• RouDi 侧：KEEPALIVE 分支调用 m_prcMgr->updateLivelinessOfProcess(runtimeName) 刷新时间戳；Mon+Discover 线程里的 monitorProcesses() 检查超时：

void ProcessManager::monitorProcesses() noexcept
{
    auto currentTimestamp = mepoo::BaseClock_t::now();

    auto processIterator = m_processList.begin();
    while (processIterator != m_processList.end())
    {
        if (processIterator->isMonitored())
        {
            auto timediff = units::Duration(currentTimestamp - processIterator->getTimestamp());
            ...
            if (timediff > runtime::PROCESS_KEEP_ALIVE_TIMEOUT)
            {
                LogWarn() << "Application " << processIterator->getName() << " not responding (last response "
                          << timediff.toMilliseconds() << " milliseconds ago) --> removing it";
                ...
                m_portManager.deletePortsOfProcess(processIterator->getName());

                m_processIntrospection->removeProcess(static_cast<int32_t>(processIterator->getPid()));

                // delete application
                processIterator = m_processList.erase(processIterator);

超时阈值 PROCESS_KEEP_ALIVE_TIMEOUT = 5 × 300 ms = 1.5 s（iceoryx_posh_types.hpp 第 266~268 行）。是否监控由 RouDi 启动时的 monitoring 模式 决定（MonitoringMode::ON/OFF，命令行 -m 切换；ON 时所有进程被监控）。

优雅退出走另一条路：应用析构 PoshRuntimeImpl 时发 TERMINATION，RouDi 清理端口后回 TERMINATION_ACK。

4. PortManager 与端口分配

4.1 PortPool：为什么端口数据也在共享内存

所有端口的数据结构本体（PublisherPortData、SubscriberPortData、ConditionVariableData、NodeData……）都存放在管理段中的 PortPool（固定容量的 FixedPositionContainer 集合，roudi/port_pool.cpp）。原因：

1. 端口是应用与 RouDi 的共享状态机。应用侧用 PublisherPortUser/SubscriberPortUser 视图操作它（写 m_offeringRequested、收发 chunk），RouDi 侧用 PublisherPortRouDi/SubscriberPortRouDi 视图操作同一块数据（处理 CaPro 状态迁移、把订阅者队列挂到发布者）。两个进程要同时看到同一对象，只能放在共享内存。
2. 零拷贝路由需要。订阅时 RouDi 把订阅者的 ChunkQueueData 的指针（相对指针形式）直接塞进发布者的 ChunkDistributor 队列列表——发布者随后 push chunk 时完全不需要 RouDi 参与。
3. 崩溃恢复。应用死掉后 RouDi 仍能访问其端口数据并回收 chunk（releaseAllChunks()）。

端口创建后，RouDi 把端口对象在管理段内的相对指针偏移通过 IPC 回给应用（CREATE_PUBLISHER_ACK 携带 offset + segmentId），应用侧换算回本进程的绝对地址：

if (stringToIpcMessageType(IpcMessage.c_str()) == IpcMessageType::CREATE_PUBLISHER_ACK)

{
    rp::BaseRelativePointer::id_t segmentId{0U};
    cxx::convert::fromString(receiveBuffer.getElementAtIndex(2U).c_str(), segmentId);
    rp::BaseRelativePointer::offset_t offset{0U};
    cxx::convert::fromString(receiveBuffer.getElementAtIndex(1U).c_str(), offset);
    auto ptr = rp::BaseRelativePointer::getPtr(segmentId, offset);
    return cxx::success<PublisherPortUserType::MemberType_t*>(
        reinterpret_cast<PublisherPortUserType::MemberType_t*>(ptr));
}

4.2 创建 Publisher 端口的完整链路

应用: iox::popo::Publisher<T> 构造
  → BasePublisher 构造: PoshRuntime::getInstance().getMiddlewarePublisher(service, options)
      (posh_runtime_impl.cpp: 发 IpcMessageType::CREATE_PUBLISHER)
RouDi: RouDi::processMessage(CREATE_PUBLISHER)          roudi.cpp:225
  → ProcessManager::addPublisherForProcess()            process_manager.cpp:452
      → SegmentManager::getSegmentInformationWithWriteAccessForUser(user)  // 找该用户可写的数据段
      → PortManager::acquirePublisherPortData()         port_manager.cpp:851
          → 检查通信策略(ONE_TO_MANY 下同一 service 只许一个 publisher)
          → PortPool::addPublisherPort(...)             // 在共享内存里构造 PublisherPortData
          → doDiscoveryForPublisherPort()               // 立即跑一轮发现
      → 回 CREATE_PUBLISHER_ACK(offset, segmentId)
应用: 相对指针 → 绝对地址，构造 PublisherPortUser 视图

注意 addPublisherForProcess 里传入的 payloadDataSegmentMemoryManager 是按发起进程的 POSIX 用户挑选的数据段 memory manager——这决定了该 publisher 之后 loan() 从哪个 mempool 拿 chunk。

4.3 doDiscovery：发现主循环

Mon+Discover 线程每 100 ms 调 PortManager::doDiscovery()，轮询所有端口收集状态变化并互相路由：

void PortManager::doDiscovery() noexcept
{
    handlePublisherPorts();

    handleSubscriberPorts();

    handleServerPorts();

    handleClientPorts();

    handleInterfaces();

    handleNodes();

    handleConditionVariables();
}

5. CaPro 协议（Canonical Protocol）

5.1 ServiceDescription：service/instance/event 三元组

capro::ServiceDescription（iceoryx_posh/capro/service_description.hpp 第 80、156~160 行）由三个字符串组成：m_serviceString / m_instanceString / m_eventString。ROS 2 的 rmw_iceoryx、cyclonedds 的 iceoryx 网关都会把 topic 名映射到这个三元组。两个端口能匹配的首要条件是三元组完全相等（PortManager::isCompatiblePubSub 第 483 行）。

5.2 CaproMessage 类型

enum class CaproMessageType : uint8_t
{
    NOTYPE = 0,
    FIND,
    OFFER,
    STOP_OFFER,
    SUB,
    UNSUB,
    CONNECT,
    DISCONNECT,
    ACK,
    NACK,
    PUB,
    REQ,
    RES,
    PING,
    PONG,
    MESSGAGE_TYPE_END
};

pub/sub 用到的核心是 OFFER/STOP_OFFER（发布者宣告）、SUB/UNSUB（订阅者请求）、ACK/NACK（应答）；client/server 用 CONNECT/DISCONNECT。CaproMessage 里最关键的字段是 m_chunkQueueData——订阅者接收队列在共享内存里的指针，它随 SUB 消息一路传给发布者。

5.3 订阅如何路由到发布者

双方端口各自是一个小状态机，RouDi 只是搬运消息：

1. 应用调 Subscriber 构造（subscribeOnCreate 默认 true）→ 设置 m_subscribeRequested。
2. 下一轮 doDiscovery() 中，SubscriberPortSingleProducer::tryGetCaProMessage() 产出 SUB 消息并携带自己的队列指针：

if (currentSubscribeRequest && (SubscribeState::NOT_SUBSCRIBED == currentSubscriptionState))
{
    getMembers()->m_subscriptionState.store(SubscribeState::SUBSCRIBE_REQUESTED, std::memory_order_relaxed);

    capro::CaproMessage caproMessage(capro::CaproMessageType::SUB, BasePort::getMembers()->m_serviceDescription);
    caproMessage.m_chunkQueueData = static_cast<void*>(&getMembers()->m_chunkReceiverData);
    caproMessage.m_historyCapacity = getMembers()->m_options.historyRequest;

    return cxx::make_optional<capro::CaproMessage>(caproMessage);
}

3. PortManager::sendToAllMatchingPublisherPorts()（port_manager.cpp 第 500 行）遍历所有 publisher，对三元组匹配且 QoS 兼容者调用 dispatchCaProMessageAndGetPossibleResponse(SUB)。
4. 发布者侧把订阅者队列挂进自己的 ChunkSender（即 ChunkDistributor 的队列容器），成功则回 ACK：

if (capro::CaproMessageType::SUB == caProMessage.m_type)
{
    const auto ret = m_chunkSender.tryAddQueue(
        static_cast<PublisherPortData::ChunkQueueData_t*>(caProMessage.m_chunkQueueData),
        caProMessage.m_historyCapacity);
    if (!ret.has_error())
    {
        responseMessage.m_type = capro::CaproMessageType::ACK;
    }
}

5. ACK 被 RouDi 转回订阅者，订阅者状态迁移到 SUBSCRIBED。若没有匹配发布者则 RouDi 直接给订阅者发 NACK，订阅者进入 WAIT_FOR_OFFER，等以后收到 OFFER 再自动补发 SUB（subscriber_port_single_producer.cpp 第 73~82 行）。

反方向：发布者 OFFER 时 sendToAllMatchingSubscriberPorts()（第 539 行）把 OFFER 派发给所有等待的订阅者，触发它们回 SUB，随后同样走上面 3-5 步（一轮 discovery 内完成握手）。

QoS 兼容性检查（阻塞策略、history 支持）：

bool PortManager::isCompatiblePubSub(const PublisherPortRouDiType& publisher,
                                     const SubscriberPortType& subscriber) const noexcept
{
    if (subscriber.getCaProServiceDescription() != publisher.getCaProServiceDescription())
    {
        return false;
    }

    auto& pubOpts = publisher.getOptions();
    auto& subOpts = subscriber.getOptions();

    const bool blockingPoliciesAreCompatible =
        !(pubOpts.subscriberTooSlowPolicy == popo::ConsumerTooSlowPolicy::DISCARD_OLDEST_DATA
          && subOpts.queueFullPolicy == popo::QueueFullPolicy::BLOCK_PRODUCER);

    const bool historyRequestIsCompatible = !subOpts.requiresPublisherHistorySupport || pubOpts.historyCapacity > 0;

    return blockingPoliciesAreCompatible && historyRequestIsCompatible;
}

另外 OFFER/STOP_OFFER 还会转发给所有 InterfacePort（sendToAllMatchingInterfacePorts）——这是 gateway（如 iceoryx-dds、rmw 网关）监听全网服务变化的机制。

5.4 服务注册表（ServiceRegistry）

RouDi 在 PortManager 内维护一份 ServiceRegistry（roudi/service_registry.cpp），OFFER 时 addPublisher、STOP_OFFER 时 removePublisher（server 同理）。每次变化都会把整个注册表作为一个 chunk 发布到内置 topic 上：

void PortManager::publishServiceRegistry() const noexcept
{
    ...
    PublisherPortUserType publisher(m_serviceRegistryPublisherPortData.value());
    publisher
        .tryAllocateChunk(sizeof(ServiceRegistry),
                          alignof(ServiceRegistry),
                          CHUNK_NO_USER_HEADER_SIZE,
                          CHUNK_NO_USER_HEADER_ALIGNMENT)
        .and_then([&](auto& chunk) {
            auto sample = static_cast<ServiceRegistry*>(chunk->userPayload());

            // It's ok to copy as the modifications happen in the same thread and not concurrently
            *sample = m_serviceRegistry;

            publisher.sendChunk(chunk);
        })
        .or_else([](auto&) { LogWarn() << "Could not allocate a chunk for the service registry!"; });
}

对应的内置服务三元组是 {SERVICE_DISCOVERY_SERVICE_NAME, SERVICE_DISCOVERY_INSTANCE_NAME, SERVICE_DISCOVERY_EVENT_NAME}（port_manager.cpp 第 71~74 行）。

6. 服务发现 API（2.0 的 ServiceDiscovery / findService）

2.0 把 1.0 里”应用发 FIND 消息、RouDi 应答”的模式改成了纯 pub/sub：应用侧 iox::runtime::ServiceDiscovery 内部就是一个订阅上述内置 topic 的 subscriber，findService() 先 take() 最新注册表快照再本地过滤：

void ServiceDiscovery::update()
{
    // allows us to use update and hence findService concurrently
    std::lock_guard<std::mutex> lock(m_serviceRegistryMutex);
    m_serviceRegistrySubscriber.take().and_then([&](popo::Sample<const roudi::ServiceRegistry>& serviceRegistrySample) {
        *m_serviceRegistry = *serviceRegistrySample;
    });
}

void ServiceDiscovery::findService(const cxx::optional<capro::IdString_t>& service,
                                   const cxx::optional<capro::IdString_t>& instance,
                                   const cxx::optional<capro::IdString_t>& event,
                                   const cxx::function_ref<void(const capro::ServiceDescription&)>& callableForEach,
                                   const popo::MessagingPattern pattern) noexcept
{
    ...
    update();

    switch (pattern)
    {
    case popo::MessagingPattern::PUB_SUB:
    {
        m_serviceRegistry->find(
            service, instance, event, [&](const roudi::ServiceRegistry::ServiceDescriptionEntry& serviceEntry) {
                if (serviceEntry.publisherCount > 0)
                {
                    callableForEach(serviceEntry.serviceDescription);
                }
            });
        break;
    }

三个参数都是 optional——nullopt 表示通配。还可以把 ServiceDiscovery 挂到 WaitSet/Listener 上监听 SERVICE_REGISTRY_CHANGED 事件（enableEvent，同文件第 82 行），实现事件驱动的发现。

ServiceRegistry::find() 本身是对最多 MAX_NUMBER_OF_SERVICES 条目的线性匹配（service_registry.cpp 第 141~165 行），每个条目带 publisherCount/serverCount 引用计数（multi-set 语义，同一 service 多个 publisher 只占一个条目）。

7. 内省（Introspection）服务

RouDi 自己作为一个”进程”注册（m_processIntrospection.addProcess(getpid(), IPC_CHANNEL_ROUDI_NAME)，roudi.cpp 第 65 行），并用内部 publisher 端口周期发布以下内置 topic（服务描述常量定义在 iceoryx_posh_roudi_types.hpp / introspection_types.hpp）：

内置服务	内容
IntrospectionMempoolService	各 mempool 的 chunk 使用量/最小剩余
IntrospectionProcessService	已注册进程列表（pid、名字、node）
IntrospectionPortService / IntrospectionPortThroughputService / IntrospectionSubscriberPortChangingDataService	端口清单、吞吐、订阅状态

这些端口在 PortManager 构造时通过 acquireInternalPublisherPortData() 创建（port_manager.cpp 第 80~96 行），chunk 从专用的 introspection mempool（管理段内，配置见 default_roudi_memory.cpp 的 introspectionMemPoolConfig()）分配。官方 iceoryx_introspection（iox-introspection-client）就是订阅这些 topic 的 ncurses 客户端。内置服务名以 Introspection/服务发现常量为 service 名，普通应用禁止创建同名 publisher（acquirePublisherPortDataWithoutDiscovery 中 isInternal() 检查，返回 INTERNAL_SERVICE_DESCRIPTION_IS_FORBIDDEN）。

8. RouDi 配置

8.1 TOML 配置文件

iox-roudi -c roudi_config.toml；不提供时先找默认路径（defaultConfigFilePath，即 /etc/iceoryx/roudi_config.toml），再退回编译期默认配置（TomlRouDiConfigFileProvider::parse()，roudi_config_toml_file_provider.cpp 第 59~68 行）。格式（版本必须为 1）：

[general]
version = 1

[[segment]]                 # 一个共享内存数据段；省略 reader/writer 则默认当前进程所在用户组
reader = "readergroup"      # 拥有只读权限的 POSIX 用户组
writer = "writergroup"      # 拥有读写权限的 POSIX 用户组

  [[segment.mempool]]       # 段内一个 mempool
  size = 128                # chunk 的 payload 大小(字节)
  count = 10000             # chunk 数量

  [[segment.mempool]]
  size = 1024
  count = 5000

解析核心（段数上限 MAX_SHM_SEGMENTS、每段 mempool 上限 MAX_NUMBER_OF_MEMPOOLS）：

for (auto mempool : *mempools)
{
    auto chunkSize = mempool->get_as<uint32_t>("size");
    auto chunkCount = mempool->get_as<uint32_t>("count");
    ...
    mempoolConfig.addMemPool({*chunkSize, *chunkCount});
}
parsedConfig.m_sharedMemorySegments.push_back(
    {iox::posix::PosixGroup::string_t(iox::cxx::TruncateToCapacity, reader),
     iox::posix::PosixGroup::string_t(iox::cxx::TruncateToCapacity, writer),
     mempoolConfig});

注意 mempool 的 size 是 payload 大小，实际 chunk 会加上 ChunkHeader 并按 MemPool::CHUNK_MEMORY_ALIGNMENT 对齐；mempool 必须按 chunk 大小递增排序（MemoryManager::addMemPool 强制检查）。

8.2 主要命令行参数

参数	作用
-c <file>	TOML 配置路径
-m on/off	monitoring 模式（cmd_line_args.hpp 第 31 行默认 MonitoringMode::ON）
-l <level>	日志级别
-u <id>	unique RouDi id（写入 UniquePortId，用于多 RouDi 隔离）
-k <sec>	关闭时等待应用退出的 kill delay（默认 PROCESS_DEFAULT_KILL_DELAY = 45 s）
-x <level>	版本兼容性检查级别

9. 崩溃行为

9.1 应用崩溃

• monitoring ON：心跳超时 1.5 s 后，RouDi 主动 deletePortsOfProcess()——对每个端口执行有序关闭（先注入 STOP_OFFER/UNSUB 走完 CaPro 流程，通知对端），然后 releaseAllChunks() 回收该进程持有的所有 chunk（port_manager.cpp 第 796~848 行）。之后同名进程可重新注册。
• monitoring OFF：RouDi 不会主动发现崩溃。同名进程重新注册时走”已存在则视为崩溃重注册”路径：

findProcess(name)
    .and_then([&](auto& process) {
        // process is already in list (i.e. registered)
        // depending on the mode we clean up the process resources and register it again
        // if it is monitored, we reject the registration and wait for automatic cleanup
        // otherwise we remove the process ourselves and register it again
        ...
        // process exists, we expect that the existing process crashed
        LogWarn() << "Application " << name << " crashed. Re-registering application";

        // remove the existing process and add the new process afterwards, we do not send ack to new process
        constexpr TerminationFeedback TERMINATION_FEEDBACK{TerminationFeedback::DO_NOT_SEND_ACK_TO_PROCESS};
        if (!this->searchForProcessAndRemoveIt(name, TERMINATION_FEEDBACK))

已知限制：如果应用恰好死在持有 ChunkDistributor 锁的瞬间，RouDi 清理时会遇到死锁风险并触发 FATAL（chunk_distributor.inl 的 cleanup() 注释明确说明，见 05 篇 9.3 节）。

9.2 RouDi 崩溃 / 关闭

• 已建立的 pub/sub 连接数据传输不受影响（数据路径不经过 RouDi），但：没有发现、没有新端口、进程无法注册/注销，keep-alive 发送失败会打印警告。共享内存文件仍在 /dev/shm 中，重启 RouDi 时以 PURGE_AND_CREATE 模式重建管理段（default_roudi_memory.cpp 第 30 行），旧应用必须重启（其映射的旧段已失效，且版本/会话不匹配）。
• 正常关闭（SIGINT/SIGTERM）：RouDi::shutdown() 先停 discovery 线程，然后向所有注册进程发 SIGTERM 并等待其发回 TERMINATION，超过 kill delay（默认 45 s）仍存活则 SIGKILL（roudi.cpp 第 102~133 行）。requestShutdownOfAllProcesses() 同时调用 m_portManager.unblockRouDiShutdown()，把所有 publisher 强制 STOP_OFFER，避免因 BLOCK_PRODUCER 策略卡死关闭流程。

10. 时序总览

---

下一篇

• 05-零拷贝数据路径详解.md：loan()/publish()/take() 端到端调用链、ChunkDistributor、无锁队列与引用计数回收。