Codekhana

torch/distributed/pipeline/sync/microbatch.py

12 issues

Module 'torch' has no 'cat' member

Error

Line: 216 Column: 18

              
    if outputs[0].atomic:
        tensors = tuple(b.tensor for b in outputs)
        output = torch.cat(tensors)
    else:
        output_buf: List[Any] = []
        for i in range(len(outputs[0])):
            output_type = type(outputs[0][i])
            current_outputs = []

Reported by Pylint.

Module 'torch' has no 'cat' member

Error

Line: 228 Column: 35

                              current_outputs.append(batch[i])

            if torch.is_tensor(outputs[0][i]):
                output_buf.append(torch.cat(current_outputs))
            else:
                output_buf.append(current_outputs)

        output = tuple(output_buf)

Reported by Pylint.

Redefining built-in 'input'

Error

Line: 183 Column: 9

                  batches: List[Any] = [[] for _ in range(chunks)]
    # Actual number of chunks produced
    num_chunks = -1
    for input in inputs:
        if torch.is_tensor(input):
            # Chunk only tensors.
            tensors = input.chunk(chunks)

            # Validate number of chunks equal across all inputs.

Reported by Pylint.

Too few public methods (1/2)

Error

Line: 23 Column: 1

              Function = Callable[[TensorOrTensors], Union[List[Any], Tensor]]


class NoChunk(object):
    """
    Wrapper for a Tensor in :meth:`Pipe.forward` indicating that the tensor
    should not be chunked on the batch dimension and instead be replicated
    as-is across all micro-batches. This is useful for tensors which might
    not have any 'batch' semantics for the model.

Reported by Pylint.

Class 'NoChunk' inherits from object, can be safely removed from bases in python3

Error

Line: 23 Column: 1

              Function = Callable[[TensorOrTensors], Union[List[Any], Tensor]]


class NoChunk(object):
    """
    Wrapper for a Tensor in :meth:`Pipe.forward` indicating that the tensor
    should not be chunked on the batch dimension and instead be replicated
    as-is across all micro-batches. This is useful for tensors which might
    not have any 'batch' semantics for the model.

Reported by Pylint.

Missing function or method docstring

Error

Line: 36 Column: 5

                      self._tensor = inp

    @property
    def tensor(self):
        return self._tensor


class Batch:
    """

Reported by Pylint.

Either all return statements in a function should return an expression, or none of them should.

Error

Line: 78 Column: 5

              
        raise TypeError("No tensor found!")

    def get_device(self):
        """
        Retrieves the device for this microbatch.
        """
        if self.atomic:
            return self._values.device  # type: ignore[union-attr]

Reported by Pylint.

Unnecessary "else" after "return"

Error

Line: 93 Column: 9

                      """Calls a function on the microbatch. It also wraps
        the output with :class:`Batch`.
        """
        if self.atomic:
            return Batch(function(self._values))
        else:
            return Batch(function(*self._values))

    def __repr__(self) -> str:

Reported by Pylint.

Line too long (104/100)

Error

Line: 137 Column: 1

                  def _setitem_by_index(self, index: int, value) -> None:
        if not self.atomic:
            i = index
            self._values = self._values[:i] + (value,) + self._values[i + 1 :]  # type: ignore[operator]
            return

        if index != 0:
            raise IndexError("atomic batch allows index 0 only")

Reported by Pylint.

Unnecessary parens after 'not' keyword

Error

Line: 146 Column: 1

                      self._values = value

    def _setitem_by_slice(self, index: slice, value) -> None:
        if not (index.start is index.stop is index.step is None):
            raise NotImplementedError("only slice [:] supported")

        if not self.atomic:
            self._values = value
            return

Reported by Pylint.

tools/code_analyzer/gen_op_registration_allowlist.py

12 issues

Redefining name 'root_ops' from outer scope (line 93)

Error

Line: 47 Column: 5

              
def gen_transitive_closure(
    dep_graph: DepGraph,
    root_ops: List[str],
    train: bool = False,
) -> List[str]:
    result = set(root_ops)
    queue = root_ops[:]

Reported by Pylint.

TODO: when FL is migrated from full-jit to lite trainer, remove '__ROOT__'

Error

Line: 62 Column: 3

                  # `__ROOT__` key to include ops which can be called from C++ code directly,
    # in addition to ops that are called from TorchScript model.
    # '__ROOT__' is only needed for full-jit. Keep it only for training.
    # TODO: when FL is migrated from full-jit to lite trainer, remove '__ROOT__'
    if train:
        queue.append('__ROOT__')

    while queue:
        cur = queue.pop()

Reported by Pylint.

Redefining name 'root_ops' from outer scope (line 93)

Error

Line: 75 Column: 53

              
    return sorted(result)

def gen_transitive_closure_str(dep_graph: DepGraph, root_ops: List[str]) -> str:
    return ' '.join(gen_transitive_closure(dep_graph, root_ops))


if __name__ == "__main__":
    parser = argparse.ArgumentParser(

Reported by Pylint.

standard import "from collections import defaultdict" should be placed before "import yaml"

Error

Line: 14 Column: 1

              import argparse
import yaml

from collections import defaultdict
from typing import Dict, List, Set


def canonical_name(opname: str) -> str:
    # Skip the overload name part as it's not supported by code analyzer yet.

Reported by Pylint.

standard import "from typing import Dict, List, Set" should be placed before "import yaml"

Error

Line: 15 Column: 1

              import yaml

from collections import defaultdict
from typing import Dict, List, Set


def canonical_name(opname: str) -> str:
    # Skip the overload name part as it's not supported by code analyzer yet.
    return opname.split('.', 1)[0]

Reported by Pylint.

Missing function or method docstring

Error

Line: 18 Column: 1

              from typing import Dict, List, Set


def canonical_name(opname: str) -> str:
    # Skip the overload name part as it's not supported by code analyzer yet.
    return opname.split('.', 1)[0]


DepGraph = Dict[str, Set[str]]

Reported by Pylint.

Missing function or method docstring

Error

Line: 26 Column: 1

              DepGraph = Dict[str, Set[str]]


def load_op_dep_graph(fname: str) -> DepGraph:
    with open(fname, 'r') as stream:
        result = defaultdict(set)
        for op in yaml.safe_load(stream):
            op_name = canonical_name(op['name'])
            for dep in op.get('depends', []):

Reported by Pylint.

Variable name "op" doesn't conform to snake_case naming style

Error

Line: 29 Column: 13

              def load_op_dep_graph(fname: str) -> DepGraph:
    with open(fname, 'r') as stream:
        result = defaultdict(set)
        for op in yaml.safe_load(stream):
            op_name = canonical_name(op['name'])
            for dep in op.get('depends', []):
                dep_name = canonical_name(dep['name'])
                result[op_name].add(dep_name)
        return dict(result)

Reported by Pylint.

Missing function or method docstring

Error

Line: 37 Column: 1

                      return dict(result)


def load_root_ops(fname: str) -> List[str]:
    result = []
    with open(fname, 'r') as stream:
        for op in yaml.safe_load(stream):
            result.append(canonical_name(op))
    return result

Reported by Pylint.

Variable name "op" doesn't conform to snake_case naming style

Error

Line: 40 Column: 13

              def load_root_ops(fname: str) -> List[str]:
    result = []
    with open(fname, 'r') as stream:
        for op in yaml.safe_load(stream):
            result.append(canonical_name(op))
    return result


def gen_transitive_closure(

Reported by Pylint.