@property
    @abstractmethod
    def join_device(self) -> torch.device:
        r"""
        Returns the device from which to perform collective communications
        needed by the join context manager implementation itself.
        """
        ...

            

Reported by Pylint.

                      Returns the number of non-joined processes by shadowing an all-reduce
        in the non-joined processes.
        """
        num_nonjoined_procs = torch.zeros(1, device=self._device)
        dist.all_reduce(num_nonjoined_procs, group=self._process_group)
        return num_nonjoined_procs.item()

    def _notify_procs_to_terminate(self):
        r"""

            

Reported by Pylint.

                      and raises a ``RuntimeError`` indicating that the current process has
        exhausted its inputs.
        """
        ones = torch.ones(1, device=self._device)
        dist.all_reduce(ones, group=self._process_group)
        # NOTE: Raising `StopIteration` does not throw an error in Python 3.6
        # and throws a `RuntimeError` in Python 3.7+ (PEP 479), so we just
        # raise a `RuntimeError` here
        raise RuntimeError(f"Rank {self._rank} exhausted all inputs.")

            

Reported by Pylint.

                      process_group = joinable.join_process_group

        # Schedule an all-reduce to indicate that the caller has not yet joined
        ones = torch.ones(1, device=device)
        work = dist.all_reduce(ones, group=process_group, async_op=True)

        if join_config.throw_on_early_termination:
            # Check if uneven inputs have been detected
            zeros = torch.zeros(1, device=device)

            

Reported by Pylint.

              
        if join_config.throw_on_early_termination:
            # Check if uneven inputs have been detected
            zeros = torch.zeros(1, device=device)
            dist.all_reduce(zeros, group=process_group)
            should_throw = zeros.item()
            if should_throw:
                raise RuntimeError(
                    "Detected at least one rank that exhausted inputs. "

            

Reported by Pylint.

                      """
        ...

    def post_hook(self, is_last_joiner: bool) -> None:
        r"""
        This hook is called after all processes have joined. It is passed an
        additional ``bool`` argument ``is_last_joiner``, which indicates if the
        rank is one of the last to join.


            

Reported by Pylint.

                      assert len(self._joinables) > 0
        is_first_joinable = True
        for joinable in self._joinables:
            joinable._join_config = _JoinConfig(
                enable=self._enable,
                throw_on_early_termination=self._throw_on_early_termination,
                is_first_joinable=is_first_joinable
            )
            is_first_joinable = False

            

Reported by Pylint.

              
    def __exit__(
        self,
        type: Optional[Type[BaseException]],
        value: Optional[BaseException],
        traceback: Optional[TracebackType]
    ):
        r"""
        Repeatedly runs the main hooks until all processes join; then, runs

            

Reported by Pylint.

                          f"Check that the {type(joinable)} constructor calls the " \
            "``Joinable`` constructor"

        join_config = joinable._join_config
        # First joinable is responsible for the collective communications
        if not join_config.is_first_joinable or not join_config.enable:
            return None

        device = joinable.join_device

            

Reported by Pylint.

              import warnings
from abc import ABC, abstractmethod
from types import TracebackType
from typing import Any, List, NamedTuple, Optional, Type

import torch
import torch.distributed as dist



            

Reported by Pylint.

              from torch.testing._internal.common_utils import TestCase, run_tests, slowTest, IS_WINDOWS

import subprocess
import tempfile
import os
import unittest

PYTORCH_COLLECT_COVERAGE = bool(os.environ.get("PYTORCH_COLLECT_COVERAGE"))


            

Reported by Pylint.

                          if disable_gpu:
                cmd += ['--gpu', '-1']

            res = subprocess.run(cmd)

            self.assertTrue(res.returncode == 0)
            # Check that something was written to the file
            out_file.seek(0, os.SEEK_END)
            self.assertTrue(out_file.tell() > 0)

            

Reported by Pylint.

              from torch.testing._internal.common_utils import TestCase, run_tests, slowTest, IS_WINDOWS

import subprocess
import tempfile
import os
import unittest

PYTORCH_COLLECT_COVERAGE = bool(os.environ.get("PYTORCH_COLLECT_COVERAGE"))


            

Reported by Pylint.

              from torch.testing._internal.common_utils import TestCase, run_tests, slowTest, IS_WINDOWS

import subprocess
import tempfile
import os
import unittest

PYTORCH_COLLECT_COVERAGE = bool(os.environ.get("PYTORCH_COLLECT_COVERAGE"))


            

Reported by Bandit.

              from torch.testing._internal.common_utils import TestCase, run_tests, slowTest, IS_WINDOWS

import subprocess
import tempfile
import os
import unittest

PYTORCH_COLLECT_COVERAGE = bool(os.environ.get("PYTORCH_COLLECT_COVERAGE"))


            

Reported by Pylint.

              from torch.testing._internal.common_utils import TestCase, run_tests, slowTest, IS_WINDOWS

import subprocess
import tempfile
import os
import unittest

PYTORCH_COLLECT_COVERAGE = bool(os.environ.get("PYTORCH_COLLECT_COVERAGE"))


            

Reported by Pylint.

              
import subprocess
import tempfile
import os
import unittest

PYTORCH_COLLECT_COVERAGE = bool(os.environ.get("PYTORCH_COLLECT_COVERAGE"))

# This is a very simple smoke test for the functional autograd benchmarking script.

            

Reported by Pylint.

              import subprocess
import tempfile
import os
import unittest

PYTORCH_COLLECT_COVERAGE = bool(os.environ.get("PYTORCH_COLLECT_COVERAGE"))

# This is a very simple smoke test for the functional autograd benchmarking script.
class TestFunctionalAutogradBenchmark(TestCase):

            

Reported by Pylint.

              PYTORCH_COLLECT_COVERAGE = bool(os.environ.get("PYTORCH_COLLECT_COVERAGE"))

# This is a very simple smoke test for the functional autograd benchmarking script.
class TestFunctionalAutogradBenchmark(TestCase):
    def _test_runner(self, model, disable_gpu=False):
        # Note about windows:
        # The temporary file is exclusively open by this process and the child process
        # is not allowed to open it again. As this is a simple smoke test, we choose for now
        # not to run this on windows and keep the code here simple.

            

Reported by Pylint.

                      # is not allowed to open it again. As this is a simple smoke test, we choose for now
        # not to run this on windows and keep the code here simple.
        with tempfile.NamedTemporaryFile() as out_file:
            cmd = ['python', '../benchmarks/functional_autograd_benchmark/functional_autograd_benchmark.py']
            # Only run the warmup
            cmd += ['--num-iters', '0']
            # Only run the vjp task (fastest one)
            cmd += ['--task-filter', 'vjp']
            # Only run the specified model

            

Reported by Pylint.

              from torch.nn.functional import batch_norm
from torch.nn.modules.batchnorm import _BatchNorm

from .checkpoint import is_recomputing

__all__ = ["DeferredBatchNorm"]


TModule = TypeVar("TModule", bound=nn.Module)

            

Reported by Pylint.

                  ) -> None:
        super().__init__(num_features, eps, momentum, affine, track_running_stats=True)

        self.register_buffer("sum", torch.zeros_like(self.running_mean))
        self.register_buffer("sum_squares", torch.zeros_like(self.running_var))

        self.counter = 0
        self.tracked = 0
        self.chunks = chunks

            

Reported by Pylint.

                      super().__init__(num_features, eps, momentum, affine, track_running_stats=True)

        self.register_buffer("sum", torch.zeros_like(self.running_mean))
        self.register_buffer("sum_squares", torch.zeros_like(self.running_var))

        self.counter = 0
        self.tracked = 0
        self.chunks = chunks


            

Reported by Pylint.

                      dim.extend(range(2, input.dim()))

        with torch.no_grad():
            self.sum += input.sum(dim)
            self.sum_squares += (input ** 2).sum(dim)

        size = input.size().numel() // input.size(1)
        self.counter += size
        self.tracked += 1

            

Reported by Pylint.

              
        with torch.no_grad():
            self.sum += input.sum(dim)
            self.sum_squares += (input ** 2).sum(dim)

        size = input.size().numel() // input.size(1)
        self.counter += size
        self.tracked += 1


            

Reported by Pylint.

                      else:  # use exponential moving average
            exponential_average_factor = self.momentum

        mean = self.sum / self.counter
        var = self.sum_squares / self.counter - mean ** 2

        # Calculate the exponential moving average here.
        m = exponential_average_factor


            

Reported by Pylint.

                          exponential_average_factor = self.momentum

        mean = self.sum / self.counter
        var = self.sum_squares / self.counter - mean ** 2

        # Calculate the exponential moving average here.
        m = exponential_average_factor

        self.running_mean *= 1 - m

            

Reported by Pylint.

                      self.running_var *= 1 - m
        self.running_var += var * m

        self.sum.zero_()
        self.sum_squares.zero_()
        self.counter = 0
        self.tracked = 0

    def forward(self, input: Tensor) -> Tensor:

            

Reported by Pylint.

                      self.running_var += var * m

        self.sum.zero_()
        self.sum_squares.zero_()
        self.counter = 0
        self.tracked = 0

    def forward(self, input: Tensor) -> Tensor:
        if not self.training:

            

Reported by Pylint.

                      self.tracked = 0
        self.chunks = chunks

    def _check_input_dim(self, input: Tensor) -> None:
        # It's the typical _check_input_dim() implementation in PyTorch.
        if input.dim() <= 2:
            raise ValueError("expected at least 3D input (got %dD input)" % input.dim())

    def _track(self, input: Tensor) -> bool:

            

Reported by Pylint.

                      )

    if server_port != -1:
        log.info(f"sever_port: {server_port}, specified, ignoring retries")

    # only retry when server_port is NOT static
    attempt = retries if server_port == -1 else 1
    while True:
        if server_port != -1:

            

Reported by Pylint.

                      else:
            port = get_free_port()

        log.info(
            f"Creating c10d store on {server_addr}:{port}\n"
            f"  world_size  : {world_size}\n"
            f"  is_server   : {is_server}\n"
            f"  timeout(sec): {timeout}\n"
        )

            

Reported by Pylint.

                          # so we parse the error msg for now, interestingly this is how torch itself
            # detects timeouts and port conflicts in their own unittests
            # see - caffe2/torch/testing/_internal/common_utils.py
            # TODO properly map the exceptions in pybind (c10d/init.cpp)
            if _CONNECT_TIMEOUT in str(e) and not is_server:
                raise TimeoutError(
                    f"timed out waiting for tcp store's server: {server_addr}:{port}"
                ) from e
            elif str(e) == _ADDRESS_IN_USE:  # this will only happen on the server

            

Reported by Pylint.

                              ) from e
            elif str(e) == _ADDRESS_IN_USE:  # this will only happen on the server
                if attempt < retries:
                    log.warning(
                        f"port: {port} already in use, attempt: [{attempt}/{retries}]"
                    )
                    attempt += 1
                else:
                    raise IOError(

            

Reported by Pylint.

                      host="localhost", port=None, family=socket.AF_UNSPEC, type=socket.SOCK_STREAM
    )
    for addr in addrs:
        family, type, proto, _, _ = addr
        s = socket.socket(family, type, proto)
        try:
            s.bind(("localhost", 0))
            s.listen(0)
            return s

            

Reported by Pylint.

              #!/usr/bin/env python3

# Copyright (c) Facebook, Inc. and its affiliates.
# All rights reserved.
#
# This source code is licensed under the BSD-style license found in the
# LICENSE file in the root directory of this source tree.
import datetime
import socket

            

Reported by Pylint.

              _LAST_MEMBER_CHECKIN = "_tcp_store/last_member"


def create_c10d_store(
    is_server: bool,
    server_addr: str,
    server_port: int = -1,
    world_size: int = 1,
    timeout: float = (60 * 10),  # 10 min

            

Reported by Pylint.

              _LAST_MEMBER_CHECKIN = "_tcp_store/last_member"


def create_c10d_store(
    is_server: bool,
    server_addr: str,
    server_port: int = -1,
    world_size: int = 1,
    timeout: float = (60 * 10),  # 10 min

            

Reported by Pylint.

              ):
    if server_port == -1 and world_size > 1:
        raise ValueError(
            f"server_port must be specified when world_size > 1, got server_port={server_port}, world_size={world_size}"
        )

    if server_port != -1:
        log.info(f"sever_port: {server_port}, specified, ignoring retries")


            

Reported by Pylint.

                          _check_full_rank(store, world_size)
            log.info("Successfully created c10d store")
            return store
        except RuntimeError as e:
            # this is brittle, but the underlying exception type is not properly pybinded
            # so we parse the error msg for now, interestingly this is how torch itself
            # detects timeouts and port conflicts in their own unittests
            # see - caffe2/torch/testing/_internal/common_utils.py
            # TODO properly map the exceptions in pybind (c10d/init.cpp)

            

Reported by Pylint.

              import collections
import unittest

import torch
from torch.testing._internal.common_utils import (
    TestCase, run_tests, TEST_WITH_ASAN)

try:
    import psutil

            

Reported by Pylint.

              import unittest

import torch
from torch.testing._internal.common_utils import (
    TestCase, run_tests, TEST_WITH_ASAN)

try:
    import psutil
    HAS_PSUTIL = True

            

Reported by Pylint.

                      p = psutil.Process()
        # warm up for 5 runs, then things should be stable for the last 5
        last_rss = collections.deque(maxlen=5)
        for n in range(10):
            for i in range(runs):
                self.model(self.x)
            last_rss.append(p.memory_info().rss)
        return last_rss


            

Reported by Pylint.

                      # warm up for 5 runs, then things should be stable for the last 5
        last_rss = collections.deque(maxlen=5)
        for n in range(10):
            for i in range(runs):
                self.model(self.x)
            last_rss.append(p.memory_info().rss)
        return last_rss

    def func_rss(self, runs):

            

Reported by Pylint.

              import collections
import unittest

import torch
from torch.testing._internal.common_utils import (
    TestCase, run_tests, TEST_WITH_ASAN)

try:
    import psutil

            

Reported by Pylint.

              device = torch.device('cpu')


class Network(torch.nn.Module):
    maxp1 = torch.nn.MaxPool2d(1, 1)

    def forward(self, x):
        return self.maxp1(x)


            

Reported by Pylint.

              device = torch.device('cpu')


class Network(torch.nn.Module):
    maxp1 = torch.nn.MaxPool2d(1, 1)

    def forward(self, x):
        return self.maxp1(x)


            

Reported by Pylint.

              class Network(torch.nn.Module):
    maxp1 = torch.nn.MaxPool2d(1, 1)

    def forward(self, x):
        return self.maxp1(x)


@unittest.skipIf(not HAS_PSUTIL, "Requires psutil to run")
@unittest.skipIf(TEST_WITH_ASAN, "Cannot test with ASAN")

            

Reported by Pylint.

              class Network(torch.nn.Module):
    maxp1 = torch.nn.MaxPool2d(1, 1)

    def forward(self, x):
        return self.maxp1(x)


@unittest.skipIf(not HAS_PSUTIL, "Requires psutil to run")
@unittest.skipIf(TEST_WITH_ASAN, "Cannot test with ASAN")

            

Reported by Pylint.

              
@unittest.skipIf(not HAS_PSUTIL, "Requires psutil to run")
@unittest.skipIf(TEST_WITH_ASAN, "Cannot test with ASAN")
class TestOpenMP_ParallelFor(TestCase):
    batch = 20
    channels = 1
    side_dim = 80
    x = torch.randn([batch, channels, side_dim, side_dim], device=device)
    model = Network()

            

Reported by Pylint.

                      self.low, self.high = broadcast_all(low, high)

        if isinstance(low, Number) and isinstance(high, Number):
            batch_shape = torch.Size()
        else:
            batch_shape = self.low.size()
        super(Uniform, self).__init__(batch_shape, validate_args=validate_args)

        if self._validate_args and not torch.lt(self.low, self.high).all():

            

Reported by Pylint.

                          batch_shape = self.low.size()
        super(Uniform, self).__init__(batch_shape, validate_args=validate_args)

        if self._validate_args and not torch.lt(self.low, self.high).all():
            raise ValueError("Uniform is not defined when low>= high")

    def expand(self, batch_shape, _instance=None):
        new = self._get_checked_instance(Uniform, _instance)
        batch_shape = torch.Size(batch_shape)

            

Reported by Pylint.

              
    def expand(self, batch_shape, _instance=None):
        new = self._get_checked_instance(Uniform, _instance)
        batch_shape = torch.Size(batch_shape)
        new.low = self.low.expand(batch_shape)
        new.high = self.high.expand(batch_shape)
        super(Uniform, new).__init__(batch_shape, validate_args=False)
        new._validate_args = self._validate_args
        return new

            

Reported by Pylint.

                  def support(self):
        return constraints.interval(self.low, self.high)

    def rsample(self, sample_shape=torch.Size()):
        shape = self._extended_shape(sample_shape)
        rand = torch.rand(shape, dtype=self.low.dtype, device=self.low.device)
        return self.low + rand * (self.high - self.low)

    def log_prob(self, value):

            

Reported by Pylint.

              
    def rsample(self, sample_shape=torch.Size()):
        shape = self._extended_shape(sample_shape)
        rand = torch.rand(shape, dtype=self.low.dtype, device=self.low.device)
        return self.low + rand * (self.high - self.low)

    def log_prob(self, value):
        if self._validate_args:
            self._validate_sample(value)

            

Reported by Pylint.

                          self._validate_sample(value)
        lb = self.low.le(value).type_as(self.low)
        ub = self.high.gt(value).type_as(self.low)
        return torch.log(lb.mul(ub)) - torch.log(self.high - self.low)

    def cdf(self, value):
        if self._validate_args:
            self._validate_sample(value)
        result = (value - self.low) / (self.high - self.low)

            

Reported by Pylint.

                          self._validate_sample(value)
        lb = self.low.le(value).type_as(self.low)
        ub = self.high.gt(value).type_as(self.low)
        return torch.log(lb.mul(ub)) - torch.log(self.high - self.low)

    def cdf(self, value):
        if self._validate_args:
            self._validate_sample(value)
        result = (value - self.low) / (self.high - self.low)

            

Reported by Pylint.

                      return result

    def entropy(self):
        return torch.log(self.high - self.low)

            

Reported by Pylint.

              from torch.distributions.utils import broadcast_all


class Uniform(Distribution):
    r"""
    Generates uniformly distributed random samples from the half-open interval
    ``[low, high)``.

    Example::

            

Reported by Pylint.

                      low (float or Tensor): lower range (inclusive).
        high (float or Tensor): upper range (exclusive).
    """
    # TODO allow (loc,scale) parameterization to allow independent constraints.
    arg_constraints = {'low': constraints.dependent(is_discrete=False, event_dim=0),
                       'high': constraints.dependent(is_discrete=False, event_dim=0)}
    has_rsample = True

    @property

            

Reported by Pylint.

              import unittest
from typing import Dict, List

from tools.stats import print_test_stats
from tools.stats.s3_stat_parser import (Commit, Report, ReportMetaMeta,
                                        Status, Version1Case,
                                        Version1Report, Version2Case,
                                        Version2Report)


            

Reported by Pylint.

              from typing import Dict, List

from tools.stats import print_test_stats
from tools.stats.s3_stat_parser import (Commit, Report, ReportMetaMeta,
                                        Status, Version1Case,
                                        Version1Report, Version2Case,
                                        Version2Report)



            

Reported by Pylint.

              # -*- coding: utf-8 -*-
import unittest
from typing import Dict, List

from tools.stats import print_test_stats
from tools.stats.s3_stat_parser import (Commit, Report, ReportMetaMeta,
                                        Status, Version1Case,
                                        Version1Report, Version2Case,
                                        Version2Report)

            

Reported by Pylint.

                                                      Version2Report)


def fakehash(char: str) -> str:
    return char * 40


def dummy_meta_meta() -> ReportMetaMeta:
    return {

            

Reported by Pylint.

                  return char * 40


def dummy_meta_meta() -> ReportMetaMeta:
    return {
        'build_pr': '',
        'build_tag': '',
        'build_sha1': '',
        'build_base_commit': '',

            

Reported by Pylint.

                  }


def makecase(
    name: str,
    seconds: float,
    *,
    errored: bool = False,
    failed: bool = False,

            

Reported by Pylint.

                  }


def make_report_v1(tests: Dict[str, List[Version1Case]]) -> Version1Report:
    suites = {
        suite_name: {
            'total_seconds': sum(case['seconds'] for case in cases),
            'cases': cases,
        }

            

Reported by Pylint.

                  }


def make_case_v2(seconds: float, status: Status = None) -> Version2Case:
    return {
        'seconds': seconds,
        'status': status,
    }


            

Reported by Pylint.

                  }


def make_report_v2(tests: Dict[str, Dict[str, Dict[str, Version2Case]]]) -> Version2Report:
    files = {}
    for file_name, file_suites in tests.items():
        suites = {
            suite_name: {
                'total_seconds': sum(case['seconds'] for case in cases.values()),

            

Reported by Pylint.

                      'total_seconds': sum(s['total_seconds'] for s in files.values()),
        'files': files,
    }
maxDiff = None

class TestPrintTestStats(unittest.TestCase):
    version1_report: Version1Report = make_report_v1({
        # input ordering of the suites is ignored
        'Grault': [

            

Reported by Pylint.

                              if param not in self.state:
                    self.state[param] = {}
                    state = self.state[param]
                    state['step'] = torch.tensor(0.0)
                    # Exponential moving average of gradient values
                    state['exp_avg'] = torch.zeros_like(param, memory_format=torch.preserve_format)
                    # Exponential moving average of squared gradient values
                    state['exp_inf'] = torch.zeros_like(param, memory_format=torch.preserve_format)


            

Reported by Pylint.

                                  state = self.state[param]
                    state['step'] = torch.tensor(0.0)
                    # Exponential moving average of gradient values
                    state['exp_avg'] = torch.zeros_like(param, memory_format=torch.preserve_format)
                    # Exponential moving average of squared gradient values
                    state['exp_inf'] = torch.zeros_like(param, memory_format=torch.preserve_format)

                state = self.state[param]


            

Reported by Pylint.

                                  state = self.state[param]
                    state['step'] = torch.tensor(0.0)
                    # Exponential moving average of gradient values
                    state['exp_avg'] = torch.zeros_like(param, memory_format=torch.preserve_format)
                    # Exponential moving average of squared gradient values
                    state['exp_inf'] = torch.zeros_like(param, memory_format=torch.preserve_format)

                state = self.state[param]


            

Reported by Pylint.

                                  # Exponential moving average of gradient values
                    state['exp_avg'] = torch.zeros_like(param, memory_format=torch.preserve_format)
                    # Exponential moving average of squared gradient values
                    state['exp_inf'] = torch.zeros_like(param, memory_format=torch.preserve_format)

                state = self.state[param]

                exp_avgs.append(state['exp_avg'])
                exp_infs.append(state['exp_inf'])

            

Reported by Pylint.

                                  # Exponential moving average of gradient values
                    state['exp_avg'] = torch.zeros_like(param, memory_format=torch.preserve_format)
                    # Exponential moving average of squared gradient values
                    state['exp_inf'] = torch.zeros_like(param, memory_format=torch.preserve_format)

                state = self.state[param]

                exp_avgs.append(state['exp_avg'])
                exp_infs.append(state['exp_inf'])

            

Reported by Pylint.

              from typing import List, Dict, Optional, Tuple
import torch
import torch.optim._functional as F

from torch import Tensor

# Define a TorchScript compatible Functional Adamax Optimizer
# where we use these optimizer in a functional way.
# Instead of using the `param.grad` when updating parameters,

            

Reported by Pylint.

              # NOTE: This should be only used by distributed optimizer internals
# and not meant to expose to the user.
@torch.jit.script
class _FunctionalAdamax(object):
    def __init__(
        self,
        params: List[Tensor],
        lr: float = 1e-3,
        betas: Tuple[float, float] = (0.9, 0.999),

            

Reported by Pylint.

              # NOTE: This should be only used by distributed optimizer internals
# and not meant to expose to the user.
@torch.jit.script
class _FunctionalAdamax(object):
    def __init__(
        self,
        params: List[Tensor],
        lr: float = 1e-3,
        betas: Tuple[float, float] = (0.9, 0.999),

            

Reported by Pylint.

              # and not meant to expose to the user.
@torch.jit.script
class _FunctionalAdamax(object):
    def __init__(
        self,
        params: List[Tensor],
        lr: float = 1e-3,
        betas: Tuple[float, float] = (0.9, 0.999),
        eps: float = 1e-8,

            

Reported by Pylint.

                      weight_decay: float = 0.0,
        _allow_empty_param_list: bool = False,
    ):
        if not 0.0 <= lr:
            raise ValueError("Invalid learning rate: {}".format(lr))
        if not 0.0 <= eps:
            raise ValueError("Invalid epsilon value: {}".format(eps))
        if not 0.0 <= betas[0] < 1.0:
            raise ValueError("Invalid beta parameter at index 0: {}".format(betas[0]))

            

Reported by Pylint.

              from typing import List, Union
import torch

from ._internals import (
    ShardMetadata,
    validate_non_overlapping_shards_metadata
)

class PlacementSpec(ABC):

            

Reported by Pylint.

                  class can be used to specify customized placements which might not be
    covered by existing APIs.
    """
    pass


@dataclass
class DevicePlacementSpec(PlacementSpec):
    """

            

Reported by Pylint.

                      device(:class:`torch.distributed._remote_device`): The device to place the entity on.
    """

    device: torch.distributed._remote_device

    def __post_init__(self):
        if not isinstance(self.device, torch.distributed._remote_device):
            self.device = torch.distributed._remote_device(self.device)


            

Reported by Pylint.

                  device: torch.distributed._remote_device

    def __post_init__(self):
        if not isinstance(self.device, torch.distributed._remote_device):
            self.device = torch.distributed._remote_device(self.device)


class ShardingSpec(PlacementSpec):
    """

            

Reported by Pylint.

              
    def __post_init__(self):
        if not isinstance(self.device, torch.distributed._remote_device):
            self.device = torch.distributed._remote_device(self.device)


class ShardingSpec(PlacementSpec):
    """
    Base class representing sharding specifications. It is special type of

            

Reported by Pylint.

                  Base class representing sharding specifications. It is special type of
    PlacementSpec.
    """
    pass


@dataclass
class ChunkShardingSpec(ShardingSpec):
    """

            

Reported by Pylint.

                  ShardingDim = Union[int, str]

    dim: ShardingDim
    placements: List[Union[torch.distributed._remote_device, str]]

    def __post_init__(self):
        self._verify_dim(self.dim)
        for i, remote_device in enumerate(self.placements):
            if not isinstance(remote_device, torch.distributed._remote_device):

            

Reported by Pylint.

                  def __post_init__(self):
        self._verify_dim(self.dim)
        for i, remote_device in enumerate(self.placements):
            if not isinstance(remote_device, torch.distributed._remote_device):
                self.placements[i] = torch.distributed._remote_device(remote_device)

    @staticmethod
    def _verify_dim(dim):
        if not (isinstance(dim, int) or isinstance(dim, str)):

            

Reported by Pylint.

                      self._verify_dim(self.dim)
        for i, remote_device in enumerate(self.placements):
            if not isinstance(remote_device, torch.distributed._remote_device):
                self.placements[i] = torch.distributed._remote_device(remote_device)

    @staticmethod
    def _verify_dim(dim):
        if not (isinstance(dim, int) or isinstance(dim, str)):
            raise ValueError(f'{dim} needs to either be an int or str')

            

Reported by Pylint.

              from abc import ABC
from dataclasses import dataclass
from typing import List, Union
import torch

from ._internals import (
    ShardMetadata,
    validate_non_overlapping_shards_metadata
)

            

Reported by Pylint.

              
import torch
from torch.testing._internal.common_utils import TestCase, run_tests
from torch._C import parse_schema


class TestFunctionSchema(TestCase):
    def test_serialize_and_deserialize(self):
        schemas = torch._C._jit_get_all_schemas()

            

Reported by Pylint.

              
import torch
from torch.testing._internal.common_utils import TestCase, run_tests
from torch._C import parse_schema


class TestFunctionSchema(TestCase):
    def test_serialize_and_deserialize(self):
        schemas = torch._C._jit_get_all_schemas()

            

Reported by Pylint.

              
import torch
from torch.testing._internal.common_utils import TestCase, run_tests
from torch._C import parse_schema


class TestFunctionSchema(TestCase):
    def test_serialize_and_deserialize(self):
        schemas = torch._C._jit_get_all_schemas()

            

Reported by Pylint.

              
class TestFunctionSchema(TestCase):
    def test_serialize_and_deserialize(self):
        schemas = torch._C._jit_get_all_schemas()
        # so far we have around 1700 registered schemas
        self.assertGreater(len(schemas), 1000)
        for schema in schemas:
            parsed_schema = parse_schema(str(schema))
            self.assertEqual(parsed_schema, schema)

            

Reported by Pylint.

              
class TestFunctionSchema(TestCase):
    def test_serialize_and_deserialize(self):
        schemas = torch._C._jit_get_all_schemas()
        # so far we have around 1700 registered schemas
        self.assertGreater(len(schemas), 1000)
        for schema in schemas:
            parsed_schema = parse_schema(str(schema))
            self.assertEqual(parsed_schema, schema)

            

Reported by Pylint.

              
    def test_schema_error(self):
        with self.assertRaisesRegex(RuntimeError, r"schemas with vararg \(...\) can't have default value args"):
            schema = parse_schema("any.foo(int arg1, int arg2=0, ...)")

if __name__ == '__main__':
    run_tests()

            

Reported by Pylint.

              
import torch
from torch.testing._internal.common_utils import TestCase, run_tests
from torch._C import parse_schema


class TestFunctionSchema(TestCase):
    def test_serialize_and_deserialize(self):
        schemas = torch._C._jit_get_all_schemas()

            

Reported by Pylint.

              from torch._C import parse_schema


class TestFunctionSchema(TestCase):
    def test_serialize_and_deserialize(self):
        schemas = torch._C._jit_get_all_schemas()
        # so far we have around 1700 registered schemas
        self.assertGreater(len(schemas), 1000)
        for schema in schemas:

            

Reported by Pylint.

              

class TestFunctionSchema(TestCase):
    def test_serialize_and_deserialize(self):
        schemas = torch._C._jit_get_all_schemas()
        # so far we have around 1700 registered schemas
        self.assertGreater(len(schemas), 1000)
        for schema in schemas:
            parsed_schema = parse_schema(str(schema))

            

Reported by Pylint.

                          self.assertEqual(parsed_schema, schema)
            self.assertTrue(parsed_schema.is_backward_compatible_with(schema))

    def test_backward_compatible_structure(self):
        old_schema = parse_schema('any.over(Tensor self, *, Tensor b) -> Tensor')
        # BC: A new schema without changes.
        new_schema = parse_schema('any.over(Tensor self, *, Tensor b) -> Tensor')
        self.assertTrue(new_schema.is_backward_compatible_with(old_schema))
        # No-BC: A new schema with different name.

            

Reported by Pylint.