Run submission runner tests in CI

.github/workflows/CI.yml

@@ -204,6 +204,7 @@ jobs:
         pytest -vx tests/test_param_shapes.py
         pytest -vx tests/test_param_types.py
         pytest -vx tests/test_ssim.py
+        pytest -vx tests/submission_runner_test.py
   pytest-baselines:
     runs-on: ubuntu-latest
     steps:

algorithms/development_algorithms/mnist/mnist_jax/submission.py

@@ -1,15 +1,12 @@
 """Training algorithm track submission functions for MNIST."""
 
-import functools
 from typing import Any, Dict, Iterator, List, Optional, Tuple
 
 import jax
 import jax.numpy as jnp
 import optax
-from flax import jax_utils
-from jax import lax
 
-from algoperf import spec
+from algoperf import jax_sharding_utils, spec
 
 
 def get_batch_size(workload_name):
@@ -39,29 +36,18 @@ def init_optimizer_state(
     ),
     optax.scale(-hyperparameters.learning_rate),
   )
-  return jax_utils.replicate(opt_init_fn(params_zeros_like)), opt_update_fn
-
-
-# We need to jax.pmap here instead of inside update_params because the latter
-# would recompile the function every step.
-@functools.partial(
-  jax.pmap,
-  axis_name='batch',
-  in_axes=(None, None, 0, 0, None, 0, 0, 0),
-  static_broadcasted_argnums=(0, 1),
-)
-def pmapped_update_params(
+  return opt_init_fn(params_zeros_like), opt_update_fn
+
+
+def train_step(
   workload: spec.Workload,
   opt_update_fn,
   current_param_container: spec.ParameterContainer,
   model_state: spec.ModelAuxiliaryState,
-  hyperparameters: spec.Hyperparameters,
   batch: Dict[str, spec.Tensor],
   optimizer_state: spec.OptimizerState,
   rng: spec.RandomState,
 ) -> spec.UpdateReturn:
-  del hyperparameters
-
   def loss_fn(params):
     logits_batch, new_model_state = workload.model_fn(
       params=params,
@@ -77,7 +63,6 @@ def loss_fn(params):
 
   grad_fn = jax.value_and_grad(loss_fn, has_aux=True)
   (_, new_model_state), grad = grad_fn(current_param_container)
-  grad = lax.pmean(grad, axis_name='batch')
   updates, new_optimizer_state = opt_update_fn(
     grad, optimizer_state, current_param_container
   )
@@ -105,18 +90,33 @@ def update_params(
   del train_state
   del eval_results
   del global_step
+  del hyperparameters
 
-  per_device_rngs = jax.random.split(rng, jax.local_device_count())
   optimizer_state, opt_update_fn = optimizer_state
-  new_optimizer_state, updated_params, new_model_state = pmapped_update_params(
+
+  replicated = jax_sharding_utils.get_replicate_sharding()
+  sharded = jax_sharding_utils.get_batch_dim_sharding()
+  jitted_train_step = jax.jit(
+    train_step,
+    static_argnums=(0, 1),
+    donate_argnums=(2, 3, 5),
+    in_shardings=(
+      replicated,
+      replicated,
+      sharded,
+      replicated,
+      replicated,
+    ),
+    out_shardings=(replicated, replicated, replicated),
+  )
+  new_optimizer_state, updated_params, new_model_state = jitted_train_step(
     workload,
     opt_update_fn,
     current_param_container,
     model_state,
-    hyperparameters,
     batch,
     optimizer_state,
-    per_device_rngs,
+    rng,
   )
   return (new_optimizer_state, opt_update_fn), updated_params, new_model_state
 

pyproject.toml

@@ -40,6 +40,7 @@ dependencies = [
   "pandas>=2.0.1",
   "tensorflow==2.19.0",
   "tensorflow-datasets==4.9.9",
+  "tensorflow-metadata==1.16.1",
   "tensorflow-probability==0.20.0",
   "gputil==1.4.0",
   "psutil==6.1.0",

tests/submission_runner_test.py

@@ -14,13 +14,14 @@
 
 import submission_runner
 from algoperf.profiler import PassThroughProfiler
+from algoperf.workloads import workloads
 
 FLAGS = flags.FLAGS
 # Needed to avoid UnparsedFlagAccessError
 # (see https://github.com/google/model_search/pull/8).
-FLAGS(sys.argv)
+FLAGS([sys.argv[0]])
 
-_MNIST_DEV_ALGO_DIR = 'reference_algorithms/development_algorithms/mnist'
+_MNIST_DEV_ALGO_DIR = 'algorithms/development_algorithms/mnist'
 
 
 class SubmissionRunnerTest(parameterized.TestCase):
@@ -46,13 +47,14 @@ def test_submission(
     self, workload, framework, submission_path, tuning_search_space
   ):
     FLAGS.framework = framework
-    workload_metadata = copy.deepcopy(submission_runner.WORKLOADS[workload])
+    FLAGS.torch_compile = False
+    workload_metadata = copy.deepcopy(workloads.WORKLOADS[workload])
     workload_metadata['workload_path'] = os.path.join(
-      submission_runner.BASE_WORKLOADS_DIR,
+      workloads.BASE_WORKLOADS_DIR,
       workload_metadata['workload_path'] + '_' + framework,
       'workload.py',
     )
-    workload_obj = submission_runner.import_workload(
+    workload_obj = workloads.import_workload(
       workload_path=workload_metadata['workload_path'],
       workload_class_name=workload_metadata['workload_class_name'],
       workload_init_kwargs={},
@@ -73,7 +75,7 @@ def test_submission(
   def test_convert_filepath_to_module(self):
     """Sample test for the `convert_filepath_to_module` function."""
     test_path = os.path.abspath(__file__)
-    module_path = submission_runner.convert_filepath_to_module(test_path)
+    module_path = workloads.convert_filepath_to_module(test_path)
     self.assertNotIn('.py', module_path)
     self.assertNotIn('/', module_path)
     self.assertIsInstance(module_path, str)