neural_tangents/utils/batch.py
def _get_jit_or_pmap_broadcast():
  """Initializes a cache of pmapped functions closed over non-`np.ndarray` args.

  Returns:
    A `jit_or_pmap_broadcast` function allowing to jit or pmap a function as a
      closure over all non-`np.ndarray` args, all `kwargs`, while broadcasting
      all `np.ndarray`s in `args` except the first one.
  """
  cache = {}

  def jit_or_pmap_broadcast(f, device_count=-1):
    """Pmap `f` over the first argument by closing over or broadcasting others.

    Args:
      f: function to pmap. First argument must be an `np.ndarray` or a Kernel.
        In either case, ndarrays should have a leading axis having the size of
        `device_count`.
      device_count: number of XLA devices. `-1` means all available devices. `0`
        means to just `jit` the function.

    Returns:
      A function of the same signature as `f` pmapped over the ndarrays in the
      first argument. Other arguments are either closed over (non-`np.ndarray`s
      in `args` and all `kwargs`) or broadcasted to
      `(device_count,) + old_shape` (for `np.ndarray`s). If `device_count == 0`,
      `f` is closed over and jitted over all non-array arguments and all
      `kwargs`.

    Raises:
      An error if `kwargs` have a `np.ndarray`.
      TODO(romann): treat `np.ndarray`s in `kwargs` when JAX allows it. See
      https://github.com/google/jax/issues/912
    """
    key = (f, device_count)

    if device_count == -1:
      device_count = xla_bridge.device_count()

    # TODO(romann): adapt this when JAX allows `axis_in` for `pmap`.
    def broadcast(arg):
      if device_count == 0:
        return arg
      # If the argument has already been sharded, no need to broadcast it.
      if isinstance(arg, ShardedDeviceArray) and arg.shape[0] == device_count:
        return arg
      return np.broadcast_to(arg, (device_count,) + arg.shape)

    def f_pmapped(x_or_kernel, *args, **kwargs):
      args_np, args_np_idxs = [], []
      args_other = {}

      is_input_kernel = isinstance(x_or_kernel, Kernel)
      x_or_kernel_np = {}
      x_or_kernel_other = {}

      if is_input_kernel:
        kernel_dict = x_or_kernel._asdict()
        for k, v in kernel_dict.items():
          if isinstance(v, np.ndarray):
            x_or_kernel_np[k] = v
          else:
            x_or_kernel_other[k] = v
      else:
        x_or_kernel_np = x_or_kernel

      # TODO(romann): treat `np.ndarray`s in `kwargs` when JAX allows it.
      # https://github.com/google/jax/issues/912
      # Filter out `np.ndarray`s from other arguments.
      for i, arg in enumerate(args):
        if _is_np_ndarray(arg):
          args_np.append(arg)
          args_np_idxs.append(i)
        else:
          args_other[i] = arg

      # Check cache before jitting.
      _key = key + \
          tuple(args_other.items()) + \
          tuple(kwargs.items()) + \
          tuple(x_or_kernel_other.items())
      if _key in cache:
        _f = cache[_key]
      else:
        # Define a `np.ndarray`-only function as a closure over other arguments.
        def _f(_x_or_kernel_np, *_args_np):
          # Merge Kernel.
          if is_input_kernel:
            _x_or_kernel_np = _merge_dicts(_x_or_kernel_np, x_or_kernel_other)
            _x_or_kernel_np = Kernel(**_x_or_kernel_np)
          # Merge args.
          _args_np = {i: _arg_np for i, _arg_np in zip(args_np_idxs, _args_np)}
          _args = _merge_dicts(_args_np, args_other)
          _args = tuple(v for k, v in sorted(_args.items()))
          return f(_x_or_kernel_np, *_args, **kwargs)

        _f = jit(_f) if device_count == 0 else pmap(_f)
        cache[_key] = _f

      # Broadcast `np.ndarray` arguments and apply the new function to them.
      args_np = tree_map(broadcast, args_np)
      return _f(x_or_kernel_np, *args_np)

    return f_pmapped

  return jit_or_pmap_broadcast
Initializes a cache of pmapped functions closed over non-`np.ndarray` args.

  Returns:
    A `jit_or_pmap_broadcast` function allowing to jit or pmap a function as a
      closure over all non-`np.ndarray` args, all `kwargs`, while broadcasting
      all `np.ndarray`s in `args` except the first one.
  
Initializes a cache of pmapped functions closed over non-`np.ndarray` args.
Initializes a cache of pmapped functions closed over non - ` np . ndarray ` args .
def _get_jit_or_pmap_broadcast():
  cache = {}
  def jit_or_pmap_broadcast(f, device_count=-1):
    key = (f, device_count)
    if device_count == -1:
      device_count = xla_bridge.device_count()
    def broadcast(arg):
      if device_count == 0:
        return arg
      if isinstance(arg, ShardedDeviceArray) and arg.shape[0] == device_count:
        return arg
      return np.broadcast_to(arg, (device_count,) + arg.shape)
    def f_pmapped(x_or_kernel, *args, **kwargs):
      args_np, args_np_idxs = [], []
      args_other = {}
      is_input_kernel = isinstance(x_or_kernel, Kernel)
      x_or_kernel_np = {}
      x_or_kernel_other = {}
      if is_input_kernel:
        kernel_dict = x_or_kernel._asdict()
        for k, v in kernel_dict.items():
          if isinstance(v, np.ndarray):
            x_or_kernel_np[k] = v
          else:
            x_or_kernel_other[k] = v
      else:
        x_or_kernel_np = x_or_kernel
      for i, arg in enumerate(args):
        if _is_np_ndarray(arg):
          args_np.append(arg)
          args_np_idxs.append(i)
        else:
          args_other[i] = arg
      _key = key + \
          tuple(args_other.items()) + \
          tuple(kwargs.items()) + \
          tuple(x_or_kernel_other.items())
      if _key in cache:
        _f = cache[_key]
      else:
        def _f(_x_or_kernel_np, *_args_np):
          if is_input_kernel:
            _x_or_kernel_np = _merge_dicts(_x_or_kernel_np, x_or_kernel_other)
            _x_or_kernel_np = Kernel(**_x_or_kernel_np)
          _args_np = {i: _arg_np for i, _arg_np in zip(args_np_idxs, _args_np)}
          _args = _merge_dicts(_args_np, args_other)
          _args = tuple(v for k, v in sorted(_args.items()))
          return f(_x_or_kernel_np, *_args, **kwargs)
        _f = jit(_f) if device_count == 0 else pmap(_f)
        cache[_key] = _f
      args_np = tree_map(broadcast, args_np)
      return _f(x_or_kernel_np, *args_np)
    return f_pmapped
  return jit_or_pmap_broadcast
def _get_jit_or_pmap_broadcast ( ) : cache = { } def jit_or_pmap_broadcast ( f , device_count = - 1 ) : """Pmap `f` over the first argument by closing over or broadcasting others.

    Args:
      f: function to pmap. First argument must be an `np.ndarray` or a Kernel.
        In either case, ndarrays should have a leading axis having the size of
        `device_count`.
      device_count: number of XLA devices. `-1` means all available devices. `0`
        means to just `jit` the function.

    Returns:
      A function of the same signature as `f` pmapped over the ndarrays in the
      first argument. Other arguments are either closed over (non-`np.ndarray`s
      in `args` and all `kwargs`) or broadcasted to
      `(device_count,) + old_shape` (for `np.ndarray`s). If `device_count == 0`,
      `f` is closed over and jitted over all non-array arguments and all
      `kwargs`.

    Raises:
      An error if `kwargs` have a `np.ndarray`.
      TODO(romann): treat `np.ndarray`s in `kwargs` when JAX allows it. See
      https://github.com/google/jax/issues/912
    """ key = ( f , device_count ) if device_count == - 1 : device_count = xla_bridge . device_count ( ) def broadcast ( arg ) : if device_count == 0 : return arg if isinstance ( arg , ShardedDeviceArray ) and arg . shape [ 0 ] == device_count : return arg return np . broadcast_to ( arg , ( device_count , ) + arg . shape ) def f_pmapped ( x_or_kernel , * args , ** kwargs ) : args_np , args_np_idxs = [ ] , [ ] args_other = { } is_input_kernel = isinstance ( x_or_kernel , Kernel ) x_or_kernel_np = { } x_or_kernel_other = { } if is_input_kernel : kernel_dict = x_or_kernel . _asdict ( ) for k , v in kernel_dict . items ( ) : if isinstance ( v , np . ndarray ) : x_or_kernel_np [ k ] = v else : x_or_kernel_other [ k ] = v else : x_or_kernel_np = x_or_kernel for i , arg in enumerate ( args ) : if _is_np_ndarray ( arg ) : args_np . append ( arg ) args_np_idxs . append ( i ) else : args_other [ i ] = arg _key = key + tuple ( args_other . items ( ) ) + tuple ( kwargs . items ( ) ) + tuple ( x_or_kernel_other . items ( ) ) if _key in cache : _f = cache [ _key ] else : def _f ( _x_or_kernel_np , * _args_np ) : if is_input_kernel : _x_or_kernel_np = _merge_dicts ( _x_or_kernel_np , x_or_kernel_other ) _x_or_kernel_np = Kernel ( ** _x_or_kernel_np ) _args_np = { i : _arg_np for i , _arg_np in zip ( args_np_idxs , _args_np ) } _args = _merge_dicts ( _args_np , args_other ) _args = tuple ( v for k , v in sorted ( _args . items ( ) ) ) return f ( _x_or_kernel_np , * _args , ** kwargs ) _f = jit ( _f ) if device_count == 0 else pmap ( _f ) cache [ _key ] = _f args_np = tree_map ( broadcast , args_np ) return _f ( x_or_kernel_np , * args_np ) return f_pmapped return jit_or_pmap_broadcast
Initializes a cache of pmapped functions closed over non-`np.ndarray` args.
Initializes a cache of pmapped functions closed over non - ` np . ndarray ` args .
"""Initializes a cache of pmapped functions closed over non-`np.ndarray` args.

  Returns:
    A `jit_or_pmap_broadcast` function allowing to jit or pmap a function as a
      closure over all non-`np.ndarray` args, all `kwargs`, while broadcasting
      all `np.ndarray`s in `args` except the first one.
  """ """Pmap `f` over the first argument by closing over or broadcasting others.

    Args:
      f: function to pmap. First argument must be an `np.ndarray` or a Kernel.
        In either case, ndarrays should have a leading axis having the size of
        `device_count`.
      device_count: number of XLA devices. `-1` means all available devices. `0`
        means to just `jit` the function.

    Returns:
      A function of the same signature as `f` pmapped over the ndarrays in the
      first argument. Other arguments are either closed over (non-`np.ndarray`s
      in `args` and all `kwargs`) or broadcasted to
      `(device_count,) + old_shape` (for `np.ndarray`s). If `device_count == 0`,
      `f` is closed over and jitted over all non-array arguments and all
      `kwargs`.

    Raises:
      An error if `kwargs` have a `np.ndarray`.
      TODO(romann): treat `np.ndarray`s in `kwargs` when JAX allows it. See
      https://github.com/google/jax/issues/912
    """ # TODO(romann): adapt this when JAX allows `axis_in` for `pmap`. # If the argument has already been sharded, no need to broadcast it. # TODO(romann): treat `np.ndarray`s in `kwargs` when JAX allows it. # https://github.com/google/jax/issues/912 # Filter out `np.ndarray`s from other arguments. # Check cache before jitting. # Define a `np.ndarray`-only function as a closure over other arguments. # Merge Kernel. # Merge args. # Broadcast `np.ndarray` arguments and apply the new function to them.
A `jit_or_pmap_broadcast` function allowing to jit or pmap a function as a
closure over all non-`np.ndarray` args, all `kwargs`, while broadcasting
all `np.ndarray`s in `args` except the first one.
A ` jit_or_pmap_broadcast ` function allowing to jit or pmap a function as a closure over all non - ` np . ndarray ` args all ` kwargs ` while broadcasting all ` np . ndarray ` s in ` args ` except the first one .