拓扑交换

TopologyCoordinator 是每个作业一个的单例，用于累积 GetMultiSliceTopologyRequest 注册，并发出一个字节稳定的响应，广播给每个被阻塞的工作进程。它位于协调器进程中的 CommunicationBackend +0x1a0（*(void**)(backend + 416)），由 0x1ccad600 处的 CommunicationBackend::InitializeCoordinator(int) 分配。每个工作进程都有一个空槽位：它的 OnTopologyRequestReceived (0x1ccac380) 会走到 coordinator == nullptr 分支并返回 MakeErrorImpl<14>("Topology Coordinator is not ready. Try later.").

对象布局

TopologyCoordinator 继承自 Coordinator<GetMultiSliceTopologyRequest, GetMultiSliceTopologyResponse, AnyInvocable<void(StatusOr<...> const&)>>，并添加自己的切片状态和地址映射。这里会用到两个 vtable：位于 0x21c9bbf0 的基类 Coordinator<GetMultiSliceTopology…> vtable，以及位于 0x21c9baf0 的派生类 TopologyCoordinator vtable（两者都由 nm -C 确认）。构造函数先安装基类 vptr off_21C9BC00（= 0x21c9bbf0 + 0x10），随后用派生类 vptr off_21C9BB00（= 0x21c9baf0 + 0x10）覆盖它。其形态如下（偏移量直接从 0x213b7a40 处的 ctor 和 0x1ccb42a0 处的基类 AddRequest 读出）：

偏移	大小	类型 / 字段
`+0x00`	8	`vptr` → 位于 `0x21c9baf0` 的派生类 vtable
`+0x08`	0x50	`TracedMutex mu_` (`TracedMutex::TracedMutex(this+8, /kind=/9)`)
`+0x58`	1	`uint8_t state_`（1=进行中，2=已完成，3=错误；ctor 将其清零）
`+0x60`	0x28	`StatusOr<GetMultiSliceTopologyResponse> cached_response_`（错误表示位于 `+0x60`，内嵌响应位于 `+0x68`；初始值 = `MakeErrorImpl<14>` "Coordinator in IN_PROGRESS", UNAVAILABLE）
`+0x88`	0x18	`std::vector<AnyInvocable<void(StatusOr<…> const&)>> response_setters_`（begin/end/cap 位于 +0x88/+0x90/+0x98，每个条目 32 B）
`+0xa0`	0x10	`absl::Notification completion_` (`Notify()` at `this+160`)
`+0xb0`	...	周期性报告 alarm 句柄（通过 `thread::AddCancellableAt` 装设，存储在 `this+176`）
`+0xc0`	4	`int32_t num_expected_slices_`（ctor 参数；`CHECK(> 0)`）
`+0xc8`	0x20	`absl::flat_hash_map<std::tuple<int,int>, NetworkAddressMapping> address_map_`（(slice_id, host_id) → address）
`+0xe8`	0x20	`absl::flat_hash_map<int, SliceState>`（slice_id → 每切片累加器）

SliceState 是 this+0xe8 处的 flat_hash_map 值。在 ProcessRequest/CreateResponse 中观察到的槽位布局会存储该切片的 TpuTopology*（读作 *(slot+16)，主机数量位于 *(int*)(topology+108)），以及一个按 host id 建键的嵌套 flat_hash_map<int, …>，用于跟踪每个主机的已见计数（在 IsComplete 中作为 slot_count >> 17 比较）。

基类 Coordinator<> 也会以另一组 Req/Resp 为 BarrierCoordinator 实例化；两者共享相同的控制流（AddRequest、ScheduleStatusReport、 response_setters_ 向量和 Notification）。

构造

位于 0x213b7a40 的 TopologyCoordinator::TopologyCoordinator(int num_slices)（在 .text.unlikely 中，节基址 0x21381900，因为每个作业只调用一次）：

cpp

TopologyCoordinator::TopologyCoordinator(int num_slices) {
  *(void**)this = off_21C9BC00;                  // base vptr (0x21c9bbf0+0x10)
  TracedMutex::TracedMutex(this + 8, /*kind=*/9);

  *(uint8_t*)(this + 88) = 0;                     // state_ = 0
  // Sticky StatusOr seeded with UNAVAILABLE:
  *(void**)(this + 96) = MakeErrorImpl<14>(       // code 14 = UNAVAILABLE
      "Coordinator in IN_PROGRESS", 26, /*line=*/46,
      "platforms/xla/megascale/runtime/communication/topology_coordinator.h");

  *(void**)(this + 184) = 0;                       // response_setters_ tail
  *(__m256*)(this + 0x90) = 0;                      // vector + alarm slots
  *(uint8_t*)(this + 176) = 0;

  *(void**)this = off_21C9BB00;                     // derived vptr (0x21c9baf0+0x10)
  *(int*)(this + 192) = num_slices;                 // num_expected_slices_
  *(__m128*)(this + 0xc8) = 0;                       // address_map_ control
  *(__m128*)(this + 0xe8) = 0;                       // slice_state_ control

  CHECK(num_slices > 0) << "num_expected_slices_ > 0";   // FATAL at line 55
  LOG(INFO) << "Megascale Topology Coordinator started for "
            << num_slices << " slices";                  // line 56, unconditional
}
```text

info 日志前缀 `"Megascale Topology Coordinator started for "` 位于
rodata `0xa1e728a`；后缀 `" slices"` 是在 `int` 之后追加的独立字面量。
操作员正是通过这一行确认哪个进程是协调器。
`num_expected_slices_ > 0` 保护是一个 `CHECK`（文件第 55 行 fatal），
而不是包围该日志的条件；一旦到达 ctor，文件第 56 行的日志就会无条件发出。

## 泛型 `Coordinator<>::AddRequest`

共享的基类模板处理整个 rendezvous 协议。
Barrier 实例（`0x1ccb42a0`，大小 `0x7d4`）和 Topology 实例
（`0x1cf559c0`，大小同为 `0x7d4`）具有逐字节相同的实例化模式，
因此 Barrier 的反编译可以套读到 Topology 上。
三个派生类 hook 通过固定 vtable 槽位到达，已由 Barrier `AddRequest`
的 `objdump` 确认：
`call *0x20(%rax)` = `ProcessRequest`, `call *0x28(%rax)` =
`IsComplete`, `call *0x30(%rax)` = `CreateResponse`。伪代码：

```cpp
void Coordinator<Req, Resp, Callback>::AddRequest(
    Req const& req, Callback cb) {

  // VLOG(5) site: "AddRequest: " << ShortFormat(req).
  absl::Time start = absl::Now();
  TracedReleasableMutexLock lock(&this->mu_);

  // State 3 = previously failed; serve sticky error and return
  // (CHECK(response_setters_.empty()) on this path).
  if (this->state_ == 3) {
    cb(this->cached_response_);    // StatusOr holding the error at this+96
    return;
  }

  // Register the request into per-coordinator state via vtable +0x20
  // (= ProcessRequest). Returns a Status: OK accepts, else rejects.
  Status st = vtable_[+0x20].ProcessRequest(this, req);

  // State 2 = already complete; serve the cached response.
  if (this->state_ == 2) {
    cb(this->cached_response_);
    return;
  }

  // State 0/1 = still gathering: queue the response setter.
  this->response_setters_.push_back(std::move(cb));   // 32 B/entry

  // Quorum check via vtable +0x28 (= IsComplete).
  if (vtable_[+0x28].IsComplete(this)) {
    this->state_ = 2;
    this->cached_response_ = vtable_[+0x30].CreateResponse(this);
    for (auto& s : this->response_setters_) s(this->cached_response_);
    this->completion_.Notify();           // absl::Notification at this+160
  } else {
    this->state_ = 1;
    this->ScheduleStatusReport();          // arms periodic ReportStatus
  }
}

Barrier 实例内部的反汇编交叉引用：

0x1ccb4347：state-3 快速路径上的 vtable +0x18 调用（TracedReleasableMutexLock / response-setter 辅助）。
0x1ccb4365：vtable +0x20 调用（ProcessRequest）。
0x1ccb4499 和 0x1ccb4669：vtable +0x28 调用（IsComplete，求值两次）。
0x1ccb44b6：vtable +0x30 调用（CreateResponse）。
当 response_setters_ 向量达到容量上限时， __emplace_back_slow_path 会扩展它（32 字节的 AnyInvocable 条目）。
未完成分支：state_ = 1，随后通过 thread::DefaultFiberExecutor + thread::AddCancellableAt + LocalInvoker<...ScheduleStatusReport()...> 装设周期性 alarm （句柄位于 this+176）。
absl::Notification::Notify(this+160) 在 setter 扇出完成后触发。

`TopologyCoordinator::ProcessRequest`

位于 0x1cf524c0 的派生类 ProcessRequest 大小为 0x1dab （7 595）字节。它返回 absl::Status（不是 bool）：干净接受时返回 OK （1/inline-OK 表示），而每个一致性失败都会返回一个不同的错误来拒绝注册。基类 AddRequest 随后通过单独的 IsComplete vtable 槽位做 quorum 决策。形态如下：

cpp

Status TopologyCoordinator::ProcessRequest(
    GetMultiSliceTopologyRequest const& req) {

  // 1. slice_id bounds: must be in [0, num_expected_slices_).
  int slice_id = req.network_address_mapping().slice_id();
  if (slice_id < 0 || slice_id >= num_expected_slices_) {
    return MakeErrorImpl<3>(  // INVALID_ARGUMENT, file line 183
        Substitute("SliceId out of bounds. Expected num slices: $0. "
                   "Request: $1", num_expected_slices_, req));
  }

  // 2. Look up or create the SliceState entry in slice_state_ (this+0xe8).
  SliceState& slot = slice_state_[slice_id];

  // 3. If a topology was already registered for this slice, compare
  //    it (proto2 MessageDifferencer, EQUIVALENT). On mismatch reject.
  if (slot.has_topology) {
    MessageDifferencer diff;
    diff.set_message_field_comparison(MessageDifferencer::EQUIVALENT);
    std::string text_diff;
    diff.ReportDifferencesToString(&text_diff);
    if (!diff.Compare(slot.topology_args.ToProto(),
                      req.tpu_topology_args())) {
      return MakeErrorImpl<3>(   // INVALID_ARGUMENT, file line 202
        Substitute("Received topology that differs from previously "
          "registered topology at same sliceID. SliceID: $0 "
          "Previous HostId: $1 New HostId: $2 Addresses: $3 Diff: $4",
          ...));
    }
  } else {
    // First registration for this slice: distill+store TpuTopology,
    // record num_expected_slices_-side state.
    slot.topology = TpuTopologySerdes::Construct(req.tpu_topology_args());
    slot.has_topology = true;
  }

  // 4. host_id bounds: must be < topology.host_count()
  //    (= *(int*)(topology + 108)).
  int host_id = req.network_address_mapping().host_id();
  if (host_id < 0 || host_id >= slot.topology->host_count()) {
    return MakeErrorImpl<3>(   // INVALID_ARGUMENT, file lines 235 / 250
        Substitute("HostId out of bounds. hostId: $0. Request: $1",
                   host_id, req));
  }

  // 5. Per-(slice,host) address mapping in address_map_ (this+0xc8).
  //    If already present and the new mapping differs, reject.
  auto& cell = address_map_[{slice_id, host_id}];
  if (cell.present &&
      !MessageDifferencer::Equivalent(cell, req.network_address_mapping())) {
    return MakeErrorImpl<3>(   // INVALID_ARGUMENT, file line 216
      Substitute("Received host address mapping that differs from "
        "previous mapping SliceID: $0 HostId: $1 Prev Address: $2 "
        "New Addresses: $3", ...));
  }
  // Incarnation-id drift on the same (slice,host) is likewise rejected.
  if (incarnation_changed) {
    return MakeErrorImpl<3>(   // INVALID_ARGUMENT, file line 226
      Substitute("Received incarnation ID that is different from "
        "previous incarnation ID. SliceID: $0 HostId: $1 "
        "Prev IncarnationId: $2 New IncarnationId: $3", ...));
  }
  cell.CopyFrom(req.network_address_mapping());

  return absl::OkStatus();
}
```text

> 每个一致性检查都是硬性*拒绝*，会返回非 OK 的 `Status`
> （全部 `MakeErrorImpl<3>` = INVALID_ARGUMENT）。它们不是
> “log and continue” 警告，也不存在单独的 `LogUniqueIds` pass：
> topology、host、address-mapping 和 incarnation 比较全都内联在
> 这一个函数中。

两条 `MessageDifferencer::Compare` 调用链（topology args 和每主机
`NetworkAddressMapping`）占据主要开销；slice_state 和 address-map 的
SwissMap 插入（`find_or_prepare_insert_large` /
`PrepareInsertSmallNonSoo`）构成其余大部分开销。

## `TopologyCoordinator::IsComplete`

在 `0x1cf543a0` 处反编译（仅 `0xb6` 字节）：

```cpp
bool TopologyCoordinator::IsComplete() const {
  // slice_state_.size() (encoded as size_field >> 17) must reach
  // num_expected_slices_ (this+0xc0, read as *(int*)(this+192)).
  if (slice_state_.size() < num_expected_slices_) return false;

  // Walk SwissMap control bytes; for each occupied slot, check that
  // the per-host seen count has reached the slice's host_count
  // (= *(int*)(topology + 108)).
  for (auto const& [slice_id, slot] : slice_state_) {
    if (slot.seen_count < slot.topology->host_count()) return false;
  }
  return true;
}

完成检查是 O(num_slices)，每次 AddRequest 会通过基类模板的 vtable +0x28 槽位调用一次（不是从 ProcessRequest 内部调用）。

`TopologyCoordinator::CreateResponse`

在 0x1cf54460 处反编译（0x9e6 / 2 534 字节）。它构造一个 MultiSliceTopologyInfo，将其序列化进 Cord，并把该 Cord 存储到 GetMultiSliceTopologyResponse 上：

cpp

GetMultiSliceTopologyResponse TopologyCoordinator::CreateResponse() {
  MultiSliceTopologyInfo info;
  GetMultiSliceTopologyResponse response;

  // Walk slice_state_ (this+0xe8). For each slice add a SliceInfo and,
  // for every host in [0, topology.host_count()), append the host's
  // NetworkAddressMapping (looked up in address_map_ at this+0xc8;
  // a miss is FATAL: "Missing addresses for SliceID: ...").
  for (auto const& [slice_id, slot] : slice_state_) {
    SliceInfo* si = info.add_slices();
    si->set_slice_id(slice_id);
    si->mutable_topology_args()->CopyFrom(slot.topology.ToProto());
    for (int h = 0; h < slot.topology->host_count(); ++h) {
      auto it = address_map_.find({slice_id, h});      // CHECK != end()
      si->add_host_addresses()->CopyFrom(it->second);
    }
  }

  // Byte-stable ordering: SliceInfo* by slice_id, host-address
  // NetworkAddressMapping* by (slice_id, host_id).
  std::__introsort(slices.begin(),    slices.end(),    $_0);   // 0x1cf56520
  std::__introsort(host_addrs.begin(), host_addrs.end(), $_1); // 0x1cf57360

  VLOG(3) << "Topology Coordinator response: " << info;  // line 311
  LOG(INFO) << "MegaScale Topology Discovery completed."; // line 312

  absl::Cord cord;
  CHECK(info.SerializeToString(&cord));                  // line 315
  response.set_serialized_topology_info(std::move(cord)); // response+24, field 1
  return response;
}
```text

> 不存在 `shared_seed` / `NewGlobalID()` 字段，
> 也不存在单独的 `endpoints` repeated 字段：二者都没有出现在二进制中。
> 载荷是单个 `MultiSliceTopologyInfo`，被序列化进存储于
> `response + 24` 的 `Cord`。`address_map_` 中缺失
> `(slice_id, host_id)` 条目是 fatal `CHECK`
> ("`address_mapping != address_map_.end()`",
> "Missing addresses for SliceID: " at rodata `0xa28aff5`).

两个 `std::__u::__introsort` 实例是字节稳定的排序器：
位于 `0x1cf56520`、作用于 `SliceInfo**` 的 `$_0`，以及位于
`0x1cf57360`、作用于 `NetworkAddressMapping**` 的 `$_1`
（两者均由 `nm -C -S` 确认）。它们的比较器会读取对应元素的
`slice_id`（对 `$_1` 还会读取 `host_id`）。

## 一致性检查错误字符串

下面三个 rodata 字符串是在 `ProcessRequest` (`0x1cf524c0`)
*内部*发出的 `MakeErrorImpl<3>`（INVALID_ARGUMENT）消息：
它们会作为错误返回，而不是作为信息性警告记录：

- `Received topology that differs from previously registered
  topology at same sliceID. SliceID: $0 Previous HostId: $1 New
  HostId: $2 Addresses: $3 Diff: $4`（`0x9b27486`，文件第 202 行）。
- `Received host address mapping that differs from previous mapping
  SliceID: $0 HostId: $1 Prev Address: $2 New Addresses: $3`
  （`0x9c14204`，文件第 216 行）。
- `Received incarnation ID that is different from previous
  incarnation ID. SliceID: $0 HostId: $1 Prev IncarnationId: $2
  New IncarnationId: $3`（`0x9c14456`，文件第 226 行）。

另外两个边界错误共享同一条代码路径：
`SliceId out of bounds. Expected num slices: $0. Request: $1`
（`0x9e6f891`，第 183 行）和 `HostId out of bounds. hostId: $0.
Request: $1`（`0x9e6f8cd`，第 235/250 行）。

## `LogUniqueIds`（工作进程侧）

真正的 `LogUniqueIds` *不是*协调器函数。它是一个 anonymous-namespace
辅助函数
`xla::megascale::runtime::(anonymous namespace)::LogUniqueIds(int
slice_id, int host_id, MultiSliceTopologyAndLocation const& info)`
内联在工作进程侧的 `Communicator::Create` (`0x1cca9aa0`) 中。
它通过记住上次看到的 ids 来去重重复日志；这些 ids 位于三个静态
`int` 槽位：
`0x223717c0/0x223717c4/0x223717c8`
（`...LogUniqueIds(...)::last_ids.{0,1,2}`），并由位于
`0x2257b030` 的静态 `absl::Mutex` `unique_id_mutex` 保护，其
`__cxa_guard` 变量位于 `0x2257b038`（全部由 `nm -C` 确认）。
当传入的 `(slice_id, host_id, …)` 三元组不同于缓存值时，工作进程会发出
"Created communicator. … Slices: … Hosts: …" 行（文件第 570 / 583 行）
并更新槽位；否则保持安静。它不会影响协调器的接受/拒绝决策。

## 响应交付

`CreateResponse` 完成后，基类 `Coordinator<>::AddRequest` 会把结果复制到
`cached_response_` (`+0x60`)，设置 `state_ = 2`，随后在清空
`response_setters_` 向量（`+0x88`，32 字节条目）之前**释放**
`TracedReleasableMutexLock`（`TracedReleasableMutexLock::Release`）：
每个 setter 都会序列化响应，并将其交给对应工作进程的 gRPC 层。
扇出完成后，它会触发位于 `+0xa0` 的
`absl::Notification completion_`（`Notify(this+160)`）。
同一路径还会记录 VLOG(5) "Num responses to send: "、
"Response sent " 和 "Done processing Request received at: … Duration: "
（以及超过阈值后的 "Long running topology coordinator AddRequest" 警告）。

成功日志行 `"MegaScale Topology Discovery completed."`
（rodata `0xa0a3869`）由 `CreateResponse` 自身发出
（`0x1cf54460`，文件第 312 行），就在它序列化
`MultiSliceTopologyInfo` 之前；不是由单独的 `ReportStatus()` 观察器发出。

## 交叉引用

- [Bootstrap 概览](overview.md)：本阶段在 rendezvous 序列中的位置，以及它与 barrier 流共享的 `Coordinator<>` 模板。
- [工作进程注册](worker-registration.md)：`GetMultiSliceTopologyRequest` schema 和服务器侧回调，它们会进入本文档记录的 `ProcessRequest` 路径。
- [收敛](convergence.md)：泛型 `Coordinator<>::AddRequest` 状态机、挂起回调向量，以及本文针对拓扑专门说明的 `absl::Notification`。

拓扑交换 ​

对象布局 ​

构造 ​

TopologyCoordinator::ProcessRequest ​

TopologyCoordinator::CreateResponse ​

拓扑交换

对象布局

构造

`TopologyCoordinator::ProcessRequest`

`TopologyCoordinator::CreateResponse`