Queue-depth ODS counters (#5913)

Summary:
X-link: https://github.com/facebookresearch/FBGEMM/pull/2832

Pull Request resolved: #5913

Add observability for the two queue depths in the RES streaming pipeline. (1) `weights_to_stream_queue_` (MPSC stream queue in fbgemm) emits `stream_mpsc_depth` after every dequeue. (2) `tensor_to_dedup_queues_[i]` (per-shard dedup queues in training_ps) emit `dedup_depth.shard_{i}` per shard plus a `dedup_depth.total` rollup, bumped at the top of `co_stream_tensors`. Backpressure today is invisible until a trainer crashes; these signals let oncall see saturation trends and per-shard hot-spots in advance.

Both surfaces use OBC via `TrainingPsOdsLogger::bumpKeyGauge` (emits P50+P99 — the right stats for depth percentiles), on the `raw_embedding_streaming` category. The handler already owned `ods_logger_`; the fbgemm-side `RawEmbeddingStreamer` reuses the `TrainingPsOdsLogger` member added by the parent diff D107811590 (constructed only when streaming is enabled). No fb303: OBC reaches ODS via the host-level agent without per-process export config. (An earlier revision used an fb303 `ResGauges.h` helper for the fbgemm gauge; removed per review feedback on the sibling silent-failure diff.)

Stacked on D107811590 (silent-failure ODS counters), which introduces the shared OBC logger member on `RawEmbeddingStreamer`. Both feed the master observability initiative T269497764.

Design note: per-shard cardinality is N OBC keys x 3 stats; `num_dedup_threads_` is configurable but typically small (4-8); `.total` is for at-a-glance, per-shard for hot-spot debug.

Is it caused by our diff? No. Two cases, both exonerating:

  1. If the ffmpeg error is in an FBGEMM OSS-CI job (all the FAILED ones): our fbgemm code is entirely behind #ifdef FBGEMM_FBCODE, which the
  OSS build never defines — so OSS compiles a byte-identical translation unit to pre-diff. An ffmpeg build/link issue there is the OSS
  environment (the established FBGEMM OSS-CI flake), not our code.
  2. If it's in a citadel/ci_workflows WARNING (videorecs_ranking, silvertorch, etc.): those are tagged WARNING_PREEXISTING_ISSUE — CI itself
  confirms they failed on the base revision before your change. Pre-existing = not us.

  Plus the clinchers: our change is OBC counter logging (bumpKey/bumpKeyGauge) in training_ps + the RES streamer — zero source/build/runtime
  path to ffmpeg (a third-party video codec lib). And the MVAI e2e ran green on this exact binary.

  So: the FFMPEG signal is not caused by your diff — it's either OSS-CI infra or a pre-existing base-rev warning.

Reviewed By: FriedCosey

Differential Revision: D108312495

fbshipit-source-id: 81381e7ca1609c98ca13cb530d60924981335c3a

Author: adityas-meta

fbgemm_gpu/src/split_embeddings_cache/raw_embedding_streamer.cpp

@@ -191,9 +191,14 @@ RawEmbeddingStreamer::RawEmbeddingStreamer(
             tensor_stream(indices, weights, identities, runtime_meta));
 
         weights_to_stream_queue_.dequeue();
+        auto post_dequeue_depth = weights_to_stream_queue_.size();
+        if (ods_logger_) {
+          ods_logger_->bumpKeyGauge(
+              "stream_mpsc_depth", static_cast<double>(post_dequeue_depth));
+        }
         XLOG_EVERY_MS(INFO, 60000)
             << "[TBE_ID" << unique_id_
-            << "] end stream queue size: " << weights_to_stream_queue_.size()
+            << "] end stream queue size: " << post_dequeue_depth
             << " stream takes " << stop_watch.elapsed().count() << "ms";
       }
     });