Skip to content

Commit 0053443

Browse files
igorts-gitparambole
andcommitted
Training engine: buffer metrics on the host
- MetricsRecorder keeps each step's metrics in Python lists and materializes them into 1D arrays once, at flush, instead of one jnp.append per micro-batch per metric. The flush is all-or-nothing, and pending items are keyed by buffer identity and dropped on eviction and cleanup. - update() flushes the step's metrics before the optimizer step, so an unconvertible value raises with nothing applied. - record_metrics skips intermediate_outputs and tuple/list aux values. Differences from atwigg/mlperf: - Checkpoints are taken only between steps (PR 0), so the metrics are never checkpointed: MetricsRecorder.restore_buffers, weighted_metric_bounds and the metric_bounds checkpoint metadata, the engine's _restore_metrics_buffer and the intra-step loss-cache rebuild are not ported, nor are the tests for them. - The pre-step flush is added to PR 5999's explicit update(self); MLPerf's update(self, **kwargs) is not adopted. Co-authored-by: Param Bole <parambole@google.com>
1 parent 47670fd commit 0053443

3 files changed

Lines changed: 392 additions & 28 deletions

File tree

‎src/maxtext/training_engine/maxtext_engine.py‎

Lines changed: 11 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -1812,6 +1812,11 @@ def update(self) -> int:
18121812
lr = self._learning_rate_schedule(self.train_step)
18131813
self.record_metrics("learning_rate", lr)
18141814

1815+
# Materialize the step's recorded metrics before the optimizer step: a value that cannot be
1816+
# converted then raises here, with nothing applied, instead of after the update has landed
1817+
# and before the step counter and accumulator are reset (a retry would apply it twice).
1818+
self._metrics_recorder.get_step_metrics(self.train_step)
1819+
18151820
# Wait for previous computations to finish before dispatching the update step to TPU.
18161821
self._throttler.wait_for_next()
18171822

@@ -2113,14 +2118,18 @@ def record_metrics(
21132118
aggregation_fn: Callable[[jax.Array], Any] | None = None,
21142119
mode: metrics_module.Mode = metrics_module.Mode.TRAIN,
21152120
) -> None:
2116-
"""Records a metric into the buffer, appending to JAX arrays.
2121+
"""Records a metric into the step's buffer (host-side; concatenated once at flush time).
21172122
21182123
Args:
21192124
name: The name of the metric.
21202125
metric: The metric to record.
21212126
aggregation_fn: The aggregation function to apply to the metric.
21222127
"""
2123-
if metric is None:
2128+
# `loss_fn` auxiliary outputs can contain `None` entries (disabled optional losses),
2129+
# `"intermediate_outputs"` (sow activations/mutable collections), or tuples/lists
2130+
# (e.g., `(total_loss, aux)`). Skip these so `buffer_metrics` only receives scalar
2131+
# metrics, arrays, or nested metric dicts.
2132+
if metric is None or name == "intermediate_outputs" or isinstance(metric, (tuple, list)):
21242133
return
21252134
if isinstance(metric, dict):
21262135
for sub_k, sub_v in metric.items():

‎src/maxtext/training_engine/metrics.py‎

Lines changed: 97 additions & 26 deletions
Original file line numberDiff line numberDiff line change
@@ -17,13 +17,13 @@
1717
from __future__ import annotations
1818

1919
from collections.abc import Callable
20-
import dataclasses
2120
import enum
2221
import os
2322
from typing import Any
2423

2524
from absl import logging
2625
import jax
26+
import jax.numpy as jnp
2727
from maxtext.configs import pyconfig
2828
from maxtext.training_engine import abstract_engine
2929
from maxtext.utils import max_logging, max_utils, maxtext_utils
@@ -42,11 +42,26 @@
4242
]
4343

4444
# Completed step buffers to keep resident. Each holds live device arrays and nothing on the
45-
# engine's step path removes one, so unbounded history grows HBM, and checkpoint size with
46-
# it. A window rather than Tunix's single prior step, so batched readers still work.
45+
# engine's step path removes one, so unbounded history grows HBM. A window rather than Tunix's
46+
# single prior step, so batched readers still work.
4747
_DEFAULT_MAX_BUFFERED_STEPS = 128
4848

4949

50+
def _concat_1d(items: list[Any]) -> jax.Array:
51+
"""Joins the items in order, with the layout the eager `jnp.append` path produced.
52+
53+
A single item keeps its shape (`atleast_1d`, as that path stored a step's first item); two
54+
or more are raveled and concatenated. A list of scalars, one per microbatch and the usual
55+
case, is a single `jnp.stack`: a ravel per item plus the concatenate would be one eager
56+
launch per microbatch per metric.
57+
"""
58+
if len(items) == 1:
59+
return jnp.atleast_1d(jnp.asarray(items[0]))
60+
if all(jnp.ndim(x) == 0 for x in items):
61+
return jnp.stack(items)
62+
return jnp.concatenate([jnp.ravel(jnp.asarray(x)) for x in items])
63+
64+
5065
class Mode(str, enum.Enum):
5166
TRAIN = "train"
5267
EVAL = "eval"
@@ -75,6 +90,13 @@ def __init__(self, max_buffered_steps: int = _DEFAULT_MAX_BUFFERED_STEPS, mode:
7590
as new steps start. Zero or less retains everything.
7691
"""
7792
self._metrics_buffer: list[abstract_engine.MetricsBuffer] = []
93+
# Items recorded but not yet materialized, per live buffer. Keyed by the buffer object's
94+
# identity rather than its step id: `buffer_metrics` only compares against the newest
95+
# buffer, so the history may hold two buffers with the same id (the eager `jnp.append`
96+
# code kept them apart), and they must not share a list. Every place a buffer leaves
97+
# `_metrics_buffer` drops its key, so an identity is never reused while an entry exists.
98+
self._pending_weighted: dict[int, dict[str, list[abstract_engine.WeightedMetric]]] = {}
99+
self._pending_scalar: dict[int, dict[str, list[Any]]] = {}
78100
self._mode = mode
79101
self._max_buffered_steps = max_buffered_steps
80102
self._dropped_buffer_count = 0
@@ -83,17 +105,21 @@ def buffer_metrics(
83105
self,
84106
train_step: int,
85107
name: str,
86-
metric: jax.Array | float | int | None = None,
108+
metric: abstract_engine.WeightedMetric | jax.Array | float | int | None = None,
87109
aggregation_fn: Callable[[jax.Array], Any] | None = None,
88-
):
89-
"""Buffers metrics for the given train step.
110+
) -> None:
111+
"""Buffers a metric for the given train step (host-side; no device work until flush).
90112
91113
Args:
92114
train_step: The train step for which to buffer metrics.
93115
name: The name of the metric.
94-
metric: The metric to buffer.
116+
metric: The metric to buffer. `None` is a no-op (nothing is recorded and no
117+
buffer is opened for the step).
95118
aggregation_fn: Optional aggregation function to apply to the metric.
96119
"""
120+
if metric is None:
121+
return
122+
97123
if not self._metrics_buffer or self._metrics_buffer[-1].id != train_step:
98124
new_buffer = abstract_engine.MetricsBuffer(id=train_step, mode=self._mode)
99125
self._metrics_buffer.append(new_buffer)
@@ -111,6 +137,9 @@ def _evict_old_buffers(self) -> None:
111137
return
112138
num_dropped = len(self._metrics_buffer) - self._max_buffered_steps
113139
oldest_dropped_id = self._metrics_buffer[0].id
140+
for evicted in self._metrics_buffer[:num_dropped]:
141+
self._pending_weighted.pop(id(evicted), None)
142+
self._pending_scalar.pop(id(evicted), None)
114143
del self._metrics_buffer[:num_dropped]
115144
self._dropped_buffer_count += num_dropped
116145
logging.log_every_n(
@@ -130,37 +159,64 @@ def _record_metric(
130159
metric: abstract_engine.WeightedMetric | jax.Array | float | int,
131160
aggregation_fn: Callable[[jax.Array], Any] | None = None,
132161
) -> None:
133-
"""Records a metric into the buffer, appending to JAX arrays.
162+
"""Records a metric into the step's pending Python list without launching XLA kernels.
134163
135164
Args:
136165
name: The name of the metric.
137166
metric: The metric to record.
138167
aggregation_fn: The aggregation function to apply to the metric.
139168
"""
140169
buffer = self._metrics_buffer[-1]
170+
key = id(buffer)
141171

142172
if aggregation_fn is not None:
143173
buffer.aggregation_fns[name] = aggregation_fn
144174

145175
if isinstance(metric, abstract_engine.WeightedMetric):
146-
if name not in buffer.weighted_metrics:
147-
buffer.weighted_metrics[name] = abstract_engine.WeightedMetric(
148-
unreduced_sum=jax.numpy.atleast_1d(metric.unreduced_sum),
149-
denominator=jax.numpy.atleast_1d(metric.denominator),
150-
eps=metric.eps,
151-
min_denom=metric.min_denom,
152-
)
153-
else:
154-
current_val = buffer.weighted_metrics[name]
155-
new_sum = jax.numpy.append(current_val.unreduced_sum, metric.unreduced_sum)
156-
new_denom = jax.numpy.append(current_val.denominator, metric.denominator)
157-
buffer.weighted_metrics[name] = dataclasses.replace(current_val, unreduced_sum=new_sum, denominator=new_denom)
176+
self._pending_weighted.setdefault(key, {}).setdefault(name, []).append(metric)
158177
else:
159-
if name not in buffer.scalar_metrics:
160-
buffer.scalar_metrics[name] = jax.numpy.atleast_1d(jax.numpy.asarray(metric))
161-
else:
162-
buffer.scalar_metrics[name] = jax.numpy.append(buffer.scalar_metrics[name], metric)
163-
self._metrics_buffer[-1] = buffer
178+
self._pending_scalar.setdefault(key, {}).setdefault(name, []).append(metric)
179+
180+
def _flush_buffer(self, buffer: abstract_engine.MetricsBuffer) -> None:
181+
"""Materializes pending Python lists into 1D JAX arrays on `buffer`.
182+
183+
All-or-nothing: every pending item is converted first, and `buffer` and the pending
184+
dicts are only updated once all conversions succeed. An item that `jnp.asarray` rejects
185+
therefore raises without dropping the step's other pending metrics, and a retry sees the
186+
same pending state.
187+
188+
Args:
189+
buffer: A live entry of `_metrics_buffer` whose pending items should be materialized.
190+
"""
191+
key = id(buffer)
192+
pending_w = self._pending_weighted.get(key)
193+
pending_s = self._pending_scalar.get(key)
194+
if not pending_w and not pending_s:
195+
return
196+
197+
new_weighted: dict[str, abstract_engine.WeightedMetric] = {}
198+
for name, items in (pending_w or {}).items():
199+
existing = buffer.weighted_metrics.get(name)
200+
prefix = [existing] if existing is not None else []
201+
# The live items carry the metric's eps/min_denom.
202+
template = items[0]
203+
new_weighted[name] = abstract_engine.WeightedMetric(
204+
unreduced_sum=_concat_1d([m.unreduced_sum for m in prefix + items]),
205+
denominator=_concat_1d([m.denominator for m in prefix + items]),
206+
eps=template.eps,
207+
min_denom=template.min_denom,
208+
)
209+
210+
new_scalar: dict[str, jax.Array] = {}
211+
for name, items in (pending_s or {}).items():
212+
prefix = [buffer.scalar_metrics[name]] if name in buffer.scalar_metrics else []
213+
new_scalar[name] = _concat_1d(prefix + items)
214+
215+
# Commit point: nothing above mutated `buffer` or the pending dicts.
216+
buffer.weighted_metrics.update(new_weighted)
217+
buffer.scalar_metrics.update(new_scalar)
218+
self._pending_weighted.pop(key, None)
219+
self._pending_scalar.pop(key, None)
164220

165221
def get_metrics_history(self, clear_cache: bool = True) -> list[abstract_engine.MetricsBuffer]:
166222
"""Returns every cached step buffer and optionally clears the metrics cache.
@@ -175,24 +231,39 @@ def get_metrics_history(self, clear_cache: bool = True) -> list[abstract_engine.
175231
Returns:
176232
One on-device MetricsBuffer per retained train step, oldest first.
177233
"""
234+
for buf in self._metrics_buffer:
235+
self._flush_buffer(buf)
178236
metrics_to_return = self._metrics_buffer
179237
if clear_cache:
180238
# Clear the metrics buffer.
181239
self._metrics_buffer = []
240+
self._pending_weighted.clear()
241+
self._pending_scalar.clear()
182242
return metrics_to_return
183243

184244
def get_step_metrics(self, step: int) -> abstract_engine.MetricsBuffer | None:
185-
"""Returns the latest metrics from the metrics buffer."""
245+
"""Returns the latest flushed metrics buffer if its ID matches `step`.
246+
247+
Args:
248+
step: The train step ID to retrieve.
249+
250+
Returns:
251+
The flushed `MetricsBuffer` for `step`, or `None` if the latest buffer
252+
does not match `step`.
253+
"""
186254
if not self._metrics_buffer:
187255
return None
188256
latest = self._metrics_buffer[-1]
189257
if latest.id == step:
258+
self._flush_buffer(latest)
190259
return latest
191260
return None
192261

193262
def cleanup(self) -> None:
194263
"""Cleans up the metrics recorder."""
195264
self._metrics_buffer = []
265+
self._pending_weighted.clear()
266+
self._pending_scalar.clear()
196267

197268

198269
class MetricsLogger:

0 commit comments

Comments
 (0)