Coverage for / home / jenkins / .local / lib / python3.10 / site-packages / hyper_parallel / core / shard / ops / parallel_sort.py: 94%
34 statements
« prev ^ index » next coverage.py v7.13.1, created at 2026-10-25 05:45 +0800
« prev ^ index » next coverage.py v7.13.1, created at 2026-10-25 05:45 +0800
1# Copyright 2026 Huawei Technologies Co., Ltd
2#
3# Licensed under the Apache License, Version 2.0 (the "License");
4# you may not use this file except in compliance with the License.
5# You may obtain a copy of the License at
6#
7# http://www.apache.org/licenses/LICENSE-2.0
8#
9# Unless required by applicable law or agreed to in writing, software
10# distributed under the License is distributed on an "AS IS" BASIS,
11# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12# See the License for the specific language governing permissions and
13# limitations under the License.
14# ============================================================================
15"""
16Distributed implementation for Sort operator.
17"""
19from typing import Tuple
21from .parallel_ops import DistributedOp
24def _normalize_sort_args(x, dim=-1, descending=False, stable=False):
25 return (x,), {'dim': dim, 'descending': descending, 'stable': stable}
28class SortDistributedOp(DistributedOp):
29 """Distributed implementation for Sort operator."""
31 def preprocess(self, args: tuple, kwargs: dict) -> tuple:
32 """
33 Preprocess arguments for Sort operator.
35 Args:
36 args (tuple): Input arguments, first element is the input tensor.
37 kwargs (dict): Keyword arguments (dim, descending, stable).
39 Returns:
40 tuple: (local_args, local_kwargs, cache_values)
41 """
42 args, kwargs = _normalize_sort_args(*args, **kwargs)
43 input_tensor = args[0]
44 dim = kwargs['dim']
45 descending = kwargs['descending']
46 stable = kwargs['stable']
48 local_args = (input_tensor.to_local(),)
49 local_kwargs = {'dim': dim, 'descending': descending, 'stable': stable}
51 cache_values = [input_tensor.layout, dim]
52 return local_args, local_kwargs, cache_values
54 def infer_layout(self, cache_values: list) -> Tuple[tuple, None]:
55 """
56 Infer output layouts for Sort operator.
58 Rules:
59 1. Input must not have Partial status.
60 2. dim must be an integer within the valid range [-ndim, ndim-1].
61 3. The sort dimension must not be sharded (including StridedShard multi-axis mappings).
62 4. Output values and indices layouts are identical to the input layout.
64 Args:
65 cache_values (list): [input_layout, dim] where dim is the sort dimension.
67 Returns:
68 tuple: ((values_layout, indices_layout), None)
70 Raises:
71 ValueError: If input has Partial status, dim is out of range, or the sort dimension
72 is sharded.
73 """
74 layout = cache_values[0]
75 dim = cache_values[1]
77 self._check_partial_inputs([layout])
79 if not isinstance(dim, int):
80 raise ValueError(
81 f"For {self.op_name}, dimension should be int, but got {type(dim)}"
82 )
84 # Get tensor map to check sharding status
85 in_tensor_map = layout.tensor_map
86 ndim = len(in_tensor_map)
88 if dim < -ndim or dim >= ndim:
89 raise ValueError(
90 f"For {self.op_name}, dimension out of range "
91 f"(expected to be in range of [{-ndim}, {ndim - 1}], but got {dim})"
92 )
94 if dim < 0:
95 dim += ndim
97 # Check if the sorting dimension is sharded.
98 # In tensor_map, -1 means Replicate (not sharded); any other value implies sharding.
99 mapping = in_tensor_map[dim]
100 if isinstance(mapping, (list, tuple)):
101 is_sharded = any(m != -1 for m in mapping)
102 else:
103 is_sharded = mapping != -1
105 if is_sharded:
106 raise ValueError(
107 f"For {self.op_name}, sorting along a sharded dimension "
108 f"(dim {dim} mapped to {mapping}) is not supported. "
109 f"Please redistribute the tensor to Replicate on this dimension before sorting."
110 )
112 return ((layout, layout), None)