Adding support for Pathways proxy

axlearn/cloud/gcp/job.py

@@ -5,6 +5,7 @@
 Note that these utilities do not handle resource management.
 """
 import atexit
+import importlib
 import io
 import logging
 import math
@@ -436,6 +437,7 @@ class Config(GKEJob.Config):
         enable_tpu_ici_resiliency: Optional[bool] = None
         location_hint: Optional[str] = None
         enable_tpu_smart_repair: bool = False
+        use_pathways: Optional[bool] = False
 
     @classmethod
     def define_flags(cls, fv: flags.FlagValues):
@@ -450,6 +452,9 @@ def define_flags(cls, fv: flags.FlagValues):
             "not all TPU types support this flag.",
             **common_kwargs,
         )
+        flags.DEFINE_boolean(
+            "use_pathways", False, "Wether the workload is pathways-enabled.", **common_kwargs
+        )
 
     @classmethod
     def from_flags(cls, fv: flags.FlagValues, **kwargs) -> Config:
@@ -473,6 +478,14 @@ def __init__(self, cfg: Config):
             raise NotImplementedError(f"Missing system characteristics for {self._tpu_type}")
         super().__init__(cfg)
         self._output_volume_mount = dict(name="shared-output", mountPath="/output")
+        if cfg.use_pathways:
+            self._import_pathways()
+
+    def _import_pathways(self):
+        try:
+            importlib.import_module("pathwaysutils")
+        except ModuleNotFoundError:
+            logging.error("An error occurred while importing pathways-utils.")
 
     def _maybe_add_volume_mount(self, volume_mounts: list[dict], *, spec: Optional[VolumeMount]):
         if spec:

axlearn/common/launch.py

@@ -103,7 +103,9 @@ def setup():
     logging.info("Devices: %s", devices)
     local_devices = jax.local_devices()
     logging.info("Local Devices: %s", local_devices)
-    if not devices or not all(device.platform == FLAGS.jax_backend for device in devices):
+    if FLAGS.jax_backend != "proxy" and (
+        not devices or not all(device.platform == FLAGS.jax_backend for device in devices)
+    ):
         raise RuntimeError(f"Expected backend {FLAGS.jax_backend}. Got {devices}.")
     if FLAGS.data_dir:
         # TODO(ruoming): Get rid of --data_dir and use only env var DATA_DIR.

axlearn/common/utils_spmd.py

@@ -53,7 +53,8 @@ def setup(
         if initialization_timeout is not None:
             init_kwargs["initialization_timeout"] = initialization_timeout
 
-        if jax_backend == "tpu":
+        # TPU resources orchestrated by Pathways use 'proxy' as the JAX backend
+        if jax_backend in ("tpu", "proxy"):
             if not (
                 distributed_coordinator is None and num_processes is None and process_id is None
             ):
@@ -115,5 +116,6 @@ def setup(
                     f"({initialization_timeout} seconds)."
                 )
         else:
-            jax.distributed.initialize(**init_kwargs)
-            _jax_distributed_initialized = True
+            if jax_backend != "proxy":
+                jax.distributed.initialize(**init_kwargs)
+                _jax_distributed_initialized = True

pyproject.toml

@@ -155,6 +155,11 @@ audio = [
     "levenshtein==0.25.1",
 ]
 
+# Pathways utilities.
+pathways = [
+    "pathwaysutils==0.0.7", # for JAX+Pathways single-controller accelerator coordinator
+]
+
 [tool.flit.module]
 # This defines the import name. https://flit.pypa.io/en/stable/pyproject_toml.html#module-section
 name = "axlearn"