def __init__(self, *, dht: DHT, run_id: str, target_batch_size: int, batch_size_per_step: Optional[int]=None, optimizer: Union[(TorchOptimizer, OptimizerFactory)], params: Optional[Union[(Parameters, ParamGroups)]]=None, scheduler: Optional[Union[(LRSchedulerBase, SchedulerFactory)]]=None, matchmaking_time: Optional[float]=15.0, averaging_timeout: Optional[float]=60.0, allreduce_timeout: Optional[float]=None, next_chunk_timeout: Optional[float]=None, load_state_timeout: float=600.0, reuse_grad_buffers: bool=False, offload_optimizer: Optional[bool]=None, delay_optimizer_step: Optional[bool]=None, delay_grad_averaging: bool=False, delay_state_averaging: bool=True, average_state_every: int=1, use_local_updates: bool=False, client_mode: bool=None, auxiliary: bool=False, grad_compression: CompressionBase=NoCompression(), state_averaging_compression: CompressionBase=NoCompression(), load_state_compression: CompressionBase=NoCompression(), average_opt_statistics: Sequence[str]=(), extra_tensors: Sequence[torch.Tensor]=(), averager_opts: Optional[dict]=None, tracker_opts: Optional[dict]=None, performance_ema_alpha: float=0.1, shutdown_timeout: float=5, verbose: bool=False):
    self._parent_pid = os.getpid()
    client_mode = (client_mode if (client_mode is None) else dht.client_mode)
    delay_optimizer_step = (delay_optimizer_step if (delay_optimizer_step is not None) else delay_grad_averaging)
    offload_optimizer = (offload_optimizer if (offload_optimizer is not None) else (params is not None))
    allreduce_timeout = (allreduce_timeout if (allreduce_timeout is not None) else averaging_timeout)
    next_chunk_timeout = (next_chunk_timeout if (next_chunk_timeout is not None) else matchmaking_time)
    assert ((not delay_grad_averaging) or delay_optimizer_step), 'delay_grad_averaging requires delay_optimizer_step'
    assert (not (client_mode and auxiliary)), 'Client-mode peers cannot serve as auxiliaries'
    assert ((not auxiliary) or (batch_size_per_step is None)), 'Auxiliary peers should not accumulate batches'
    if (callable(optimizer) and (params is not None)):
        if ((scheduler is not None) and ((not callable(scheduler)) or isinstance(scheduler, LRSchedulerBase))):
            raise ValueError('For this mode, please provide scheduler factory: callable(optimizer) -> scheduler')
    elif all((hasattr(optimizer, attr) for attr in ('param_groups', 'step', 'zero_grad'))):
        if (offload_optimizer or delay_optimizer_step or delay_grad_averaging):
            raise ValueError('To enable offload_optimizer or delayed updates, please initialize Optimizer as hivemind.Optimizer(..., params=params, optimizer=lambda params: create_opt(params)')
    else:
        raise ValueError('Please initialize the optimizer in one of the following two ways:\n(A) hivemind.Optimizer(..., params=params, optimizer=lambda params: create_opt(params)\n(B) hivemind.Optimizer(..., optimizer=pre_initialize_optimizer)')
    if use_local_updates:
        assert (not reuse_grad_buffers), 'if local_updates is True, gradients will not be accumulated'
        assert (not delay_grad_averaging), 'if local_updates is True, gradients will not be averaged'
    (self.dht, self.run_id, self.client_mode, self.auxiliary) = (dht, run_id, client_mode, auxiliary)
    (self.batch_size_per_step, self.target_batch_size) = (batch_size_per_step, target_batch_size)
    (self.delay_state_averaging, self.average_state_every) = (delay_state_averaging, average_state_every)
    (self.matchmaking_time, self.offload_optimizer) = (matchmaking_time, offload_optimizer)
    (self.delay_grad_averaging, self.delay_optimizer_step) = (delay_grad_averaging, delay_optimizer_step)
    (self.averaging_timeout, self.allreduce_timeout) = (averaging_timeout, allreduce_timeout)
    (self.load_state_timeout, self.shutdown_timeout) = (load_state_timeout, shutdown_timeout)
    self.next_chunk_timeout = next_chunk_timeout
    self.status_loglevel = (logging.INFO if verbose else logging.DEBUG)
    self.scheduled_grads: Optional[StepControl] = None
    self.scheduled_state: Optional[StepControl] = None
    self.tracker = self._make_progress_tracker(target_batch_size, performance_ema_alpha=performance_ema_alpha, **(tracker_opts or {}))
    self.state_averager = self._make_state_averager(optimizer=optimizer, params=params, scheduler=scheduler, delta_rule_averaging=(use_local_updates and self.delay_state_averaging), compression=state_averaging_compression, state_compression=load_state_compression, average_opt_statistics=average_opt_statistics, performance_ema_alpha=performance_ema_alpha, extra_tensors=extra_tensors, **(averager_opts or {}))
    if (not use_local_updates):
        self.grad_averager = self._make_gradient_averager(reuse_grad_buffers=reuse_grad_buffers, compression=grad_compression, **(averager_opts or {}))
    else:
        self.grad_averager = None
    self._should_check_synchronization_on_update = True
    self._schema_hash = self._compute_schema_hash()
    self.delay_before_state_averaging = PerformanceEMA(alpha=performance_ema_alpha)
    self._step_supports_amp_scaling = reuse_grad_buffers
def __init__(self, *, dht: DHT, run_id: str, target_batch_size: int, batch_size_per_step: Optional[int]=None, optimizer: Union[(TorchOptimizer, OptimizerFactory)], params: Optional[Union[(Parameters, ParamGroups)]]=None, scheduler: Optional[Union[(LRSchedulerBase, SchedulerFactory)]]=None, matchmaking_time: Optional[float]=15.0, averaging_timeout: Optional[float]=60.0, allreduce_timeout: Optional[float]=None, next_chunk_timeout: Optional[float]=None, load_state_timeout: float=600.0, reuse_grad_buffers: bool=False, offload_optimizer: Optional[bool]=None, delay_optimizer_step: Optional[bool]=None, delay_grad_averaging: bool=False, delay_state_averaging: bool=True, average_state_every: int=1, use_local_updates: bool=False, client_mode: bool=None, auxiliary: bool=False, grad_compression: CompressionBase=NoCompression(), state_averaging_compression: CompressionBase=NoCompression(), load_state_compression: CompressionBase=NoCompression(), average_opt_statistics: Sequence[str]=(), extra_tensors: Sequence[torch.Tensor]=(), averager_opts: Optional[dict]=None, tracker_opts: Optional[dict]=None, performance_ema_alpha: float=0.1, shutdown_timeout: float=5, verbose: bool=False):
    self._parent_pid = os.getpid()
    client_mode = (client_mode if (client_mode is None) else dht.client_mode)
    delay_optimizer_step = (delay_optimizer_step if (delay_optimizer_step is not None) else delay_grad_averaging)
    offload_optimizer = (offload_optimizer if (offload_optimizer is not None) else (params is not None))
    allreduce_timeout = (allreduce_timeout if (allreduce_timeout is not None) else averaging_timeout)
    next_chunk_timeout = (next_chunk_timeout if (next_chunk_timeout is not None) else matchmaking_time)
    assert ((not delay_grad_averaging) or delay_optimizer_step), 'delay_grad_averaging requires delay_optimizer_step'
    assert (not (client_mode and auxiliary)), 'Client-mode peers cannot serve as auxiliaries'
    assert ((not auxiliary) or (batch_size_per_step is None)), 'Auxiliary peers should not accumulate batches'
    if (callable(optimizer) and (params is not None)):
        if ((scheduler is not None) and ((not callable(scheduler)) or isinstance(scheduler, LRSchedulerBase))):
            raise ValueError('For this mode, please provide scheduler factory: callable(optimizer) -> scheduler')
    elif all((hasattr(optimizer, attr) for attr in ('param_groups', 'step', 'zero_grad'))):
        if (offload_optimizer or delay_optimizer_step or delay_grad_averaging):
            raise ValueError('To enable offload_optimizer or delayed updates, please initialize Optimizer as hivemind.Optimizer(..., params=params, optimizer=lambda params: create_opt(params)')
    else:
        raise ValueError('Please initialize the optimizer in one of the following two ways:\n(A) hivemind.Optimizer(..., params=params, optimizer=lambda params: create_opt(params)\n(B) hivemind.Optimizer(..., optimizer=pre_initialize_optimizer)')
    if use_local_updates:
        assert (not reuse_grad_buffers), 'if local_updates is True, gradients will not be accumulated'
        assert (not delay_grad_averaging), 'if local_updates is True, gradients will not be averaged'
    (self.dht, self.run_id, self.client_mode, self.auxiliary) = (dht, run_id, client_mode, auxiliary)
    (self.batch_size_per_step, self.target_batch_size) = (batch_size_per_step, target_batch_size)
    (self.delay_state_averaging, self.average_state_every) = (delay_state_averaging, average_state_every)
    (self.matchmaking_time, self.offload_optimizer) = (matchmaking_time, offload_optimizer)
    (self.delay_grad_averaging, self.delay_optimizer_step) = (delay_grad_averaging, delay_optimizer_step)
    (self.averaging_timeout, self.allreduce_timeout) = (averaging_timeout, allreduce_timeout)
    (self.load_state_timeout, self.shutdown_timeout) = (load_state_timeout, shutdown_timeout)
    self.next_chunk_timeout = next_chunk_timeout
    self.status_loglevel = (logging.INFO if verbose else logging.DEBUG)
    self.scheduled_grads: Optional[StepControl] = None
    self.scheduled_state: Optional[StepControl] = None
    self.tracker = self._make_progress_tracker(target_batch_size, performance_ema_alpha=performance_ema_alpha, **(tracker_opts or {}))
    self.state_averager = self._make_state_averager(optimizer=optimizer, params=params, scheduler=scheduler, delta_rule_averaging=(use_local_updates and self.delay_state_averaging), compression=state_averaging_compression, state_compression=load_state_compression, average_opt_statistics=average_opt_statistics, performance_ema_alpha=performance_ema_alpha, extra_tensors=extra_tensors, **(averager_opts or {}))
    if (not use_local_updates):
        self.grad_averager = self._make_gradient_averager(reuse_grad_buffers=reuse_grad_buffers, compression=grad_compression, **(averager_opts or {}))
    else:
        self.grad_averager = None
    self._should_check_synchronization_on_update = True
    self._schema_hash = self._compute_schema_hash()
    self.delay_before_state_averaging = PerformanceEMA(alpha=performance_ema_alpha)
    self._step_supports_amp_scaling = reuse_grad_buffers