@@ -16,7 +16,9 @@
     python -m pytest backends/cuda/tests/test_int4_quantized_gemm.py -v
 """
 
+import concurrent.futures
 import statistics
+import tempfile
 import unittest
 from unittest import mock
 
@@ -183,6 +185,78 @@ def counting_pool():
         self.assertFalse(quantized_gemm_family._COMPILED_AHEAD)
         _check_close(self, out, _dequant_matmul(x, *weights, GROUP_SIZE))
 
+    def _cache_hits(self, bucket: int):
+        """Cache hits and misses of the bucket's autotuned kernel in this
+        process (its helper kernels, such as the split-K reduce, are not
+        autotuned and compile here as before)."""
+        name = int4_kernel._BUCKET_KERNELS[bucket].fn.fn.__name__
+        hits = {"hit": 0, "miss": 0}
+
+        def listener(src, metadata, metadata_group, times, cache_hit):
+            if src.name == name:
+                hits["hit" if cache_hit else "miss"] += 1
+
+        return hits, mock.patch.object(triton.knobs.compilation, "listener", listener)
+
+    def test_configs_compile_in_worker_processes(self) -> None:
+        # A shape no other test tunes; with an empty Triton cache, every config
+        # this process loads is one a worker compiled into the shared cache.
+        n, k, bucket = 448, 1792, 2
+        weights = _packed(n, k, seed=13)
+        x = torch.randn(bucket, k, dtype=torch.bfloat16, device="cuda")
+        hits, patch = self._cache_hits(bucket)
+        with tempfile.TemporaryDirectory() as cache_dir, mock.patch.object(
+            triton.knobs.cache, "dir", cache_dir
+        ), patch:
+            out = int4_kernel._launch(bucket, x, *weights, GROUP_SIZE)
+        self.assertIsNotNone(quantized_gemm_family._compile_process_pool())
+        self.assertGreater(hits["hit"], 1)
+        self.assertEqual(hits["miss"], 0)
+        _check_close(self, out, _dequant_matmul(x, *weights, GROUP_SIZE))
+
+    def test_workers_that_cannot_import_the_kernel_fall_back_to_this_process(
+        self,
+    ) -> None:
+        n, k, bucket = 448, 2816, 2
+        weights = _packed(n, k, seed=17)
+        x = torch.randn(bucket, k, dtype=torch.bfloat16, device="cuda")
+        request = quantized_gemm_family._process_compile_request
+
+        def unimportable(compile_fn):
+            args = request(compile_fn)
+            return None if args is None else (args[0], "no_such_module", *args[2:])
+
+        hits, patch = self._cache_hits(bucket)
+        # An empty Triton cache: what this process compiles shows up as misses.
+        with tempfile.TemporaryDirectory() as cache_dir, mock.patch.object(
+            triton.knobs.cache, "dir", cache_dir
+        ), mock.patch.object(
+            quantized_gemm_family, "_process_compile_request", unimportable
+        ), mock.patch.object(
+            quantized_gemm_family, "_process_pool_failed", False
+        ), mock.patch.object(
+            quantized_gemm_family, "_process_pool", None
+        ), patch:
+            out = int4_kernel._launch(bucket, x, *weights, GROUP_SIZE)
+            self.assertTrue(quantized_gemm_family._process_pool_failed)
+        # Compiled here instead, and still correct.
+        self.assertGreater(hits["miss"], 0)
+        _check_close(self, out, _dequant_matmul(x, *weights, GROUP_SIZE))
+
+    def test_only_unusable_workers_stop_process_compiles(self) -> None:
+        from torch._inductor.compile_worker.subproc_pool import SubprocException
+
+        failed = quantized_gemm_family._workers_failed
+        # One config failing to compile in a worker keeps the workers.
+        self.assertFalse(failed(error=SubprocException("ptxas error")))
+        self.assertFalse(failed(result=None))
+        # The kernel cannot be imported there, or the pool closed or broke.
+        self.assertTrue(failed(result=quantized_gemm_family._UNIMPORTABLE))
+        self.assertTrue(failed(error=RuntimeError("SubprocPool closed")))
+        self.assertTrue(
+            failed(error=concurrent.futures.process.BrokenProcessPool("died"))
+        )
+
     def test_fake_tensors_are_not_precompiled(self) -> None:
         # A full launch of a shape no other test tunes, so only the fake-tensor
         # check can keep it off the pool: the same launch on real tensors uses it.
 
@@ -27,6 +27,7 @@
 import concurrent.futures
 import contextlib
 import contextvars
+import importlib
 import inspect
 import logging
 import os
@@ -47,8 +48,9 @@
 ):  # A Triton without parallel compile compiles each config on first use.
     _async_compile = None
 
-# Threads that compile one kernel's autotune configs. Compiling is mostly ptxas
-# and LLVM, which run outside the GIL.
+# Concurrent compiles of autotune configs. Each runs in a worker process: in
+# one process Triton's MLIR and LLVM passes hold the GIL, so threads alone keep
+# only a few compiles busy.
 _MAX_COMPILE_THREADS = 32
 
 logger: logging.Logger = logging.getLogger(__name__)
@@ -181,6 +183,169 @@ def _has_untuned_configs(kernel, args: Sequence, kwargs: dict) -> bool:
 )
 
 
+def _compile_workers() -> int:
+    return min(_MAX_COMPILE_THREADS, os.cpu_count() or 1)
+
+
+_PROCESS_POOL_LOCK = threading.Lock()
+_process_pool = None
+_process_pool_failed = False
+
+
+def _compile_process_pool():
+    """The worker processes, started on first use and kept for later shapes;
+    None once they have failed. Inductor's ``SubprocPool``: a fresh Python
+    process that forks the workers, so they neither re-run the caller's main
+    module (as ``spawn`` would) nor inherit its CUDA state (as ``fork`` would)."""
+    global _process_pool
+    with _PROCESS_POOL_LOCK:
+        if _process_pool is None and not _process_pool_failed:
+            from torch._inductor.compile_worker.subproc_pool import (
+                SubprocKind,
+                SubprocPool,
+            )
+
+            _process_pool = SubprocPool(_compile_workers(), kind=SubprocKind.FORK)
+        return _process_pool
+
+
+def _stop_using_processes(error: BaseException) -> None:
+    global _process_pool, _process_pool_failed
+    with _PROCESS_POOL_LOCK:
+        if _process_pool_failed:
+            return
+        _process_pool_failed = True
+        pool, _process_pool = _process_pool, None
+    logger.warning(
+        "Compiling quantized GEMM configs in this process from now on: worker "
+        "processes failed (%s)",
+        error,
+    )
+    if pool is not None:
+        pool.shutdown()
+
+
+def _compile_in_process(
+    cache_dir: str,
+    module: str,
+    qualname: str,
+    source_module: str,
+    source_qualname: str,
+    signature: dict,
+    constants: dict,
+    attrs: dict,
+    target,
+    options: dict,
+    env_vars: dict,
+) -> None:
+    """Compiles one config in a worker process into the shared Triton cache.
+    Returns ``_UNIMPORTABLE`` if the worker cannot find the kernel or source
+    class, which no other config would find either."""
+    # Compiling needs no GPU; keep the worker from initializing one.
+    os.environ["CUDA_VISIBLE_DEVICES"] = ""
+    triton.knobs.cache.dir = cache_dir
+    try:
+        fn = importlib.import_module(module)
+        for part in qualname.split("."):
+            fn = getattr(fn, part)
+        while not isinstance(fn, triton.runtime.JITFunction):
+            fn = fn.fn
+        source = importlib.import_module(source_module)
+        for part in source_qualname.split("."):
+            source = getattr(source, part)
+    except (ImportError, AttributeError):
+        return _UNIMPORTABLE
+    triton.compile(
+        source(fn, signature, constants, attrs),
+        target=target,
+        options=options,
+        _env_vars=env_vars,
+    )
+    return None
+
+
+def _process_compile_request(compile_fn) -> Optional[tuple]:
+    """The arguments of the ``triton.compile`` call that Triton's async-compile
+    closure makes, so a worker process can make the same call. None when the
+    closure or kernel is not the expected shape (it then compiles in-process)."""
+    names = compile_fn.__code__.co_freevars
+    if set(names) != {"env_vars", "options", "self", "src", "target"}:
+        return None
+    cells = dict(zip(names, (cell.cell_contents for cell in compile_fn.__closure__)))
+    src = cells["src"]
+    jit_fn = getattr(src, "fn", None)
+    module = getattr(jit_fn, "module", None)
+    qualname = getattr(getattr(jit_fn, "fn", None), "__qualname__", "")
+    if module is None or not qualname or "<locals>" in qualname:
+        return None
+    return (
+        triton.knobs.cache.dir,
+        module,
+        qualname,
+        type(src).__module__,
+        type(src).__qualname__,
+        src.signature,
+        src.constants,
+        src.attrs,
+        cells["target"],
+        cells["options"].__dict__,
+        cells["env_vars"],
+    )
+
+
+_UNIMPORTABLE = "unimportable"
+
+
+def _workers_failed(result=None, error: Optional[BaseException] = None) -> bool:
+    """Whether a worker's answer means the workers themselves are unusable, as
+    opposed to one config failing to compile (which ``SubprocPool`` reports as
+    a ``SubprocException``): they cannot import the kernel, or the pool closed
+    or broke."""
+    from torch._inductor.compile_worker.subproc_pool import SubprocException
+
+    if error is not None:
+        return not isinstance(error, SubprocException)
+    return result == _UNIMPORTABLE
+
+
+def _compiled_in_a_process_first(compile_fn):
+    request = _process_compile_request(compile_fn)
+    if request is None:
+        return compile_fn
+
+    def compile_then_load():
+        pool = _compile_process_pool()
+        if pool is not None:
+            try:
+                result = pool.submit(_compile_in_process, *request).result()
+            except Exception as error:
+                if _workers_failed(error=error):
+                    _stop_using_processes(error)
+                # Otherwise this config also fails to compile in-process, where
+                # the benchmark skips it as before.
+            else:
+                if _workers_failed(result=result):
+                    _stop_using_processes(
+                        RuntimeError(f"workers cannot import {request[1]}")
+                    )
+        # Loads what the worker cached (same source, options and key), or
+        # compiles here if it did not.
+        return compile_fn()
+
+    return compile_then_load
+
+
+if _async_compile is not None:
+
+    class _ProcessCompileMode(_async_compile.AsyncCompileMode):
+        """Triton's async compile, with each compile done in a worker process."""
+
+        def submit(self, key, compile_fn, finalize_fn):
+            return super().submit(
+                key, _compiled_in_a_process_first(compile_fn), finalize_fn
+            )
+
+
 _COMPILE_THREADS_LOCK = threading.Lock()
 _compile_thread_pool: Optional[concurrent.futures.ThreadPoolExecutor] = None
 
@@ -191,15 +356,15 @@ def _compile_threads() -> concurrent.futures.ThreadPoolExecutor:
     with _COMPILE_THREADS_LOCK:
         if _compile_thread_pool is None:
             _compile_thread_pool = concurrent.futures.ThreadPoolExecutor(
-                min(_MAX_COMPILE_THREADS, os.cpu_count() or 1),
+                _compile_workers(),
                 thread_name_prefix="quantized_gemm_compile",
             )
         return _compile_thread_pool
 
 
 @contextlib.contextmanager
 def _compile_pool():
-    with _async_compile.AsyncCompileMode(_compile_threads(), ignore_errors=True):
+    with _ProcessCompileMode(_compile_threads(), ignore_errors=True):
         yield