Code (deepdiff/_multiprocessing.py)

  - New helpers _extract_worker_stats and _aggregate_worker_stats.
  - _distance_worker and _subtree_diff_worker now return a stats delta as a third tuple element.
  - compute_distances_parallel and compute_subtree_diffs_parallel now return (result,
  aggregated_stats) instead of bare result.

  Code (deepdiff/diff.py)
  - New stats keys WORKER_DIFF_COUNT, WORKER_PASSES_COUNT, WORKER_DISTANCE_CACHE_HIT_COUNT,
  WORKER_BATCH_COUNT added to _stats init.
  - New helper _merge_worker_stats (sums counters, OR-merges limit flags).
  - _maybe_compute_pair_distances_parallel and _dispatch_subtree_jobs unpack the new orchestrator
  return shape and merge.

  Tests
  - New classes TestWorkerStatsUnit, TestStatsKeys, TestWorkerStatsAggregationSlow (8 tests).
  - Updated TestSubtreeParallelHelper.test_empty_jobs_returns_empty_list for new return shape.
  - Updated expected_stats dicts in tests/test_cache.py (3 tests) and tests/test_ignore_order.py (2
   tests) with the four new zeroed keys.
  - Full suite: 1148 pass, 35 multiprocessing pass with --runslow.

  Doc (docs/multi_processing.md)
  - Phase 4 implementation status, code locations, test summary, and Subticket #5 removed from "Not
   yet implemented".

Author: seperman

deepdiff/_multiprocessing.py

@@ -23,6 +23,46 @@
 DEFAULT_MAX_WORKERS = 4
 DEFAULT_THRESHOLD = 64
 
+# Keys we lift out of a worker's internal _stats and ship back to the parent.
+# These mirror the same string constants used by ``deepdiff/diff.py``; we keep
+# string literals here to avoid importing diff.py at module load (which would
+# create an import cycle under spawn).
+_WORKER_STATS_COUNTER_KEYS = ('DIFF COUNT', 'PASSES COUNT', 'DISTANCE CACHE HIT COUNT')
+_WORKER_STATS_FLAG_KEYS = ('MAX PASS LIMIT REACHED', 'MAX DIFF LIMIT REACHED')
+
+
+def _extract_worker_stats(diff_instance: Any) -> Dict[str, Any]:
+    """Pull a small, picklable stats snapshot off a worker-local DeepDiff.
+
+    Returns a dict with integer counters plus boolean limit flags. Missing keys
+    are tolerated so this stays robust if ``_stats`` shrinks at the end of
+    ``__init__`` (it currently deletes ``DISTANCE CACHE ENABLED`` and the
+    ``PREVIOUS *`` bookkeeping keys before we get here).
+    """
+    stats = getattr(diff_instance, '_stats', None) or {}
+    delta: Dict[str, Any] = {}
+    for key in _WORKER_STATS_COUNTER_KEYS:
+        delta[key] = int(stats.get(key, 0) or 0)
+    for key in _WORKER_STATS_FLAG_KEYS:
+        delta[key] = bool(stats.get(key, False))
+    return delta
+
+
+def _aggregate_worker_stats(deltas: List[Dict[str, Any]]) -> Dict[str, Any]:
+    """Sum counter keys and OR-merge limit flags across worker deltas."""
+    out: Dict[str, Any] = {key: 0 for key in _WORKER_STATS_COUNTER_KEYS}
+    for key in _WORKER_STATS_FLAG_KEYS:
+        out[key] = False
+    for delta in deltas:
+        if not delta:
+            continue
+        for key in _WORKER_STATS_COUNTER_KEYS:
+            out[key] += int(delta.get(key, 0) or 0)
+        for key in _WORKER_STATS_FLAG_KEYS:
+            if delta.get(key):
+                out[key] = True
+    return out
+
 
 @dataclass(frozen=True)
 class MPConfig:
@@ -114,7 +154,9 @@ def _sanitize_parameters_for_worker(parameters: Dict[str, Any]) -> Dict[str, Any
     return sanitized
 
 
-def _distance_worker(job: Tuple[int, Dict[str, Any], Any, Any, Any, Any]) -> Tuple[int, float]:
+def _distance_worker(
+    job: Tuple[int, Dict[str, Any], Any, Any, Any, Any],
+) -> Tuple[int, float, Dict[str, Any]]:
     """Compute the rough distance between two items in a worker process.
 
     ``job`` layout matches what ``compute_distances_parallel`` ships:
@@ -123,7 +165,9 @@ def _distance_worker(job: Tuple[int, Dict[str, Any], Any, Any, Any, Any]) -> Tup
 
     The worker constructs a fresh root ``DeepDiff`` (no shared parent state),
     requests the DELTA_VIEW so we hit the same code path as the serial call in
-    ``_get_rough_distance_of_hashed_objs``, and returns the resulting float.
+    ``_get_rough_distance_of_hashed_objs``, and returns the resulting float
+    plus a ``_extract_worker_stats`` snapshot so the parent can aggregate
+    diff/pass/cache-hit counts into its WORKER_* stats keys.
     """
     # Imported here to keep module import cheap and to dodge any circular
     # import surprises under spawn.
@@ -144,7 +188,7 @@ def _distance_worker(job: Tuple[int, Dict[str, Any], Any, Any, Any, Any]) -> Tup
         # call below, hence cache_purge_level=0.
         cache_purge_level=0,
     )
-    return job_index, cast(float, diff._get_rough_distance())
+    return job_index, cast(float, diff._get_rough_distance()), _extract_worker_stats(diff)
 
 
 def compute_distances_parallel(
@@ -153,25 +197,28 @@ def compute_distances_parallel(
     original_type: Any,
     iterable_compare_func: Optional[Callable],
     config: MPConfig,
-) -> Optional[Dict[Tuple[Any, Any], float]]:
+) -> Optional[Tuple[Dict[Tuple[Any, Any], float], Dict[str, Any]]]:
     """Run ``_distance_worker`` over ``jobs`` and return distances by pair.
 
     ``jobs`` is a list of ``(added_hash, removed_hash, added_item, removed_item)``
     tuples in the exact order the serial nested loop visits them. The parent
     is responsible for that ordering; this helper does not reorder anything.
 
     Returns:
-        A dict ``{(added_hash, removed_hash): distance}``, or ``None`` if the
-        section is unsafe to parallelize (unpickleable inputs/parameters,
-        worker import error, etc.). On ``None`` the caller MUST fall back to
-        the serial path so correctness is preserved.
+        ``(distances_by_pair, aggregated_worker_stats)`` where the first item
+        is a dict ``{(added_hash, removed_hash): distance}`` and the second is
+        the aggregated ``_extract_worker_stats`` snapshot summed across all
+        workers (counter keys summed, limit flags OR-merged). Returns
+        ``None`` if the section is unsafe to parallelize (unpickleable
+        inputs/parameters, worker import error, etc.). On ``None`` the caller
+        MUST fall back to the serial path so correctness is preserved.
 
     Workers may finish out of order; we collect results into a dict keyed by
     the original job index, so callers see the same result regardless of
     completion order.
     """
     if not jobs:
-        return {}
+        return {}, _aggregate_worker_stats([])
 
     sanitized_params = _sanitize_parameters_for_worker(parameters)
 
@@ -200,14 +247,16 @@ def compute_distances_parallel(
         )
 
     results_by_index: Dict[int, float] = {}
+    stats_deltas: List[Dict[str, Any]] = []
     try:
         with ProcessPoolExecutor(max_workers=config.workers) as executor:
             futures = [executor.submit(_distance_worker, payload) for payload in payloads]
             for future in as_completed(futures):
                 # Re-raise worker exceptions in the parent so they surface as
                 # normal DeepDiff exceptions instead of being swallowed.
-                idx, distance = future.result()
+                idx, distance, stats_delta = future.result()
                 results_by_index[idx] = distance
+                stats_deltas.append(stats_delta)
     except (pickle.PicklingError, AttributeError, TypeError):
         # Pickling/spawn-related failures: surface as a serial fallback rather
         # than crashing the diff. Other exceptions (worker logic bugs, user
@@ -217,7 +266,7 @@ def compute_distances_parallel(
     out: Dict[Tuple[Any, Any], float] = {}
     for i, job in enumerate(jobs):
         out[(job[0], job[1])] = results_by_index[i]
-    return out
+    return out, _aggregate_worker_stats(stats_deltas)
 
 
 def _hash_worker(job: Tuple[int, Any, str, Dict[str, Any]]) -> Tuple[int, Optional[str]]:
@@ -256,7 +305,7 @@ def _hash_worker(job: Tuple[int, Any, str, Dict[str, Any]]) -> Tuple[int, Option
 
 def _subtree_diff_worker(
     job: Tuple[int, Dict[str, Any], Any, Any, Any],
-) -> Tuple[int, List[Tuple[str, Any]]]:
+) -> Tuple[int, List[Tuple[str, Any]], Dict[str, Any]]:
     """Run one paired-item subtree diff in a worker process.
 
     ``job`` layout: ``(job_index, sanitized_parameters, t1, t2, _original_type)``.
@@ -290,30 +339,33 @@ def _subtree_diff_worker(
             continue
         for leaf in levels:
             entries.append((report_type, leaf))
-    return job_index, entries
+    return job_index, entries, _extract_worker_stats(diff)
 
 
 def compute_subtree_diffs_parallel(
     jobs: List[Tuple[Any, Any]],
     parameters: Dict[str, Any],
     original_type: Any,
     config: MPConfig,
-) -> Optional[List[List[Tuple[str, Any]]]]:
+) -> Optional[Tuple[List[List[Tuple[str, Any]]], Dict[str, Any]]]:
     """Run ``_subtree_diff_worker`` over ``jobs`` and return per-job entries.
 
     ``jobs`` is a list of ``(t1_item, t2_item)`` tuples in the exact order
-    the serial paired-iteration code visits them. Returns a list aligned to
-    that order; each element is ``[(report_type, leaf_difflevel), ...]``
-    suitable for the parent to rebase and merge into its tree. Returns
-    ``None`` when the section is unsafe to parallelize (unpickleable
+    the serial paired-iteration code visits them. Returns
+    ``(entries_by_job, aggregated_worker_stats)`` where ``entries_by_job`` is
+    a list aligned to job order — each element is ``[(report_type,
+    leaf_difflevel), ...]`` suitable for the parent to rebase and merge into
+    its tree — and ``aggregated_worker_stats`` is the per-batch ``_stats``
+    deltas summed across workers (counters summed, limit flags OR-merged).
+    Returns ``None`` when the section is unsafe to parallelize (unpickleable
     parameters/items, worker import error). On ``None`` the caller MUST run
     the same jobs serially so correctness is preserved.
 
     Workers may finish out of order; results are collected by their original
     job index so the merge order is identical regardless of completion order.
     """
     if not jobs:
-        return []
+        return [], _aggregate_worker_stats([])
 
     sanitized_params = _sanitize_parameters_for_worker(parameters)
 
@@ -332,16 +384,21 @@ def compute_subtree_diffs_parallel(
     ]
 
     results_by_index: Dict[int, List[Tuple[str, Any]]] = {}
+    stats_deltas: List[Dict[str, Any]] = []
     try:
         with ProcessPoolExecutor(max_workers=config.workers) as executor:
             futures = [executor.submit(_subtree_diff_worker, payload) for payload in payloads]
             for future in as_completed(futures):
-                idx, entries = future.result()
+                idx, entries, stats_delta = future.result()
                 results_by_index[idx] = entries
+                stats_deltas.append(stats_delta)
     except (pickle.PicklingError, AttributeError, TypeError):
         return None
 
-    return [results_by_index[i] for i in range(len(jobs))]
+    return (
+        [results_by_index[i] for i in range(len(jobs))],
+        _aggregate_worker_stats(stats_deltas),
+    )
 
 
 def compute_hashes_parallel(

deepdiff/diff.py

@@ -86,6 +86,10 @@ def _report_progress(_stats: Dict[str, Any], progress_logger: Callable[[str], No
 DISTANCE_CACHE_ENABLED = 'DISTANCE CACHE ENABLED'
 PREVIOUS_DIFF_COUNT = 'PREVIOUS DIFF COUNT'
 PREVIOUS_DISTANCE_CACHE_HIT_COUNT = 'PREVIOUS DISTANCE CACHE HIT COUNT'
+WORKER_DIFF_COUNT = 'WORKER DIFF COUNT'
+WORKER_PASSES_COUNT = 'WORKER PASSES COUNT'
+WORKER_DISTANCE_CACHE_HIT_COUNT = 'WORKER DISTANCE CACHE HIT COUNT'
+WORKER_BATCH_COUNT = 'WORKER BATCH COUNT'
 CANT_FIND_NUMPY_MSG = 'Unable to import numpy. This must be a bug in DeepDiff since a numpy array is detected.'
 INVALID_VIEW_MSG = "view parameter must be one of 'text', 'tree', 'delta', 'colored' or 'colored_compact'. But {} was passed."
 CUTOFF_RANGE_ERROR_MSG = 'cutoff_distance_for_pairs needs to be a positive float max 1.'
@@ -340,6 +344,13 @@ def _group_by_sort_key(x):
                 MAX_PASS_LIMIT_REACHED: False,
                 MAX_DIFF_LIMIT_REACHED: False,
                 DISTANCE_CACHE_ENABLED: bool(cache_size),
+                # Multiprocessing aggregates: each parallel batch sums per-worker
+                # _stats deltas into these keys. Parent-side counters above stay
+                # comparable to a serial run so existing tests are unaffected.
+                WORKER_DIFF_COUNT: 0,
+                WORKER_PASSES_COUNT: 0,
+                WORKER_DISTANCE_CACHE_HIT_COUNT: 0,
+                WORKER_BATCH_COUNT: 0,
             }
             self.hashes = dict_() if hashes is None else hashes
             self._numpy_paths = dict_()  # if _numpy_paths is None else _numpy_paths
@@ -1350,13 +1361,38 @@ def _maybe_compute_pair_distances_parallel(
         if not mp_config.should_parallelize(len(jobs)):
             return None
 
-        return compute_distances_parallel(
+        result = compute_distances_parallel(
             jobs=jobs,
             parameters=self._parameters,
             original_type=_original_type,
             iterable_compare_func=self.iterable_compare_func,
             config=mp_config,
         )
+        if result is None:
+            return None
+        distances, worker_stats = result
+        self._merge_worker_stats(worker_stats)
+        return distances
+
+    def _merge_worker_stats(self, worker_stats):
+        """Aggregate one parallel-batch's worker ``_stats`` delta into self._stats.
+
+        Counters (DIFF / PASSES / DISTANCE CACHE HIT) sum into the matching
+        ``WORKER_*`` keys; limit flags OR-merge into the parent's existing
+        MAX_*_LIMIT_REACHED flags so any worker hitting a guard surfaces the
+        same warning state on the public ``get_stats()`` output.
+        """
+        if not worker_stats:
+            return
+        self._stats[WORKER_DIFF_COUNT] += int(worker_stats.get('DIFF COUNT', 0) or 0)
+        self._stats[WORKER_PASSES_COUNT] += int(worker_stats.get('PASSES COUNT', 0) or 0)
+        self._stats[WORKER_DISTANCE_CACHE_HIT_COUNT] += int(
+            worker_stats.get('DISTANCE CACHE HIT COUNT', 0) or 0)
+        self._stats[WORKER_BATCH_COUNT] += 1
+        if worker_stats.get(MAX_PASS_LIMIT_REACHED):
+            self._stats[MAX_PASS_LIMIT_REACHED] = True
+        if worker_stats.get(MAX_DIFF_LIMIT_REACHED):
+            self._stats[MAX_DIFF_LIMIT_REACHED] = True
 
     def _get_most_in_common_pairs_in_iterables(
             self, hashes_added, hashes_removed, t1_hashtable, t2_hashtable, parents_ids, _original_type):
@@ -1578,12 +1614,15 @@ def _dispatch_subtree_jobs(self, pending_jobs, _original_type, local_tree):
         if (mp_config is not None and mp_config.enabled
                 and mp_config.should_parallelize(len(pending_jobs))):
             jobs_payload = [(t1_item, t2_item) for (_, t1_item, t2_item, _) in pending_jobs]
-            parallel_results = compute_subtree_diffs_parallel(
+            outcome = compute_subtree_diffs_parallel(
                 jobs=jobs_payload,
                 parameters=self._parameters,
                 original_type=_original_type,
                 config=mp_config,
             )
+            if outcome is not None:
+                parallel_results, worker_stats = outcome
+                self._merge_worker_stats(worker_stats)
 
         if parallel_results is None:
             # Below threshold or unsafe inputs — run inline-equivalent serial.

docs/multi_processing.md

@@ -33,7 +33,28 @@ Without this, identity checks like `change.t2 is not notpresent` (used by
 `TextResult._from_tree_default` to decide t1-vs-t2 reporting) break on any
 DiffLevel that travels through `pickle`, which is exactly the Phase 3 path.
 
-Subtickets #5, #6 (extended matrix), and #7 are still open.
+**Phase 4 — landed (2026-04-27).** Subticket #5 (multiprocessing-aware stats)
+is implemented. Workers now return their internal `_stats` snapshot alongside
+their primary result; the parent aggregates those deltas into four new keys on
+its own `_stats` dict — `WORKER DIFF COUNT`, `WORKER PASSES COUNT`,
+`WORKER DISTANCE CACHE HIT COUNT`, and `WORKER BATCH COUNT` — and OR-merges
+worker `MAX PASS LIMIT REACHED` / `MAX DIFF LIMIT REACHED` flags into the
+parent's existing flags so any worker hitting a guard surfaces the same
+warning state on the public `get_stats()` output. Parent counters
+(`DIFF COUNT`, `PASSES COUNT`, `DISTANCE CACHE HIT COUNT`) stay scoped to the
+parent process so they remain comparable to a serial run; this is what lets
+existing stats-asserting tests pass with multiprocessing on.
+
+`max_diffs` and `max_passes` continue to act as approximate stop guards.
+Workers run their own `DeepDiff` with the same constructor params, so they
+trip the limit locally; the OR-merge means the parent's
+`MAX_*_LIMIT_REACHED` flags reflect "any worker hit it" without requiring
+exact serial-equivalent counts (which the doc explicitly does not require).
+`get_stats()` always exposes the new `WORKER_*` keys, even on serial runs,
+so consumers can read them unconditionally — they just stay zero when
+multiprocessing is off or below threshold.
+
+Subtickets #6 (extended matrix) and #7 (benchmarks) are still open.
 
 What works today:
 
@@ -61,6 +82,15 @@ What works today:
   fallback, `exclude_obj_callback` fallback, plus direct unit tests for
   `compute_subtree_diffs_parallel`). All other test files still pass
   unchanged.
+- Phase 4 adds 8 stats-aggregation tests in `tests/test_multiprocessing.py`
+  (`TestWorkerStatsUnit` for `_extract_worker_stats` / `_aggregate_worker_stats`,
+  `TestStatsKeys` for the always-present `WORKER_*` keys on serial runs, and
+  `TestWorkerStatsAggregationSlow` covering paired-subtree aggregation,
+  distance-loop aggregation, and the no-double-counting invariant). The
+  pre-existing stats-asserting tests in `tests/test_cache.py` and
+  `tests/test_ignore_order.py` were updated to include the four new zeroed
+  keys in their `expected_stats` dicts; all of them continue to pass with
+  unchanged primary counter values.
 
 Code locations:
 
@@ -89,6 +119,24 @@ Code locations:
 - `deepdiff/helper.py` — `NotPresent` / `Unprocessed` / `Skipped` /
   `NotHashed` gained `__reduce__` so the singleton sentinels survive
   `spawn`-based pickle round-trips.
+- `deepdiff/_multiprocessing.py::_extract_worker_stats`,
+  `_aggregate_worker_stats` — Phase 4 helpers. Each worker dispatch returns
+  a small picklable stats dict (`DIFF COUNT`, `PASSES COUNT`,
+  `DISTANCE CACHE HIT COUNT`, plus the two limit flags); the orchestrator
+  sums counters and OR-merges flags before handing them back.
+- `deepdiff/_multiprocessing.py::compute_distances_parallel`,
+  `compute_subtree_diffs_parallel` — both now return
+  `(primary_result, aggregated_worker_stats)` instead of just
+  `primary_result` (the `None` failure-case sentinel is unchanged).
+- `deepdiff/diff.py::DeepDiff._merge_worker_stats` — Phase 4 helper that
+  takes one orchestrator's aggregated stats dict and folds it into the
+  parent's `self._stats`. Called by both
+  `_maybe_compute_pair_distances_parallel` and `_dispatch_subtree_jobs`.
+- `deepdiff/diff.py` — four new module-level constants
+  (`WORKER_DIFF_COUNT`, `WORKER_PASSES_COUNT`,
+  `WORKER_DISTANCE_CACHE_HIT_COUNT`, `WORKER_BATCH_COUNT`) plus
+  initialization in `__init__` so the keys are always present in
+  `get_stats()`.
 
 Not yet implemented (deferred, intentional):
 
@@ -105,8 +153,6 @@ Not yet implemented (deferred, intentional):
   the current tests don't cover. Worker-side `_iterable_opcodes` are also
   not propagated, so `DELTA_VIEW` of a paired subtree containing ordered
   iterables is not yet covered by Phase 3.
-- **Subticket #5** — multiprocessing-aware stats semantics. Parent-only stats
-  remain meaningful in Phase 1, but no aggregation across workers.
 - **Subticket #6** — extended test matrix (numpy, pydantic, namedtuple, group_by,
   large-mixed structures, worker exception propagation tests). Phase 1 ships
   the core determinism harness; the rest is additive.