Queue-depth ODS counters (#5913)

Summary:
X-link: facebookresearch/FBGEMM#2832


Add observability for the two queue depths in the RES streaming pipeline. (1) `weights_to_stream_queue_` (MPSC stream queue in fbgemm) emits `res.queue.stream_mpsc_depth` after every dequeue. (2) `tensor_to_dedup_queues_[i]` (per-shard dedup queues in training_ps) emit `res.queue.dedup_depth.shard_{i}` per shard plus a `res.queue.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::bumpKeyLatency` (emits AVG+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.

Differential Revision: D108312495

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_->bumpKeyLatency(
+              "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";
       }
     });