Change MetricsAggregatorService to publisher to support live metrics without shmem

[nv-alicheng]

What does this PR do?

shmem implementation of KVStore in MetricsAggregatorService causes issues on ARM. Several solutions exist:

1. Rewrite shmem implementation in C/C++ where memory fencing primitives are exposed.
2. Restructure the design of the MetricsAggregatorService to be a ZMQ Publisher which publishes metrics at a fixed rate, which other processes, such as a TUI, can subscribe to.

This PR implements (2).

Type of change

• Bug fix
• New feature
• Documentation update
• Refactor/cleanup

Related issues

Testing

• Tests added/updated
• All tests pass locally
• Manual testing completed

Checklist

• Code follows project style
• Pre-commit hooks pass
• Documentation updated (if needed)

[github-actions]

MLCommons CLA bot All contributors have signed the MLCommons CLA ✍️ ✅

[gemini-code-assist]

Code Review

This pull request refactors the metrics aggregation system to use a registry-based architecture with HDR histograms and a generalized pub/sub transport layer, replacing the legacy mmap-backed storage. The update introduces periodic snapshot publishing with disk fallback and updates reporting logic to consume these snapshots. Feedback suggests improving encapsulation by exposing in-flight task metrics through public properties and adopting a more numerically stable variance formula for high-precision latency calculations. Additionally, several legacy tests have been skipped pending migration to the new system.

[gemini-code-assist]

Directly accessing the protected member _in_flight_tasks of MetricsTable breaks encapsulation. It is better to expose this information through a public property on the MetricsTable class.

Suggested change

	len(table._in_flight_tasks),
	len(table.in_flight_tasks_count),

[gemini-code-assist]

Directly accessing the protected member _in_flight_tasks of MetricsTable breaks encapsulation. It is better to expose this information through a public property on the MetricsTable class.

Suggested change

	logger.info("Draining %d async tasks...", len(table._in_flight_tasks))
	logger.info("Draining %d async tasks...", table.in_flight_tasks_count)

[gemini-code-assist]

Iterating over and cancelling tasks directly from the aggregator violates the encapsulation of MetricsTable. This logic should be moved into a dedicated method within MetricsTable.

                table.cancel_all_tasks()

[gemini-code-assist]

Directly accessing the protected member _in_flight_tasks of MetricsTable breaks encapsulation. It is better to expose this information through a public property on the MetricsTable class.

Suggested change

	n_pending = len(table._in_flight_tasks)
	n_pending = table.in_flight_tasks_count

[gemini-code-assist]

To improve encapsulation and support the refactored MetricsAggregatorService, please expose the count of in-flight tasks and a method to cancel them.

        self._in_flight_tasks: set[asyncio.Task] = set()

    @property
    def in_flight_tasks_count(self) -> int:
        """Number of async trigger tasks currently in flight."""
        return len(self._in_flight_tasks)

    def cancel_all_tasks(self) -> None:
        """Cancel all in-flight async trigger tasks."""
        for task in self._in_flight_tasks:
            if not task.done():
                task.cancel()

[gemini-code-assist]

The naive variance formula sum_sq - total^2 / n is numerically unstable when total^2 / n is very close to sum_sq, which is common for high-precision latency metrics in nanoseconds. Since total and sum_sq are Python integers for latency series, using the integer-based formula (n * sum_sq - total^2) / (n * (n - 1)) provides exact precision before the final square root.

Suggested change

	var_num = stat.sum_sq - stat.total * stat.total / n
	std_dev = math.sqrt(var_num / (n - 1)) if var_num > 0 else 0.0
	# Use a more numerically stable formula for variance when aggregates are large.
	# If using integers, (n * sum_sq - total^2) is exact.
	if isinstance(stat.total, int) and isinstance(stat.sum_sq, int):
	var_num = (n * stat.sum_sq - stat.total**2)
	std_dev = math.sqrt(max(0, var_num)) / math.sqrt(n * (n - 1))
	else:
	var_num = stat.sum_sq - stat.total * stat.total / n
	std_dev = math.sqrt(max(0, var_num / (n - 1)))