basic_ops.py

import functools
from typing import Tuple, Callable, List, Optional
import time
import dataclasses

import numpy as np

import jax
import jax.numpy as jnp
from jax.experimental import mesh_utils, shard_map
from jax.sharding import PositionalSharding


from jax.sharding import Mesh
from jax.sharding import PartitionSpec
from jax.sharding import NamedSharding

devices = jax.devices()
P = PartitionSpec

devices = mesh_utils.create_device_mesh((len(devices),))
mesh = Mesh(devices, axis_names=("x",))
# y = jax.device_put(x, NamedSharding(mesh, P('a', 'b')))

L = 1 << 15


@dataclasses.dataclass
class BenchmarkCase:
  """BenchmarkCase."""

  name: str
  function: Callable
  args_shape: List[Tuple]
  args_sharding: List[PartitionSpec]
  profiler_output: Optional[str] = None


start_key = jax.random.key(0)


def _new_arg(shape, dtype):
  global start_key  #  pylint: disable=all
  start_key, _ = jax.random.split(start_key)
  with jax.default_device(jax.devices("cpu")[0]):
    if dtype == jnp.int8.dtype:
      return jax.random.randint(start_key, shape, 0, 100, dtype=dtype)
    else:
      return jax.random.normal(start_key, shape, dtype=dtype) + 1


def _new_args(case, dtype):
  args = []
  for shape, sharding in zip(case.args_shape, case.args_sharding):
    arg = _new_arg(shape, dtype)
    if sharding is not None:
      arg = jax.device_put(arg, NamedSharding(mesh, sharding))
    args.append(arg)
  return args


def _run_case(case, warmup=2, runtimes=5, dtype=jnp.bfloat16.dtype):
  for _ in range(warmup):
    args = _new_args(case, dtype)
    case.function(*args)

  stamps = []
  for i in range(runtimes):
    args = _new_args(case, dtype)
    jax.block_until_ready(args)
    if case.profiler_output is not None and i == (runtimes - 1):
      jax.profiler.start_trace(case.profiler_output)
    start = time.perf_counter()
    jax.block_until_ready(case.function(*args))
    end = time.perf_counter()
    if case.profiler_output is not None and i == (runtimes - 1):
      jax.profiler.stop_trace()
    stamps.append(end - start)
  return sum(stamps) / runtimes


def _llama_ffn(x, w1, w2, w3):
  w1_res = jax.nn.silu((x @ w1).astype(jnp.bfloat16.dtype))
  w3_res = x @ w3
  res = (w1_res * w3_res) @ w2
  return res


@jax.jit
@functools.partial(
    shard_map.shard_map,
    mesh=mesh,
    in_specs=(P(), P(None, "x"), P("x"), P(None, "x")),
    out_specs=(P()),
)
def _llama_ffn_shmap(x, w1, w2, w3):
  for _ in range(3):
    x = _llama_ffn(x, w1, w2, w3)
    x = jax.lax.psum(x, "x")
  return x


@jax.jit
def _llama_ffn_spmd(x, w1, w2, w3):
  for _ in range(3):
    x = _llama_ffn(x, w1, w2, w3)
    x = jax.lax.with_sharding_constraint(x, NamedSharding(mesh, P()))
  return x


dim = 4096
multiple_of = 256
# hidden_dim = 4 * dim
# hidden_dim = int(2 * hidden_dim / 3)
# hidden_dim = multiple_of * ((hidden_dim + multiple_of - 1) // multiple_of)
hidden_dim = 11008
BATCH = 1024


@jax.jit
@functools.partial(
    shard_map.shard_map,
    mesh=mesh,
    in_specs=(P("x"),),
    out_specs=(P()),
    check_rep=False,
)
def _all_gather(x):
  return jax.lax.all_gather(x, "x")


@jax.jit
@functools.partial(
    shard_map.shard_map, mesh=mesh, in_specs=(P("x"),), out_specs=(P())
)
def _all_reduce(x):
  return jax.lax.psum(x, "x")


allcases = [
    BenchmarkCase(
        name="Matmul replicated",
        function=jax.jit(jnp.matmul),
        args_shape=((L, L), (L, L)),
        args_sharding=(P(), P()),  # replicated
    ),
    BenchmarkCase(
        name="Matmul sharded colrow",
        function=jax.jit(jnp.matmul),
        args_shape=((L, L), (L, L)),
        args_sharding=(P(None, "x"), P("x")),  # replicated
    ),
    BenchmarkCase(
        name="matmul sharded rowcol",
        function=jax.jit(jnp.matmul),
        args_shape=((L, L), (L, L)),
        args_sharding=(P("x"), P("x", None)),  # replicated
    ),
    BenchmarkCase(
        name="all_gather",
        function=_all_gather,
        args_shape=((L, L),),
        args_sharding=(P("x"),),  # replicated
    ),
    BenchmarkCase(
        name="all_reduce",
        function=_all_reduce,
        args_shape=((L, L),),
        args_sharding=(P("x"),),  # replicated
    ),
    BenchmarkCase(
        name="Llama 3xffn shardmap",
        function=_llama_ffn_shmap,
        args_shape=(
            (BATCH, dim),
            (dim, hidden_dim),
            (hidden_dim, dim),
            (dim, hidden_dim),
        ),
        args_sharding=(P(), P(None, "x"), P("x"), P(None, "x")),
    ),
    BenchmarkCase(
        name="Llama 3xffn gspmd",
        function=_llama_ffn_spmd,
        args_shape=(
            (BATCH, dim),
            (dim, hidden_dim),
            (hidden_dim, dim),
            (dim, hidden_dim),
        ),
        args_sharding=(P(), P(None, "x"), P("x"), P(None, "x")),
    ),
]


def _run_call_cases(cases):
  for dtype in (jnp.bfloat16.dtype, jnp.int8.dtype):
    for case in cases:
      avg = _run_case(case, dtype=dtype)
      dtype_size = 2 if dtype == jnp.bfloat16.dtype else 1
      input_sizes = tuple(
          [
              f"{np.prod(size) * dtype_size / (1<<20) :.6} MiB"
              for size in case.args_shape
          ]
      )
      print(
          f"{dtype} \t {case.name}: \t{avg * 1000 :.6} ms \t sizes: {input_sizes}"
      )


def main():
  print("Number of devices: ", len(devices))
  _run_call_cases(allcases)


if __name__ == "__main__":
  main()