|
1 | 1 | # Copyright (c) 2026, NVIDIA CORPORATION & AFFILIATES. All rights reserved. |
2 | 2 |
|
3 | | -"""Owner assignment and P2P packing for MFSDP v2's all-`RowAtomic` layout. |
| 3 | +"""Owner assignment for MFSDP v2's all-`RowAtomic` layout. |
4 | 4 |
|
5 | 5 | Tensor shapes and shard ranges come from the group's existing `GlobalLayout`. |
6 | | -Owner assignments and peer dictionaries use global process ranks. |
| 6 | +Owner assignments use global process ranks. |
7 | 7 | """ |
8 | 8 |
|
9 | 9 | import dataclasses |
|
16 | 16 | from .layout import GlobalLayout, non_leading_numel |
17 | 17 | from .parameter_group import FsdpParameterGroup |
18 | 18 | from .placement import RowAtomic |
19 | | -from .range import Range, intersect_ranges |
| 19 | +from .range import intersect_ranges |
20 | 20 |
|
21 | 21 |
|
22 | 22 | def select_ge_2d_params(param: torch.Tensor) -> bool: |
@@ -148,306 +148,3 @@ def from_group( |
148 | 148 | layout = group.main_weight.layout |
149 | 149 | tensor_to_owner = assign_owner_work(layout, group.mesh, tensor_indices, cost_fn) |
150 | 150 | return cls(mesh=group.mesh, layout=layout, tensor_to_owner=tensor_to_owner) |
151 | | - |
152 | | - |
153 | | -@dataclasses.dataclass |
154 | | -class OwnerGatherPlan: |
155 | | - """Metadata and send buffers for the owner-gather P2P step of a set of parameters. |
156 | | -
|
157 | | - The owner keeps its own shard locally (no self-send), so it only receives from the other |
158 | | - shard-holding ranks. `reconstruct_full` reconstructs each owned tensor by concatenating the |
159 | | - per-rank shards in buffer order. |
160 | | -
|
161 | | - Example: |
162 | | -
|
163 | | - ``` |
164 | | - # We are also using some pseudocode here for brevity. |
165 | | -
|
166 | | - # Assume: |
167 | | - torch.distributed.get_world_size() == 2 |
168 | | - torch.distributed.get_rank() == 0 # We're observing from rank 0. |
169 | | - mesh.mesh.tolist() == [0, 1] # Global process ranks. |
170 | | - param_0: torch.Tensor |
171 | | - param_1: torch.Tensor |
172 | | - # Params are in this order as observed by MFSDP. |
173 | | - model.param_groups == [{"params": [param_0, param_1]}] |
174 | | - param_0.shape == (6, 4) # Global shape. |
175 | | - param_1.shape == (4, 4) # Global shape. |
176 | | - layout.tensor_to_offset == (0, 24) |
177 | | - layout.size == 40 |
178 | | - # The buffer, rather than each parameter separately, is split evenly: |
179 | | - # rank 0 holds [0, 20); rank 1 holds [20, 40). |
180 | | - # Choose rank 1 as both owners for this example. It holds part of param_0 |
181 | | - # and all of param_1; these are explicit assignments, not the default balancer's output. |
182 | | - owner_layout = GroupOwnerLayout(mesh=mesh, layout=layout, tensor_to_owner={0: 1, 1: 1}) |
183 | | -
|
184 | | - param_0.local_shard.shape == (5, 4) # Rank 0 holds param_0[0:5, ...]. |
185 | | - param_1.local_shard.shape == (0, 4) # Rank 0 holds none of param_1. |
186 | | - param_0.local_shard.numel() == 20 |
187 | | - param_1.local_shard.numel() == 0 |
188 | | -
|
189 | | - owner_gather_plan.send_buffers == {1: tensor(20)} # 20 elements from param_0. |
190 | | - # `owner_gather_plan.send_buffers[1]` represents the following flat buffer: |
191 | | - # +--------------------+ |
192 | | - # | param_0 (20 elems) | |
193 | | - # +--------------------+ |
194 | | - # element order: --> |
195 | | -
|
196 | | - # Rank 0 owns nothing. |
197 | | - owner_gather_plan.recv_sizes == {} |
198 | | - owner_gather_plan.own_shards == {} |
199 | | - owner_gather_plan.recv_offsets == {} |
200 | | -
|
201 | | - # --- |
202 | | -
|
203 | | - # Same settings as above, now observing from rank 1 (the owner): |
204 | | - torch.distributed.get_rank() == 1 |
205 | | -
|
206 | | - param_0.local_shard.shape == (1, 4) # Rank 1 holds param_0[5:6, ...]. |
207 | | - param_1.local_shard.shape == (4, 4) # Rank 1 holds all of param_1. |
208 | | -
|
209 | | - send_buffers = {} # Rank 1 owns everything. |
210 | | - recv_sizes = {0: 20} # 20 elements from rank 0. |
211 | | - # Rank 1's own shards, flattened (views): |
212 | | - own_shards = {0: param_0.local_shard.view(-1), 1: param_1.local_shard.view(-1)} |
213 | | - recv_offsets = { |
214 | | - (0, 0): 0, # `param_0` (tensor index 0) from rank 0: offset 0. |
215 | | - # No entry for param_1: it is fully local to rank 1. |
216 | | - } |
217 | | - # Reconstruct param_0 from the received 20 elements followed by its own 4. |
218 | | - # Reconstruct param_1 directly from its own 16 elements, without communication. |
219 | | - ``` |
220 | | -
|
221 | | - Attributes: |
222 | | - send_buffers: Per-destination-owner flat send buffer (this rank's shards for that owner's |
223 | | - params, in tensor-index order). Only owners with non-zero send size appear. |
224 | | - recv_sizes: Per-source-rank element count this rank (as an owner) receives. Only sources |
225 | | - with non-zero total size appear. |
226 | | - own_shards: This rank's flat local shard per owned parameter, keyed by tensor index (used |
227 | | - directly in reconstruction, not communicated). |
228 | | - recv_offsets: Per `(tensor_index, src_rank)`, the flat offset of this param's shard inside |
229 | | - the recv buffer received from `src_rank`. Only contains tuples for which `src_rank` |
230 | | - holds elements. |
231 | | - """ |
232 | | - |
233 | | - layout: GlobalLayout |
234 | | - # Global rank -> buffer offset/length, ordered by buffer offset. |
235 | | - rank_ranges: dict[int, Range] |
236 | | - this_rank: int |
237 | | - send_buffers: dict[int, torch.Tensor] |
238 | | - recv_sizes: dict[int, int] |
239 | | - own_shards: dict[int, torch.Tensor] |
240 | | - recv_offsets: dict[tuple[int, int], int] |
241 | | - |
242 | | - @classmethod |
243 | | - def pack(cls, owner_layout: GroupOwnerLayout, local_shards: dict[int, torch.Tensor]) -> Self: |
244 | | - """Pack this rank's local shards into per-owner P2P send buffers. |
245 | | -
|
246 | | - Args: |
247 | | - owner_layout: The group's owner layout. |
248 | | - local_shards: This rank's local shard per parameter, only required for every parameter |
249 | | - it holds elements of. Shards may be passed in any shape. |
250 | | - """ |
251 | | - mesh = owner_layout.mesh |
252 | | - this_rank = mesh.get_rank() |
253 | | - layout = owner_layout.layout |
254 | | - placements = (RowAtomic(),) * mesh.ndim |
255 | | - tensor_to_owner = owner_layout.tensor_to_owner |
256 | | - tensor_indices = sorted(tensor_to_owner) |
257 | | - owned_indices = [i for i in tensor_indices if tensor_to_owner[i] == this_rank] |
258 | | - |
259 | | - # Query each source once and compute both its receive size and offsets. |
260 | | - rank_ranges: dict[int, Range] = {} |
261 | | - recv_sizes: dict[int, int] = {} |
262 | | - recv_offsets: dict[tuple[int, int], int] = {} |
263 | | - for src in mesh.mesh.flatten().tolist(): |
264 | | - buffer_range = layout.get_rank_range(mesh, placements, src) |
265 | | - rank_ranges[src] = buffer_range |
266 | | - if src == this_rank: |
267 | | - continue |
268 | | - offset = 0 |
269 | | - for tensor_index in owned_indices: |
270 | | - numel = intersect_ranges(layout.get_tensor_range(tensor_index), buffer_range).numel |
271 | | - if numel > 0: |
272 | | - recv_offsets[(tensor_index, src)] = offset |
273 | | - offset += numel |
274 | | - if offset > 0: |
275 | | - recv_sizes[src] = offset |
276 | | - |
277 | | - # Reconstruction concatenates shards in buffer order, including on multi-axis meshes. |
278 | | - rank_ranges = dict(sorted(rank_ranges.items(), key=lambda item: item[1].start)) |
279 | | - local_range = rank_ranges[this_rank] |
280 | | - send_sizes: dict[int, int] = {} |
281 | | - for tensor_index in tensor_indices: |
282 | | - owner = tensor_to_owner[tensor_index] |
283 | | - if owner == this_rank: |
284 | | - continue |
285 | | - numel = intersect_ranges(layout.get_tensor_range(tensor_index), local_range).numel |
286 | | - if numel > 0: |
287 | | - send_sizes[owner] = send_sizes.get(owner, 0) + numel |
288 | | - |
289 | | - send_buffers: dict[int, torch.Tensor] = {} |
290 | | - if send_sizes: |
291 | | - first_shard = next(iter(local_shards.values())) |
292 | | - for owner, size in send_sizes.items(): |
293 | | - send_buffers[owner] = torch.empty( |
294 | | - size, dtype=first_shard.dtype, device=first_shard.device |
295 | | - ) |
296 | | - |
297 | | - # Fill each owner's send buffer in tensor-index order. |
298 | | - cursors: dict[int, int] = {owner: 0 for owner in send_buffers} |
299 | | - own_shards: dict[int, torch.Tensor] = {} |
300 | | - for tensor_index in tensor_indices: |
301 | | - owner = tensor_to_owner[tensor_index] |
302 | | - if owner == this_rank: |
303 | | - own_shards[tensor_index] = local_shards[tensor_index].flatten() |
304 | | - continue |
305 | | - numel = intersect_ranges(layout.get_tensor_range(tensor_index), local_range).numel |
306 | | - if numel == 0: |
307 | | - continue |
308 | | - shard = local_shards[tensor_index] |
309 | | - buf = send_buffers[owner] |
310 | | - buf[cursors[owner] : cursors[owner] + numel].copy_(shard.flatten()) |
311 | | - cursors[owner] += numel |
312 | | - |
313 | | - return cls( |
314 | | - layout=layout, |
315 | | - rank_ranges=rank_ranges, |
316 | | - this_rank=this_rank, |
317 | | - send_buffers=send_buffers, |
318 | | - recv_sizes=recv_sizes, |
319 | | - own_shards=own_shards, |
320 | | - recv_offsets=recv_offsets, |
321 | | - ) |
322 | | - |
323 | | - def reconstruct_full( |
324 | | - self, param_index: int, recv_buffers: dict[int, torch.Tensor] |
325 | | - ) -> torch.Tensor: |
326 | | - """Reconstruct the full flat tensor for one owned parameter from its per-rank shards. |
327 | | -
|
328 | | - Concatenates the per-rank shards in buffer order. Results can be |
329 | | - `view`ed into the desired shape. For a parameter only this rank holds elements of, the own |
330 | | - flat shard is returned directly. |
331 | | -
|
332 | | - Args: |
333 | | - param_index: Tensor index of an owned parameter in the group layout. |
334 | | - recv_buffers: Per-source-rank received buffer (only sources that sent). |
335 | | - """ |
336 | | - tensor_range = self.layout.get_tensor_range(param_index) |
337 | | - shards: list[torch.Tensor] = [] |
338 | | - for src, buffer_range in self.rank_ranges.items(): |
339 | | - if src == self.this_rank: |
340 | | - shards.append(self.own_shards[param_index]) |
341 | | - continue |
342 | | - |
343 | | - numel = intersect_ranges(tensor_range, buffer_range).numel |
344 | | - if numel == 0: |
345 | | - continue |
346 | | - |
347 | | - offset = self.recv_offsets[(param_index, src)] |
348 | | - buf = recv_buffers[src] |
349 | | - shards.append(buf[offset : offset + numel]) |
350 | | - if len(shards) == 1: |
351 | | - return shards[0] |
352 | | - return torch.cat(shards) |
353 | | - |
354 | | - |
355 | | -@dataclasses.dataclass |
356 | | -class OwnerScatterPlan: |
357 | | - """Metadata and send buffers for the owner-scatter P2P step of a set of parameters. |
358 | | -
|
359 | | - The owner keeps its own result shard (applied directly), so it only sends to the other |
360 | | - shard-holding ranks. |
361 | | -
|
362 | | - Attributes: |
363 | | - send_buffers: Per-destination-rank flat send buffer (this owner's result shards for the |
364 | | - params it owns, in tensor-index order). Only destinations with non-zero send size |
365 | | - appear. |
366 | | - recv_sizes: Per-owner-rank element count this rank (as a destination) receives. Only owners |
367 | | - with non-zero total size appear. |
368 | | - recv_offsets: Per `(tensor_index, owner_rank)`, the flat offset of this param's result shard |
369 | | - inside the recv buffer received from `owner_rank`. Only contains tuples for which this |
370 | | - rank holds elements. |
371 | | - """ |
372 | | - |
373 | | - layout: GlobalLayout |
374 | | - local_range: Range |
375 | | - send_buffers: dict[int, torch.Tensor] |
376 | | - recv_sizes: dict[int, int] |
377 | | - recv_offsets: dict[tuple[int, int], int] |
378 | | - |
379 | | - @classmethod |
380 | | - def pack(cls, owner_layout: GroupOwnerLayout, full_results: dict[int, torch.Tensor]) -> Self: |
381 | | - """Pack this owner rank's full results into per-destination P2P send buffers. |
382 | | -
|
383 | | - Args: |
384 | | - owner_layout: The group's owner layout. |
385 | | - full_results: Full result tensor per parameter this rank owns. Tensors may be passed in |
386 | | - any shape. |
387 | | - """ |
388 | | - mesh = owner_layout.mesh |
389 | | - this_rank = mesh.get_rank() |
390 | | - layout = owner_layout.layout |
391 | | - placements = (RowAtomic(),) * mesh.ndim |
392 | | - local_range = layout.get_local_range(mesh, placements) |
393 | | - tensor_to_owner = owner_layout.tensor_to_owner |
394 | | - tensor_indices = sorted(tensor_to_owner) |
395 | | - flat_results = { |
396 | | - i: full_results[i].flatten() for i in tensor_indices if tensor_to_owner[i] == this_rank |
397 | | - } |
398 | | - |
399 | | - # Query each destination once and pack its shards in tensor-index order. |
400 | | - send_buffers: dict[int, torch.Tensor] = {} |
401 | | - for dest in mesh.mesh.flatten().tolist(): |
402 | | - if dest == this_rank: |
403 | | - continue |
404 | | - buffer_range = layout.get_rank_range(mesh, placements, dest) |
405 | | - chunks: list[torch.Tensor] = [] |
406 | | - for tensor_index, flat in flat_results.items(): |
407 | | - tensor_range = layout.get_tensor_range(tensor_index) |
408 | | - shard_range = intersect_ranges(tensor_range, buffer_range) |
409 | | - if shard_range.numel > 0: |
410 | | - offset = shard_range.start - tensor_range.start |
411 | | - chunks.append(flat.narrow(0, offset, shard_range.numel)) |
412 | | - if chunks: |
413 | | - send_buffers[dest] = torch.cat(chunks) |
414 | | - |
415 | | - # Compute each owner's receive size and offsets together from our local range. |
416 | | - recv_sizes: dict[int, int] = {} |
417 | | - recv_offsets: dict[tuple[int, int], int] = {} |
418 | | - for tensor_index in tensor_indices: |
419 | | - owner = tensor_to_owner[tensor_index] |
420 | | - if owner == this_rank: |
421 | | - continue |
422 | | - numel = intersect_ranges(layout.get_tensor_range(tensor_index), local_range).numel |
423 | | - if numel > 0: |
424 | | - offset = recv_sizes.get(owner, 0) |
425 | | - recv_offsets[(tensor_index, owner)] = offset |
426 | | - recv_sizes[owner] = offset + numel |
427 | | - |
428 | | - return cls( |
429 | | - layout=layout, |
430 | | - local_range=local_range, |
431 | | - send_buffers=send_buffers, |
432 | | - recv_sizes=recv_sizes, |
433 | | - recv_offsets=recv_offsets, |
434 | | - ) |
435 | | - |
436 | | - def unpack(self, recv_buffers: dict[int, torch.Tensor]) -> dict[int, torch.Tensor]: |
437 | | - """Extract this rank's local flat result shards from the per-owner recv buffers. |
438 | | -
|
439 | | - Args: |
440 | | - recv_buffers: Per-owner-rank received buffer (only owners that sent). |
441 | | -
|
442 | | - Returns: |
443 | | - Mapping from tensor index to this rank's local flat result shard, for parameters this |
444 | | - rank holds elements of but does NOT own. |
445 | | - """ |
446 | | - results: dict[int, torch.Tensor] = {} |
447 | | - for (tensor_index, owner), offset in self.recv_offsets.items(): |
448 | | - numel = intersect_ranges( |
449 | | - self.layout.get_tensor_range(tensor_index), self.local_range |
450 | | - ).numel |
451 | | - buf = recv_buffers[owner] |
452 | | - results[tensor_index] = buf[offset : offset + numel] |
453 | | - return results |
0 commit comments