diff --git a/docs/source/design/transfer-engine/jetty-ack-timeout-rebuild.md b/docs/source/design/transfer-engine/jetty-ack-timeout-rebuild.md new file mode 100644 index 0000000000..4965d9c919 --- /dev/null +++ b/docs/source/design/transfer-engine/jetty-ack-timeout-rebuild.md @@ -0,0 +1,15 @@ +--- +orphan: true +--- + +# Jetty ACK Timeout 重建方案（已废弃） + +**状态：已废弃。请勿按本文实现或评审。** + +权威方案见： + +[jetty-single-rebuild-plan.md](./jetty-single-rebuild-plan.md) + +废弃原因：本文把「对端 import/bind 同步」写成阻塞前置条件；经对照 kunpeng +`UrmaEndpoint` 数据面（单边 READ/WRITE 打到 remote segment，不依赖对端 jetty +收发语义）后，采用**本端单 Jetty 排空重建、不需要对端协议同步**的方案。 diff --git a/docs/source/design/transfer-engine/jetty-single-rebuild-plan.md b/docs/source/design/transfer-engine/jetty-single-rebuild-plan.md new file mode 100644 index 0000000000..956cb201ce --- /dev/null +++ b/docs/source/design/transfer-engine/jetty-single-rebuild-plan.md @@ -0,0 +1,237 @@ +--- +orphan: true +--- + +# Jetty ACK Timeout（status=9）单 Jetty 重建方案 + +状态：实现中（分支 `feat/jetty-ack-timeout-single-rebuild`）— **权威方案**（取代已废弃的 `jetty-ack-timeout-rebuild.md`） +范围：kunpeng / UB 传输路径（`UrmaEndpoint` / `UbWorkerPool`）；不涉及 tent +关联错误码： + +| CR status | 枚举 | 本方案处理 | +|---|---|---| +| 4 | `URMA_CR_LOC_ACCESS_ERR` | 不重建；保持现有 slice 重试 / 失败逻辑 | +| 9 | `URMA_CR_ACK_TIMEOUT_ERR` | 单 Jetty 排空 → 删除 → 重建，**纯本地操作** | + +--- + +## 1. 背景与问题 + +8 节点场景下，Jetty completion 出现 status=9（ACK 超时 / 重传超限）后，现有逻辑： + +1. poll 到非 SUCCESS → slice 进入重试 +2. 重试耗尽 → `deleteEndpointByPtr` → `UrmaEndpoint::deconstruct` +3. `deconstruct` 直接 `unbind / unimport / delete_jetty`，**没有** ERROR 排空与 `FLUSH_ERR_DONE` 栅栏 + +结果：短暂路径问题被放大成整 endpoint（甚至整节点）不可用，常需重启才能恢复。 + +--- + +## 2. 核心思路 + +**只重建出问题的那个 jetty，不删整个 EP，且不需要对端同步协议。** + +``` +status=9 出现在 jetty_list_[i] + ① urma_modify_jetty(jetty_list_[i], ERROR) ← 排空 + ② poll 到 FLUSH_ERR_DONE(local_id) ← 排空完成栅栏 + ③ flush → unbind/unimport → delete → create + → import(原 peer id) → bind → 更新槽位 ← 安全替换（仍纯本地） +``` + +**为什么不需要对端同步？** + +kunpeng 数据面是单边 `URMA_OPC_READ/WRITE`，DMA 目标是对端 **segment** +（`slice->ub.r_seg`），不是往对端 jetty 做 SEND/RECV。 + +`urma_post_jetty_send_wr` 的第一个参数是**本地 jetty**。握手虽会 +`import` + `bind` 并对 `wr.tjetty` 赋值，但： + +- 现有路径在 `tjetty` 缺失时只 `LOG(ERROR)`，仍继续 post +- `remote_jetty == NULL` 为空分支，未当硬失败 +- 重建后本端用**仍有效的对端 jetty id** 做 re-import/rebind 即可恢复本端发出方向；无需通知对端改其 import 视图 + +反向流量若仍绑旧本端 id，对端可能自行报错并走自己的本地重建——两侧各自恢复，不上同步握手。 + +--- + +## 3. 状态机 + +每个 jetty 槽位有独立状态： + +``` +ACTIVE ──(status=9，无其它槽在重建)──► DRAINING ──(FLUSH_ERR_DONE)──► REBUILDING + │ │ + │ ◄── create + 本端 rebind + 更新槽位 ───────────────┘ + │ + └──(status=9，已有槽在重建)──► PENDING_DRAIN ──(前槽重建完成后触发)──► DRAINING + + 任一环节失败 / 超时 ──► 回退：deleteEndpointByPtr +``` + +- `ACTIVE`：正常收发 +- `DRAINING`：已 `modify(ERROR)`，禁止 post，等待 `FLUSH_ERR_DONE` +- `REBUILDING`：排空完成，正在 flush / 删建 / rebind +- `PENDING_DRAIN`：本槽也收到 status=9，但同 EP 已有 jetty 在重建；禁止 post + （选槽跳过，避免故障槽继续吃流量），等前一个重建完成后由其尾部串行触发 + `modify(ERROR)` 进入 DRAINING +- 同 EP 任意时刻最多一个 jetty 处于 DRAINING/REBUILDING（串行） + +--- + +## 4. 关键设计 + +### 4.1 per-jetty 状态 + +```cpp +enum JettyState { ACTIVE, DRAINING, REBUILDING, PENDING_DRAIN }; +std::vector jetty_state_; // 与 jetty_list_ 平行 +std::unordered_map jetty_id_map_; // jetty_id → slot index +std::vector peer_jetty_id_; // 每槽对端 id，delete 前保留 +std::vector jetty_epoch_; // 每槽重建代次，丢弃旧代次的迟到 CR +``` + +`peer_jetty_id_` 在握手 `doSetupConnection` 时写入；段 C 在 delete 本地 jetty +前依赖它做 re-import，不得只依赖即将销毁的 `imported_jetty_map_` 指针键。 + +### 4.2 选槽策略 + +`submitPostSend` 选槽时跳过非 ACTIVE 槽： + +``` +随机选一个槽 → 非 ACTIVE 则重试 → 全部不可用则返回 0（上层重试） +``` + +### 4.3 锁策略 + +| 阶段 | 是否持 `lock_` | 说明 | +|---|---|---| +| 段 A：modify(ERROR) | 是 | 与 post 互斥（UMDK 约束） | +| 段 B：等 FLUSH_ERR_DONE | 否 | 由 performPoll 顺带消费，不占锁 | +| 段 C：flush/删建/rebind | 是 | 替换 `jetty_list_` 与 `imported_jetty_map_` | + +### 4.4 FLUSH_ERR_DONE 消费 + +走现有 `performPoll` → `UrmaContext::poll`。**必须在按 `user_ctx` 解 slice +之前分流**（假 CQE 的 `user_ctx` 无效；现有空 ctx `continue` 会丢掉栅栏）： + +``` +if cr.status == FLUSH_ERR_DONE: + local_id → jetty_id_map_ → (endpoint, slot) + endpoint->onFlushDone(slot) // DRAINING → REBUILDING，触发段 C + continue // 假 CQE，不走 slice 路径，不改 outstanding slice 语义外的 depth 时需单独约定 +``` + +### 4.5 status=9 分流与触发时机 + +**首次** poll 到 `ACK_TIMEOUT_ERR (9)` 即触发排空（不等重试耗尽）；`onJettyError` +必须幂等（已在 DRAINING/REBUILDING/PENDING_DRAIN 则不再 `modify`）。 +若同 EP 已有 jetty 在重建，则把本槽标记为 `PENDING_DRAIN` 排队：选槽立即跳过 +它，等前一个重建完成后由其尾部调用 `startDrainUnlocked` 串行启动排空。 + +``` +if cr.status == ACK_TIMEOUT_ERR (9): + slot ← slice->ub.jetty_depth 反查，或 cr.local_id → jetty_id_map_ + slice->ub.endpoint → UrmaEndpoint + endpoint->onJettyError(slot) // 段 A + slice 计入 retry（换 ACTIVE 槽重发） +``` + +### 4.6 outstanding 记账 + +排空期间 inflight WR 会以 error/flush 类 CQE 回来。要求： + +- 带有效 `user_ctx` 的失败 CQE：仍走现有 `jetty_depth_set` / retry 路径扣 + `wr_depth_list_` 与 JFC outstanding +- `FLUSH_ERR_DONE`：不当作 slice；不得 `markSuccess` / 不得当失败 slice 入队 +- **每个 WR 恰好完成一次**：`modify(ERROR)` 后 inflight WR 要么以 + `WR_FLUSH_ERR` 经 JFC poll 回来，要么被段 C 的 `urma_flush_jetty` 回收；两者 + 统一经 `processWrCompletion` 记账（`jetty_depth_set` 延迟扣减 + poll 返回值 + 累计 JFC outstanding）。因此段 C 删除旧 jetty 后**不做**额外的 + `wr_depth_list_[slot]` / JFC 清零——延迟扣减要等 poll 返回后才 apply，在段 C + 里提前清零会双扣（实现后已删除原方案中的"归零兜底"） +- 旧 jetty 删除后 `++jetty_epoch_[slot]`；slice 在 post 时记录 + `slice->ub.jetty_epoch`，旧代次的迟到/重复 CR 在 `processWrCompletion` 里按 + epoch 不匹配直接丢弃，不参与记账、不再入 retry 队列 + +### 4.7 段 C 完整顺序（持锁） + +``` +1. urma_flush_jetty（回收残余 WR，逐条经 processWrCompletion 交付； + 注意与 poll 可能重复，见 §7） +2. unbind → unimport（旧本地 jetty 上的对端视图） +3. delete_jetty；从 jetty_id_map_ 删旧 id；++jetty_epoch_[slot] +4. create_jetty（同 JFC/JFR 配置） +5. import(peer_jetty_id_[slot]) → bind(新 jetty, imported) +6. 更新 jetty_list_[slot]、imported_jetty_map_、jetty_id_map_ +7. jetty_state_[slot] = ACTIVE +8. 扫描 PENDING_DRAIN 槽：有则立即 startDrainUnlocked（modify(ERROR)）， + 仍保持同 EP 串行 +``` + +任一步失败 → 回退 `deleteEndpointByPtr`。 + +--- + +## 5. 执行计划 + +### Step 1：加状态与映射 + +- `UrmaEndpoint` 增加 `jetty_state_`、`jetty_id_map_`、`peer_jetty_id_` +- `construct()` / 握手初始化，`deconstruct()` 清理 +- `submitPostSend` 选槽跳过非 ACTIVE + +### Step 2：poll 分流 + +- `UrmaContext::poll`：先识别 `FLUSH_ERR_DONE`（`local_id`），再处理 + `ACK_TIMEOUT_ERR`（endpoint + 槽位） +- 假 CQE 不走 slice 路径 + +### Step 3：段 A — 触发排空 + +- `onJettyError(slot)`：持锁 → 幂等检查 → `modify(ERROR)` → 标 DRAINING +- 记录 drain 起始时间，供超时用 + +### Step 4：段 B — 等待栅栏 + +- `onFlushDone(slot)`：标 REBUILDING，触发段 C + +### Step 5：段 C — 重建 + +- 按 §4.7 完整顺序执行 + +### Step 6：超时降级 + +- flush-done 等待超时（可配置，默认 3s）→ 回退 `deleteEndpointByPtr` +- create / import / bind 失败同样降级 + +### Step 7：日志与测试 + +- 关键路径日志：`jetty_id`、slot、耗时、降级原因 +- 单测：状态机、选槽跳过、假 CQE 路由、幂等 `onJettyError` +- 集成 / 故障注入：status=9 后该槽恢复 ACTIVE，同 EP 其它槽可继续；超时路径删 EP +- 无 UMDK 硬件时，flush 与真实 ACK timeout 行为标为硬件验证项 + +--- + +## 6. 改动文件 + +| 文件 | 改动 | +|---|---| +| `urma_endpoint.h` | `JettyState`（含 `PENDING_DRAIN`）、`jetty_state_`、`jetty_id_map_`、`peer_jetty_id_`、`jetty_epoch_`、`onJettyError()` / `onFlushDone()` / `startDrainUnlocked()` | +| `urma_endpoint.cpp` | 状态机、选槽跳过、段 A/C、握手写入 `peer_jetty_id_` | +| `ub_context.cpp` | poll / worker 侧配合 status=9 与 `FLUSH_ERR_DONE`（若分流落在 context poll 则改 `urma_endpoint.cpp` 中 `UrmaContext::poll`） | + +--- + +## 7. 风险与开放问题 + +1. **urma_flush_jetty 与 poll 重复**：flush 返回的 WR 级 CR 是否已在 JFC poll + 中出现过，需实测确认，避免 double-complete。实现按「每个 WR 恰好完成一次」 + 记账，并用 `jetty_epoch_` 把旧代次的迟到 CR 整体丢弃兜底；若实测发现 flush + 与 poll 会重复交付同一 WR，需重新评估 slice 指针解引用的安全性（届时 CR 里 + 的 `user_ctx` 可能指向已回收的 slice）。 +2. **共享 JFC 假 CQE 过滤**：多 jetty 共享 JFC 时，严格按 `local_id` 匹配，不能假设顺序。 +3. **超时阈值**：flush-done 等待 3s 是否合适，需结合 `err_timeout` 和现场标定。 +4. **硬件 hang 场景**：本方案解决软件放大故障；若根因是设备/驱动 hang，重建仍可能失败，保留删 EP 降级路径。 +5. **反向路径**：对端仍绑旧本端 id 时可能自行报 9 并本地重建；观察即可，本期不上对端协议。 diff --git a/mooncake-transfer-engine/include/transport/kunpeng_transport/ub_context.h b/mooncake-transfer-engine/include/transport/kunpeng_transport/ub_context.h index 56767c9dee..db61ca1985 100644 --- a/mooncake-transfer-engine/include/transport/kunpeng_transport/ub_context.h +++ b/mooncake-transfer-engine/include/transport/kunpeng_transport/ub_context.h @@ -188,13 +188,19 @@ class UbContext { // * Successful slices have markSuccess() called in place and are NOT // returned (they may be recycled by the submitting thread the // moment markSuccess() runs). - // * Failed slices are returned in failed_slices[0..num_failed-1] for - // the caller to apply retry / markFailed. - // Returns the total number of completions polled (>= 0), or a negative - // error code. - virtual int poll(int num_entries, Transport::Slice** failed_slices, - int& num_failed, + // * Failed slices are appended to failed_slices for the caller to apply + // retry / markFailed. Implementations may also append completions + // recovered while draining a jetty, so the vector is unbounded. + // * Stale completions from a previous jetty generation are dropped and + // are not counted in the return value. + // * Endpoints scheduled for deletion are appended to deferred_deletes; + // the caller must delete them only after jetty_depth_set accounting. + // Returns the number of resolved WR completions (>= 0), or a negative + // error code. Fence markers and dropped stale completions are excluded. + virtual int poll(int num_entries, + std::vector& failed_slices, std::unordered_map& jetty_depth_set, + std::vector& deferred_deletes, int jfc_index = 0) = 0; virtual volatile int* outstandingCount(int jfc_index) = 0; @@ -259,6 +265,10 @@ class UbContext { UbTransport& engine() const { return engine_; } + bool traceWorkRequestFlushedErrors() const { + return show_work_request_flushed_error_; + } + uint8_t portNum() const { return port_; } int activeSpeed() const { return active_speed_; } diff --git a/mooncake-transfer-engine/include/transport/kunpeng_transport/urma/urma_endpoint.h b/mooncake-transfer-engine/include/transport/kunpeng_transport/urma/urma_endpoint.h index 6b6fa7076a..25d3e71a2d 100644 --- a/mooncake-transfer-engine/include/transport/kunpeng_transport/urma/urma_endpoint.h +++ b/mooncake-transfer-engine/include/transport/kunpeng_transport/urma/urma_endpoint.h @@ -18,7 +18,10 @@ #include #include #include +#include +#include #include +#include #include "common.h" #include "config.h" #include "urma_api.h" @@ -47,6 +50,8 @@ static urma_import_seg_flag_t import_flag = { .reserved = 0}}; // define the UrmaContext class +class UrmaEndpoint; + class UrmaContext : public UbContext { friend class UrmaEndpoint; @@ -58,8 +63,9 @@ class UrmaContext : public UbContext { int unregisterMemoryRegion(uint64_t va) override; int doProcessContextEvents() override; void* retrieveRemoteSeg(const std::string& value) override; - int poll(int num_entries, Transport::Slice** failed_slices, int& num_failed, + int poll(int num_entries, std::vector& failed_slices, std::unordered_map& jetty_depth_set, + std::vector& deferred_deletes, int jfc_index) override; volatile int* outstandingCount(int jfc_index) override; int submitPostSend( @@ -81,6 +87,17 @@ class UrmaContext : public UbContext { static bool uninit(); static bool init(); + void registerJettyOwner(uint32_t jetty_id, UrmaEndpoint* endpoint, + int slot); + void unregisterJettyOwner(uint32_t jetty_id); + bool findJettyOwner(uint32_t jetty_id, UrmaEndpoint** endpoint, int* slot); + void addDrainingEndpoint(UrmaEndpoint* endpoint); + void removeDrainingEndpoint(UrmaEndpoint* endpoint); + void checkJettyDrainTimeouts( + std::unordered_map& jetty_depth_set, + std::vector& failed_slices, + std::vector& deferred_deletes); + private: int construct(GlobalConfig& config) override; int deconstruct() override; @@ -146,11 +163,33 @@ class UrmaContext : public UbContext { urma_import_seg_flag_t import_flag_ = mooncake::import_flag; std::unordered_map import_tseg_map; + + RWSpinlock jetty_owner_lock_; + struct JettyOwner { + UrmaEndpoint* endpoint = nullptr; + int slot = -1; + }; + std::unordered_map jetty_owner_map_; + std::unordered_set draining_endpoints_; }; // define the UrmaEndpoint class class UrmaEndpoint : public UbEndPoint { + // UrmaContext::poll drives the jetty state machine via processWrCompletion. + friend class UrmaContext; + public: + // PENDING_DRAIN: the slot hit ACK timeout while another jetty of this + // endpoint was draining/rebuilding. It is excluded from post selection + // and drained once the in-flight rebuild completes (rebuilds are + // serialized per endpoint). + enum JettyState { + ACTIVE = 0, + DRAINING = 1, + REBUILDING = 2, + PENDING_DRAIN = 3 + }; + UrmaEndpoint(UrmaContext* context) : context_(context), jfc_outstanding_(nullptr) {} @@ -173,6 +212,21 @@ class UrmaEndpoint : public UbEndPoint { const std::string toString() const override; + // Called from UrmaContext::poll on ACK timeout / flush-done / drain + // timeout. + void onJettyError(int slot, std::vector& deferred_deletes); + void onFlushDone(int slot, + std::unordered_map& jetty_depth_set, + std::vector& failed_slices, + std::vector& deferred_deletes, + int& resolved_wr_count); + void checkDrainTimeout( + std::unordered_map& jetty_depth_set, + std::vector& failed_slices, + std::vector& deferred_deletes); + + int findSlotByDepth(volatile int* depth) const; + private: void disconnectUnlocked() override; @@ -187,6 +241,36 @@ class UrmaEndpoint : public UbEndPoint { uint32_t peer_jetty_num, std::string* reply_msg = nullptr); + bool hasNonActiveJettyUnlocked() const; + int selectActiveJettyUnlocked(); + // Transitions `slot` to DRAINING via urma_modify_jetty(ERROR) and arms + // the flush-done wait. Caller must hold lock_. Returns ERR_ENDPOINT if + // modify failed (caller should fall back to deleting the endpoint). + int startDrainUnlocked(int slot); + int rebuildJettyUnlocked( + int slot, std::unordered_map& jetty_depth_set, + std::vector& failed_slices, + std::vector& deferred_deletes, int& resolved_wr_count); + + // Delivers one WR completion to the normal success/failure path. + // Returns true when the completion resolved a live WR (and must be counted + // in the JFC outstanding accounting). Returns false for completions from a + // stale jetty generation, which were already resolved during the rebuild + // flush and must be dropped entirely. When allow_error_trigger is false + // (the caller already holds lock_ while rebuilding/draining), an + // ACK_TIMEOUT completion is delivered as a plain failure without + // re-entering onJettyError. + bool processWrCompletion( + urma_cr_t& cr, std::unordered_map& jetty_depth_set, + std::vector& failed_slices, + std::vector& deferred_deletes, int jfc_index, + bool allow_error_trigger); + + int recreateJettyUnlocked(int slot, urma_jfc_t* reuse_jfc, + urma_jfr_t* reuse_jfr); + + static constexpr uint64_t kJettyDrainTimeoutNs = 3000000000ull; // 3s + private: UrmaContext* context_; urma_token_t urma_token = {.token = 0xACFE}; @@ -195,6 +279,14 @@ class UrmaEndpoint : public UbEndPoint { int max_wr_depth_; volatile int* jfc_outstanding_; std::unordered_map imported_jetty_map_; + + std::vector jetty_state_; + std::unordered_map jetty_id_map_; + std::vector peer_jetty_id_; + std::string peer_eid_; + uint64_t drain_start_ns_ = 0; + int draining_slot_ = -1; + std::vector jetty_epoch_; }; } // namespace mooncake #endif // URMA_ENDPOINT_H diff --git a/mooncake-transfer-engine/include/transport/transport.h b/mooncake-transfer-engine/include/transport/transport.h index c235bb3424..256d81e5a2 100644 --- a/mooncake-transfer-engine/include/transport/transport.h +++ b/mooncake-transfer-engine/include/transport/transport.h @@ -157,6 +157,7 @@ class Transport { struct { uint64_t dest_addr; volatile int *jetty_depth; + uint64_t jetty_epoch; uint32_t retry_cnt; uint32_t max_retry_cnt; void *r_seg; diff --git a/mooncake-transfer-engine/src/transport/kunpeng_transport/ub_context.cpp b/mooncake-transfer-engine/src/transport/kunpeng_transport/ub_context.cpp index 46a3d1e467..60e58fe2f0 100644 --- a/mooncake-transfer-engine/src/transport/kunpeng_transport/ub_context.cpp +++ b/mooncake-transfer-engine/src/transport/kunpeng_transport/ub_context.cpp @@ -406,22 +406,26 @@ void UbWorkerPool::performPoll(int thread_id) { // are NOT returned from poll(), so this worker never dereferences them // after they may have been recycled by the submitting thread. std::unordered_map jetty_depth_set; + std::vector deferred_deletes; std::vector failed_slices; for (int jfc_index = thread_id; jfc_index < context_.jfcCount(); jfc_index += kTransferWorkerCount) { - UbTransport::Slice* failed[kPollCount]; - int num_failed = 0; - int nr_poll = context_.poll(kPollCount, failed, num_failed, - jetty_depth_set, jfc_index); - if (nr_poll < 0) { + // poll() may also append completions recovered while draining a jetty + // (up to the full queue depth), so this must be a vector rather than + // a fixed-size array of kPollCount. + std::vector failed; + const size_t failed_before = failed.size(); + int nr_resolved = context_.poll(kPollCount, failed, jetty_depth_set, + deferred_deletes, jfc_index); + if (nr_resolved < 0) { LOG(ERROR) << "Worker: Failed to poll jetty for complete"; continue; } - int num_success = nr_poll - num_failed; + const size_t new_failed = failed.size() - failed_before; + const int num_success = nr_resolved - static_cast(new_failed); success_nr_polls += num_success; processed_slice_count += num_success; - for (int i = 0; i < num_failed; ++i) { - UbTransport::Slice* slice = failed[i]; + for (auto* slice : failed) { assert(slice); failed_nr_polls++; if (context_.active() && failed_nr_polls > 32 && @@ -439,13 +443,17 @@ void UbWorkerPool::performPoll(int thread_id) { redispatch_counter_++; } } - if (nr_poll) - __sync_fetch_and_sub(context_.outstandingCount(jfc_index), nr_poll); + if (nr_resolved) + __sync_fetch_and_sub(context_.outstandingCount(jfc_index), + nr_resolved); } for (auto& entry : jetty_depth_set) __sync_fetch_and_sub(entry.first, entry.second); + for (auto* endpoint : deferred_deletes) + context_.deleteEndpointByPtr(endpoint); + // Slices that hit max_retry: final markFailed() after all reads (and the // jetty depth returns above) are done. Failed slices were never published // by poll(), so they remained safe to deref up to this point. diff --git a/mooncake-transfer-engine/src/transport/kunpeng_transport/urma/mock_urma.cpp b/mooncake-transfer-engine/src/transport/kunpeng_transport/urma/mock_urma.cpp index c392577a6b..a05e04280c 100644 --- a/mooncake-transfer-engine/src/transport/kunpeng_transport/urma/mock_urma.cpp +++ b/mooncake-transfer-engine/src/transport/kunpeng_transport/urma/mock_urma.cpp @@ -346,11 +346,12 @@ urma_jetty_t *urma_create_jetty(urma_context_t *ctx, urma_jetty_cfg_t *cfg) { if (!ctx || !cfg || context_map.find(ctx) == context_map.end()) { return nullptr; } + static std::atomic next_jetty_id{1}; urma_jetty_t *jetty = new urma_jetty_t; memset(&jetty->jetty_id.eid, 0, sizeof(urma_eid_t)); jetty->jetty_id.eid.raw[0] = 1; jetty->jetty_id.uasid = 0; - jetty->jetty_id.id = 1; + jetty->jetty_id.id = next_jetty_id.fetch_add(1); jetty->jetty_cfg = *cfg; jetty->remote_jetty = nullptr; jetty_map[jetty] = 1; @@ -420,6 +421,16 @@ urma_status_t urma_modify_jetty(urma_jetty_t *jetty, urma_jetty_attr_t *attr) { return URMA_SUCCESS; } +int urma_flush_jetty(urma_jetty_t *jetty, int cr_cnt, urma_cr_t *cr) { + (void)cr_cnt; + (void)cr; + std::shared_lock lock(g_rw_mutex); + if (!jetty || jetty_map.find(jetty) == jetty_map.end()) { + return -1; + } + return 0; +} + urma_status_t urma_post_jetty_send_wr(urma_jetty_t *jetty, urma_jfs_wr_t *wr, urma_jfs_wr_t **bad_wr) { { diff --git a/mooncake-transfer-engine/src/transport/kunpeng_transport/urma/urma_endpoint.cpp b/mooncake-transfer-engine/src/transport/kunpeng_transport/urma/urma_endpoint.cpp index 06c65d0385..3a7e18e401 100644 --- a/mooncake-transfer-engine/src/transport/kunpeng_transport/urma/urma_endpoint.cpp +++ b/mooncake-transfer-engine/src/transport/kunpeng_transport/urma/urma_endpoint.cpp @@ -518,11 +518,69 @@ bool UrmaContext::transEidFromString(const std::string& eid_str, return index == URMA_EID_SIZE; } -int UrmaContext::poll(int num_entries, Transport::Slice** failed_slices, - int& num_failed, +void UrmaContext::registerJettyOwner(uint32_t jetty_id, UrmaEndpoint* endpoint, + int slot) { + RWSpinlock::WriteGuard guard(jetty_owner_lock_); + jetty_owner_map_[jetty_id] = JettyOwner{endpoint, slot}; +} + +void UrmaContext::unregisterJettyOwner(uint32_t jetty_id) { + RWSpinlock::WriteGuard guard(jetty_owner_lock_); + jetty_owner_map_.erase(jetty_id); +} + +bool UrmaContext::findJettyOwner(uint32_t jetty_id, UrmaEndpoint** endpoint, + int* slot) { + RWSpinlock::ReadGuard guard(jetty_owner_lock_); + auto it = jetty_owner_map_.find(jetty_id); + if (it == jetty_owner_map_.end()) return false; + if (endpoint) *endpoint = it->second.endpoint; + if (slot) *slot = it->second.slot; + return true; +} + +void UrmaContext::addDrainingEndpoint(UrmaEndpoint* endpoint) { + RWSpinlock::WriteGuard guard(jetty_owner_lock_); + draining_endpoints_.insert(endpoint); +} + +void UrmaContext::removeDrainingEndpoint(UrmaEndpoint* endpoint) { + RWSpinlock::WriteGuard guard(jetty_owner_lock_); + draining_endpoints_.erase(endpoint); +} + +void UrmaContext::checkJettyDrainTimeouts( + std::unordered_map& jetty_depth_set, + std::vector& failed_slices, + std::vector& deferred_deletes) { + std::vector endpoints; + { + RWSpinlock::ReadGuard guard(jetty_owner_lock_); + endpoints.assign(draining_endpoints_.begin(), + draining_endpoints_.end()); + } + for (auto* endpoint : endpoints) { + if (endpoint) + endpoint->checkDrainTimeout(jetty_depth_set, failed_slices, + deferred_deletes); + } +} + +namespace { +void deferEndpointDelete(UrmaEndpoint* endpoint, + std::vector& deferred_deletes) { + for (auto* ep : deferred_deletes) { + if (ep == endpoint) return; + } + deferred_deletes.push_back(endpoint); +} +} // namespace + +int UrmaContext::poll(int num_entries, + std::vector& failed_slices, std::unordered_map& jetty_depth_set, + std::vector& deferred_deletes, int jfc_index) { - num_failed = 0; urma_cr_t cr[num_entries]; int nr_poll = urma_poll_jfc(jfc_list_[jfc_index].native, num_entries, cr); if (nr_poll < 0) { @@ -530,53 +588,43 @@ int UrmaContext::poll(int num_entries, Transport::Slice** failed_slices, << device_name_; return ERR_CONTEXT; } + int wr_completions = 0; for (int i = 0; i < nr_poll; ++i) { + // Fake fence CQE: user_ctx is invalid; match by local_id only. + if (cr[i].status == URMA_CR_WR_FLUSH_ERR_DONE) { + UrmaEndpoint* endpoint = nullptr; + int slot = -1; + if (findJettyOwner(cr[i].local_id, &endpoint, &slot) && endpoint) { + endpoint->onFlushDone(slot, jetty_depth_set, failed_slices, + deferred_deletes, wr_completions); + } else { + LOG(WARNING) << "FLUSH_ERR_DONE for unknown jetty local_id=" + << cr[i].local_id << " on " << device_name_; + } + continue; + } + auto slice = (Transport::Slice*)cr[i].user_ctx; if (!slice) { continue; } - - // All deref of `slice` (including the jetty_depth aggregation below) - // MUST happen before markSuccess(): once that publishes completion, - // the submitting thread may recycle the slice immediately. - auto* depth = slice->ub.jetty_depth; - auto it = jetty_depth_set.find(depth); - if (it != jetty_depth_set.end()) - it->second++; - else - jetty_depth_set[depth] = 1; - - if (cr[i].status == URMA_CR_SUCCESS) { - // Safe to publish here — we are done with this slice and do not - // return it to the caller, so no one else will deref it. - slice->markSuccess(); + auto* endpoint = static_cast(slice->ub.endpoint); + if (!endpoint) { + // Keep the legacy accounting: the WR resolved even though we no + // longer know its endpoint. + ++wr_completions; continue; } - - if (cr[i].status != URMA_CR_WR_FLUSH_ERR || - show_work_request_flushed_error_) - LOG(ERROR) << "Worker: Process failed for slice (opcode: " - << slice->opcode - << ", source_addr: " << slice->source_addr - << ", length: " << slice->length - << ", dest_addr: " << (void*)slice->ub.dest_addr - << ", local_nic: " << deviceName() - << ", peer_nic: " << slice->peer_nic_path - << ", dest_seg_tokenid: " - << static_cast(slice->ub.r_seg) - ->seg.token_id - << ", retry_cnt: " << slice->ub.retry_cnt - << "): " << cr[i].status << ", jfc idx : " << jfc_index - << ", comp_events_acked: " - << jfc_list_[jfc_index].native->comp_events_acked << " " - << jfc_list_[jfc_index].native->async_events_acked; - - // Failed: hand the slice back so the caller can decide retry vs - // final markFailed(). Slice is NOT published, so the caller may - // safely deref it. - failed_slices[num_failed++] = slice; - } - return nr_poll; + if (endpoint->processWrCompletion(cr[i], jetty_depth_set, failed_slices, + deferred_deletes, jfc_index, + /*allow_error_trigger=*/true)) { + ++wr_completions; + } + } + checkJettyDrainTimeouts(jetty_depth_set, failed_slices, deferred_deletes); + // Exclude FLUSH_ERR_DONE and dropped stale completions from outstanding + // accounting; the rebuild path already accounted for the stale ones. + return wr_completions; } volatile int* UrmaContext::outstandingCount(int jfc_index) { @@ -626,6 +674,13 @@ int UrmaEndpoint::construct(GlobalConfig& config) { } jetty_list_.resize(num_jetty_list); + jetty_state_.assign(num_jetty_list, ACTIVE); + jetty_epoch_.assign(num_jetty_list, 1); + peer_jetty_id_.assign(num_jetty_list, 0); + jetty_id_map_.clear(); + peer_eid_.clear(); + drain_start_ns_ = 0; + draining_slot_ = -1; auto* jfc = context_->jfc(); jfc_outstanding_ = (volatile int*)jfc->jfc_cfg.user_ctx; @@ -658,10 +713,26 @@ int UrmaEndpoint::construct(GlobalConfig& config) { jetty_list_[i] = urma_create_jetty(context_->urma_context_, &attr); if (!jetty_list_[i]) { PLOG(ERROR) << "Failed to create jetty"; + for (size_t j = 0; j < i; ++j) { + if (!jetty_list_[j]) continue; + context_->unregisterJettyOwner(jetty_list_[j]->jetty_id.id); + urma_delete_jetty(jetty_list_[j]); + jetty_list_[j] = nullptr; + } + jetty_list_.clear(); + jetty_state_.clear(); + jetty_epoch_.clear(); + peer_jetty_id_.clear(); + jetty_id_map_.clear(); + delete[] wr_depth_list_; + wr_depth_list_ = nullptr; return ERR_ENDPOINT; } - LOG(INFO) << "Create jetty success, jetty id = " - << jetty_list_[i]->jetty_id.id << " ,jetty jfc id = " + uint32_t jetty_id = jetty_list_[i]->jetty_id.id; + jetty_id_map_[jetty_id] = static_cast(i); + context_->registerJettyOwner(jetty_id, this, static_cast(i)); + LOG(INFO) << "Create jetty success, jetty id = " << jetty_id + << " ,jetty jfc id = " << jetty_list_[i]->jetty_cfg.jfs_cfg.jfc->jfc_id.id << " : " << jfc->jfc_id.id; } @@ -672,19 +743,25 @@ int UrmaEndpoint::construct(GlobalConfig& config) { int UrmaEndpoint::deconstruct() { int ret = 0; + context_->removeDrainingEndpoint(this); for (size_t i = 0; i < jetty_list_.size(); ++i) { auto imported_it = imported_jetty_map_.find(jetty_list_[i]); auto imported_jetty = (imported_it != imported_jetty_map_.end()) ? imported_it->second : nullptr; - ret = urma_unbind_jetty(jetty_list_[i]); - if (ret) PLOG(ERROR) << "Failed to unbind jetty"; + if (jetty_list_[i]) { + context_->unregisterJettyOwner(jetty_list_[i]->jetty_id.id); + ret = urma_unbind_jetty(jetty_list_[i]); + if (ret) PLOG(ERROR) << "Failed to unbind jetty"; + } if (imported_jetty != nullptr) { ret = urma_unimport_jetty(imported_jetty); if (ret) PLOG(ERROR) << "Failed to unimport jetty"; } - ret = urma_delete_jetty(jetty_list_[i]); - if (ret) PLOG(ERROR) << "Failed to delete jetty"; + if (jetty_list_[i]) { + ret = urma_delete_jetty(jetty_list_[i]); + if (ret) PLOG(ERROR) << "Failed to delete jetty"; + } // After destroying QP, the wr_depth_list_ won't change bool displayed = false; if (wr_depth_list_[i] != 0) { @@ -698,7 +775,14 @@ int UrmaEndpoint::deconstruct() { } } jetty_list_.clear(); + jetty_state_.clear(); + jetty_id_map_.clear(); + peer_jetty_id_.clear(); + peer_eid_.clear(); + draining_slot_ = -1; + drain_start_ns_ = 0; delete[] wr_depth_list_; + wr_depth_list_ = nullptr; imported_jetty_map_.clear(); return 0; } @@ -791,13 +875,28 @@ int UrmaEndpoint::setupConnectionsByActive() { void UrmaEndpoint::disconnectUnlocked() { urma_jetty_attr_t attr; memset(&attr, 0, sizeof(attr)); + attr.mask = JETTY_STATE; attr.state = URMA_JETTY_STATE_RESET; + context_->removeDrainingEndpoint(this); + draining_slot_ = -1; + drain_start_ns_ = 0; for (size_t i = 0; i < jetty_list_.size(); ++i) { - int ret = urma_modify_jetty(jetty_list_[i], &attr); - if (ret) PLOG(ERROR) << "Failed to modify jetty to RESET"; + if (!jetty_list_[i]) continue; + // Only jettys that entered ERROR (modify already called) are + // flushable; PENDING_DRAIN ones have not been modified yet and go + // through the normal RESET path below. + if (jetty_state_[i] == DRAINING || jetty_state_[i] == REBUILDING) { + urma_cr_t flush_crs[64]; + while (true) { + int flushed = urma_flush_jetty(jetty_list_[i], 64, flush_crs); + if (flushed <= 0) break; + } + } + int reset_ret = urma_modify_jetty(jetty_list_[i], &attr); + if (reset_ret) PLOG(ERROR) << "Failed to modify jetty to RESET"; auto imported_jetty = imported_jetty_map_[jetty_list_[i]]; - ret = urma_unbind_jetty(jetty_list_[i]); + int ret = urma_unbind_jetty(jetty_list_[i]); if (ret) PLOG(ERROR) << "Failed to unbind jetty"; ret = urma_unimport_jetty(imported_jetty); if (ret) PLOG(ERROR) << "Failed to unimport jetty"; @@ -812,7 +911,11 @@ void UrmaEndpoint::disconnectUnlocked() { __sync_fetch_and_sub(jfc_outstanding_, wr_depth_list_[i]); wr_depth_list_[i] = 0; } + if (!reset_ret) jetty_state_[i] = ACTIVE; } + imported_jetty_map_.clear(); + peer_jetty_id_.assign(jetty_list_.size(), 0); + peer_eid_.clear(); status_.store(UNCONNECTED, std::memory_order_release); } @@ -873,7 +976,8 @@ int UrmaEndpoint::submitPostSend( std::vector& failed_slice_list) { RWSpinlock::WriteGuard guard(lock_); if (!active_) return 0; - int jetty_index = SimpleRandom::Get().next(jetty_list_.size()); + int jetty_index = selectActiveJettyUnlocked(); + if (jetty_index < 0) return 0; int wr_count = std::min(max_wr_depth_ - wr_depth_list_[jetty_index], (int)slice_list.size()); wr_count = @@ -924,13 +1028,12 @@ int UrmaEndpoint::submitPostSend( slice->ts = getCurrentTimeInNano(); slice->status = Transport::Slice::POSTED; slice->ub.jetty_depth = &wr_depth_list_[jetty_index]; + slice->ub.jetty_epoch = jetty_epoch_[jetty_index]; // Set endpoint pointer for each slice before submitting slice->ub.endpoint = this; } __sync_fetch_and_add(&wr_depth_list_[jetty_index], wr_count); __sync_fetch_and_add(jfc_outstanding_, wr_count); - if (jetty_list_[jetty_index]->remote_jetty == NULL) { - } int rc = urma_post_jetty_send_wr(jetty_list_[jetty_index], wr_list, &bad_wr); if (rc) { @@ -968,6 +1071,7 @@ int UrmaEndpoint::doSetupConnection(const std::string& peer_eid, return ERR_INVALID_ARGUMENT; } + peer_eid_ = peer_eid; for (int jetty_index = 0; jetty_index < (int)jetty_list_.size(); ++jetty_index) { int ret = doSetupConnection( @@ -1011,12 +1115,380 @@ int UrmaEndpoint::doSetupConnection(int jetty_index, return ERR_ENDPOINT; } imported_jetty_map_[jetty] = imported_jetty; + peer_jetty_id_[jetty_index] = peer_jetty_num; + jetty_state_[jetty_index] = ACTIVE; LOG(INFO) << "Bind jetty success, local jetty id:" << jetty->jetty_id.id << ", remote jetty id:" << peer_jetty_num; return 0; } +int UrmaEndpoint::findSlotByDepth(volatile int* depth) const { + if (!depth || !wr_depth_list_) return -1; + for (size_t i = 0; i < jetty_list_.size(); ++i) { + if (&wr_depth_list_[i] == depth) return static_cast(i); + } + return -1; +} + +bool UrmaEndpoint::hasNonActiveJettyUnlocked() const { + for (auto state : jetty_state_) { + if (state != ACTIVE) return true; + } + return false; +} + +int UrmaEndpoint::selectActiveJettyUnlocked() { + if (jetty_list_.empty()) return -1; + const int n = static_cast(jetty_list_.size()); + int start = SimpleRandom::Get().next(n); + for (int i = 0; i < n; ++i) { + int idx = (start + i) % n; + if (jetty_state_[idx] == ACTIVE) return idx; + } + return -1; +} + +int UrmaEndpoint::startDrainUnlocked(int slot) { + urma_jetty_attr_t attr{}; + attr.mask = JETTY_STATE; + attr.state = URMA_JETTY_STATE_ERROR; + int ret = urma_modify_jetty(jetty_list_[slot], &attr); + if (ret) { + PLOG(ERROR) << "Failed to modify jetty to ERROR, slot=" << slot + << " jetty_id=" << jetty_list_[slot]->jetty_id.id; + context_->removeDrainingEndpoint(this); + draining_slot_ = -1; + drain_start_ns_ = 0; + return ERR_ENDPOINT; + } + jetty_state_[slot] = DRAINING; + draining_slot_ = slot; + drain_start_ns_ = getCurrentTimeInNano(); + context_->addDrainingEndpoint(this); + LOG(WARNING) << "Jetty ACK timeout: start drain slot=" << slot + << " jetty_id=" << jetty_list_[slot]->jetty_id.id << " on " + << toString(); + return 0; +} + +void UrmaEndpoint::onJettyError(int slot, + std::vector& deferred_deletes) { + bool delete_ep = false; + { + RWSpinlock::WriteGuard guard(lock_); + if (slot < 0 || slot >= static_cast(jetty_list_.size())) return; + if (jetty_state_[slot] != ACTIVE) return; // idempotent + if (!jetty_list_[slot]) return; + // Serial rebuild: at most one draining/rebuilding jetty per endpoint. + // Queue this slot as PENDING_DRAIN so that submitPostSend stops + // selecting it; the rebuild tail starts its drain once serialized + // state is free again. + if (hasNonActiveJettyUnlocked()) { + jetty_state_[slot] = PENDING_DRAIN; + LOG(WARNING) << "Queue jetty drain for slot " << slot + << " jetty_id=" << jetty_list_[slot]->jetty_id.id + << ": another jetty is already draining/rebuilding on " + << toString(); + return; + } + if (startDrainUnlocked(slot)) delete_ep = true; + } + if (delete_ep) { + LOG(ERROR) << "Jetty rebuild fallback to deleteEndpoint: " + << "modify_jetty(ERROR) failed on " << toString(); + deferEndpointDelete(this, deferred_deletes); + } +} + +void UrmaEndpoint::onFlushDone( + int slot, std::unordered_map& jetty_depth_set, + std::vector& failed_slices, + std::vector& deferred_deletes, int& resolved_wr_count) { + bool delete_ep = false; + { + RWSpinlock::WriteGuard guard(lock_); + if (slot < 0 || slot >= static_cast(jetty_list_.size())) return; + if (jetty_state_[slot] != DRAINING) return; + jetty_state_[slot] = REBUILDING; + LOG(INFO) << "Jetty flush-done: rebuild slot=" << slot << " on " + << toString(); + if (rebuildJettyUnlocked(slot, jetty_depth_set, failed_slices, + deferred_deletes, resolved_wr_count)) { + context_->removeDrainingEndpoint(this); + draining_slot_ = -1; + drain_start_ns_ = 0; + delete_ep = true; + } + } + if (delete_ep) { + LOG(ERROR) << "Jetty rebuild fallback to deleteEndpoint: " + << "rebuildJetty failed on " << toString(); + deferEndpointDelete(this, deferred_deletes); + } +} + +void UrmaEndpoint::checkDrainTimeout( + std::unordered_map& /*jetty_depth_set*/, + std::vector& /*failed_slices*/, + std::vector& deferred_deletes) { + bool delete_ep = false; + { + RWSpinlock::WriteGuard guard(lock_); + if (draining_slot_ < 0) return; + int slot = draining_slot_; + if (slot >= static_cast(jetty_state_.size()) || + jetty_state_[slot] != DRAINING) { + return; + } + uint64_t now = getCurrentTimeInNano(); + if (now - drain_start_ns_ < kJettyDrainTimeoutNs) return; + LOG(ERROR) << "Jetty drain timed out after " + << ((now - drain_start_ns_) / 1000000ull) + << "ms, slot=" << slot << " on " << toString(); + context_->removeDrainingEndpoint(this); + draining_slot_ = -1; + drain_start_ns_ = 0; + delete_ep = true; + } + if (delete_ep) { + LOG(ERROR) << "Jetty rebuild fallback to deleteEndpoint: " + << "flush-done timeout on " << toString(); + deferEndpointDelete(this, deferred_deletes); + } +} + +bool UrmaEndpoint::processWrCompletion( + urma_cr_t& cr, std::unordered_map& jetty_depth_set, + std::vector& failed_slices, + std::vector& deferred_deletes, int jfc_index, + bool allow_error_trigger) { + auto slice = reinterpret_cast(cr.user_ctx); + if (!slice) return false; + + // All deref of `slice` (including jetty_depth aggregation below) MUST + // happen before markSuccess(): once that publishes completion, the + // submitting thread may recycle the slice immediately. + auto* depth = slice->ub.jetty_depth; + int slot = findSlotByDepth(depth); + if (slot >= 0 && slot < static_cast(jetty_epoch_.size()) && + slice->ub.jetty_epoch != jetty_epoch_[slot]) { + LOG(WARNING) << "Dropping stale jetty completion for slot " << slot + << " on " << toString(); + return false; + } + + auto it = jetty_depth_set.find(depth); + if (it != jetty_depth_set.end()) + it->second++; + else + jetty_depth_set[depth] = 1; + + if (cr.status == URMA_CR_SUCCESS) { + slice->markSuccess(); + return true; + } + + if (cr.status == URMA_CR_ACK_TIMEOUT_ERR && allow_error_trigger) { + if (slot < 0) { + UrmaEndpoint* mapped = nullptr; + int mapped_slot = -1; + if (context_->findJettyOwner(cr.local_id, &mapped, &mapped_slot) && + mapped == this) { + slot = mapped_slot; + } + } + if (slot >= 0) onJettyError(slot, deferred_deletes); + } + + if (cr.status != URMA_CR_WR_FLUSH_ERR || + context_->traceWorkRequestFlushedErrors()) + LOG(ERROR) + << "Worker: Process failed for slice (opcode: " << slice->opcode + << ", source_addr: " << slice->source_addr + << ", length: " << slice->length + << ", dest_addr: " << (void*)slice->ub.dest_addr + << ", local_nic: " << context_->deviceName() + << ", peer_nic: " << slice->peer_nic_path << ", dest_seg_tokenid: " + << static_cast(slice->ub.r_seg)->seg.token_id + << ", retry_cnt: " << slice->ub.retry_cnt << "): " << cr.status + << ", jfc idx : " << jfc_index; + + failed_slices.push_back(slice); + return true; +} + +int UrmaEndpoint::rebuildJettyUnlocked( + int slot, std::unordered_map& jetty_depth_set, + std::vector& failed_slices, + std::vector& deferred_deletes, int& resolved_wr_count) { + auto* old_jetty = jetty_list_[slot]; + if (!old_jetty) return ERR_ENDPOINT; + const uint32_t old_id = old_jetty->jetty_id.id; + const uint32_t peer_id = peer_jetty_id_[slot]; + const uint64_t started_ns = drain_start_ns_; + urma_jfc_t* reuse_jfc = old_jetty->jetty_cfg.jfs_cfg.jfc; + urma_jfr_t* reuse_jfr = old_jetty->jetty_cfg.shared.jfr; + + // 1) Flush residual WRs and deliver their completions. + urma_cr_t flush_crs[64]; + while (true) { + int flushed = urma_flush_jetty(old_jetty, 64, flush_crs); + if (flushed < 0) { + PLOG(ERROR) << "urma_flush_jetty failed, slot=" << slot; + return ERR_ENDPOINT; + } + if (flushed == 0) break; + for (int j = 0; j < flushed; ++j) { + if (flush_crs[j].status == URMA_CR_WR_FLUSH_ERR_DONE) continue; + if (processWrCompletion(flush_crs[j], jetty_depth_set, + failed_slices, deferred_deletes, -1, + /*allow_error_trigger=*/false)) { + ++resolved_wr_count; + } + } + } + + // 2) Unbind / unimport old peer view. + auto imported_it = imported_jetty_map_.find(old_jetty); + urma_target_jetty_t* old_imported = + (imported_it != imported_jetty_map_.end()) ? imported_it->second + : nullptr; + int ret = urma_unbind_jetty(old_jetty); + if (ret) PLOG(ERROR) << "Failed to unbind jetty before rebuild"; + if (old_imported) { + ret = urma_unimport_jetty(old_imported); + if (ret) PLOG(ERROR) << "Failed to unimport jetty before rebuild"; + imported_jetty_map_.erase(imported_it); + } + + // 3) Delete old jetty and clear outstanding depth for this slot. + context_->unregisterJettyOwner(old_id); + jetty_id_map_.erase(old_id); + ret = urma_delete_jetty(old_jetty); + if (ret) { + PLOG(ERROR) << "Failed to delete jetty during rebuild"; + jetty_list_[slot] = nullptr; + return ERR_ENDPOINT; + } + jetty_list_[slot] = nullptr; + // No explicit depth/outstanding adjustment here: every WR of the old + // jetty completes exactly once, either via JFC poll (WR_FLUSH_ERR) before + // the flush-done fence or via the urma_flush_jetty loop above, and each + // completion was already accounted through processWrCompletion (deferred + // via jetty_depth_set / resolved_wr_count). Adjusting wr_depth_list_[slot] + // here would double-count. The epoch bump makes any late duplicate + // completion for the old generation a no-op instead. + ++jetty_epoch_[slot]; + + if (recreateJettyUnlocked(slot, reuse_jfc, reuse_jfr)) { + return ERR_ENDPOINT; + } + urma_jetty_t* new_jetty = jetty_list_[slot]; + + // 5) Re-import peer and bind locally (no peer protocol). + if (peer_eid_.empty()) { + LOG(ERROR) << "Missing peer eid during jetty rebuild"; + context_->unregisterJettyOwner(new_jetty->jetty_id.id); + jetty_id_map_.erase(new_jetty->jetty_id.id); + urma_delete_jetty(new_jetty); + jetty_list_[slot] = nullptr; + return ERR_ENDPOINT; + } + urma_eid_t eid; + if (!context_->transEidFromString(peer_eid_, eid)) { + LOG(ERROR) << "Invalid peer eid during jetty rebuild: " << peer_eid_; + context_->unregisterJettyOwner(new_jetty->jetty_id.id); + jetty_id_map_.erase(new_jetty->jetty_id.id); + urma_delete_jetty(new_jetty); + jetty_list_[slot] = nullptr; + return ERR_ENDPOINT; + } + urma_rjetty_t rjetty = {}; + rjetty.jetty_id.id = peer_id; + rjetty.jetty_id.eid = eid; + rjetty.trans_mode = URMA_TM_RC; + rjetty.type = URMA_JETTY; + rjetty.tp_type = URMA_CTP; + rjetty.flag.value = 0; + urma_target_jetty_t* imported = + urma_import_jetty(context_->urma_context_, &rjetty, &urma_token); + if (!imported) { + PLOG(ERROR) << "Failed to import peer jetty during rebuild"; + context_->unregisterJettyOwner(new_jetty->jetty_id.id); + jetty_id_map_.erase(new_jetty->jetty_id.id); + urma_delete_jetty(new_jetty); + jetty_list_[slot] = nullptr; + return ERR_ENDPOINT; + } + urma_status_t bind_ret = urma_bind_jetty(new_jetty, imported); + if (bind_ret != URMA_SUCCESS && bind_ret != URMA_EEXIST) { + PLOG(ERROR) << "Failed to bind rebuilt jetty"; + urma_unimport_jetty(imported); + context_->unregisterJettyOwner(new_jetty->jetty_id.id); + jetty_id_map_.erase(new_jetty->jetty_id.id); + urma_delete_jetty(new_jetty); + jetty_list_[slot] = nullptr; + return ERR_ENDPOINT; + } + + imported_jetty_map_[new_jetty] = imported; + jetty_state_[slot] = ACTIVE; + draining_slot_ = -1; + drain_start_ns_ = 0; + context_->removeDrainingEndpoint(this); + + LOG(WARNING) << "Jetty rebuilt successfully slot=" << slot + << " old_id=" << old_id << " new_id=" << new_jetty->jetty_id.id + << " peer_id=" << peer_id << " elapsed_ms=" + << ((getCurrentTimeInNano() - started_ns) / 1000000ull) + << " on " << toString(); + + // Another jetty may have hit ACK timeout while this one was rebuilding; + // start draining the queued slot now (still one at a time). + for (size_t i = 0; i < jetty_state_.size(); ++i) { + if (jetty_state_[i] != PENDING_DRAIN || !jetty_list_[i]) continue; + if (startDrainUnlocked(static_cast(i))) { + LOG(ERROR) << "Failed to start drain for queued slot " << i + << " on " << toString(); + return ERR_ENDPOINT; + } + break; + } + return 0; +} + +int UrmaEndpoint::recreateJettyUnlocked(int slot, urma_jfc_t* reuse_jfc, + urma_jfr_t* reuse_jfr) { + urma_jfs_cfg_t jfs_cfg = { + .depth = 2048, + .trans_mode = URMA_TM_RC, + .priority = 15, + .max_sge = 5, + .rnr_retry = 7, + .err_timeout = 17, + .user_ctx = 0, + }; + urma_jetty_flag_t jetty_flag = {}; + jetty_flag.bs.share_jfr = 1; + urma_jetty_cfg_t attr{}; + attr.flag = jetty_flag; + attr.jfs_cfg = jfs_cfg; + attr.jfs_cfg.jfc = reuse_jfc ? reuse_jfc : context_->jfc(); + attr.shared.jfr = reuse_jfr ? reuse_jfr : context_->jfr(); + urma_jetty_t* new_jetty = urma_create_jetty(context_->urma_context_, &attr); + if (!new_jetty) { + PLOG(ERROR) << "Failed to create jetty during rebuild"; + return ERR_ENDPOINT; + } + + jetty_list_[slot] = new_jetty; + const uint32_t new_id = new_jetty->jetty_id.id; + jetty_id_map_[new_id] = slot; + context_->registerJettyOwner(new_id, this, slot); + return 0; +} + std::shared_ptr UrmaContext::makeEndpoint() { return std::make_shared(this); }