Bug 1671466 - Implement IPC and Dispatcher for Memory Distributions r=janerik

Differential Revision: https://phabricator.services.mozilla.com/D94180

toolkit/components/glean/api/src/ipc.rs

@@ -23,6 +23,7 @@ use {
 pub struct IPCPayload {
     pub counters: HashMap<MetricId, i32>,
     pub events: HashMap<MetricId, Vec<(Instant, Option<HashMap<i32, String>>)>>,
+    pub memory_samples: HashMap<MetricId, Vec<u64>>,
 }
 
 /// Uniquely identifies a single metric within its metric type.

toolkit/components/glean/api/src/private/memory_distribution.rs

@@ -2,21 +2,39 @@
 // License, v. 2.0. If a copy of the MPL was not distributed with this
 // file, You can obtain one at https://mozilla.org/MPL/2.0/.
 
+use std::sync::Arc;
+
 use super::{CommonMetricData, DistributionData, MemoryUnit};
 
+use crate::dispatcher;
+use crate::ipc::{need_ipc, with_ipc_payload, MetricId};
+
 /// A memory distribution metric.
 ///
 /// Memory distributions are used to accumulate and store memory measurements for analyzing distributions of the memory data.
 #[derive(Debug)]
-pub struct MemoryDistributionMetric {
+pub enum MemoryDistributionMetric {
+    Parent(Arc<MemoryDistributionMetricImpl>),
+    Child(MemoryDistributionMetricIpc),
+}
+#[derive(Debug)]
+pub struct MemoryDistributionMetricImpl {
     inner: glean_core::metrics::MemoryDistributionMetric,
 }
+#[derive(Debug)]
+pub struct MemoryDistributionMetricIpc(MetricId);
 
 impl MemoryDistributionMetric {
     /// Create a new memory distribution metric.
     pub fn new(meta: CommonMetricData, memory_unit: MemoryUnit) -> Self {
-        let inner = glean_core::metrics::MemoryDistributionMetric::new(meta, memory_unit);
-        Self { inner }
+        if need_ipc() {
+            MemoryDistributionMetric::Child(MemoryDistributionMetricIpc(MetricId::new(meta)))
+        } else {
+            MemoryDistributionMetric::Parent(Arc::new(MemoryDistributionMetricImpl::new(
+                meta,
+                memory_unit,
+            )))
+        }
     }
 
     /// Accumulates the provided sample in the metric.
@@ -31,7 +49,23 @@ impl MemoryDistributionMetric {
     /// Values bigger than 1 Terabyte (2<sup>40</sup> bytes) are truncated
     /// and an `ErrorType::InvalidValue` error is recorded.
     pub fn accumulate(&self, sample: u64) {
-        crate::with_glean(|glean| self.inner.accumulate(glean, sample))
+        match self {
+            MemoryDistributionMetric::Parent(p) => {
+                let metric = Arc::clone(&p);
+                dispatcher::launch(move || metric.accumulate(sample));
+            }
+            MemoryDistributionMetric::Child(c) => {
+                with_ipc_payload(move |payload| {
+                    if let Some(v) = payload.memory_samples.get_mut(&c.0) {
+                        v.push(sample);
+                    } else {
+                        let mut v = vec![];
+                        v.push(sample);
+                        payload.memory_samples.insert(c.0.clone(), v);
+                    }
+                });
+            }
+        }
     }
 
     /// **Test-only API.**
@@ -46,6 +80,30 @@ impl MemoryDistributionMetric {
     /// ## Return value
     ///
     /// Returns the stored value or `None` if nothing stored.
+    pub fn test_get_value(&self, storage_name: &str) -> Option<DistributionData> {
+        match self {
+            MemoryDistributionMetric::Parent(p) => {
+                dispatcher::block_on_queue();
+                p.test_get_value(storage_name)
+            }
+            MemoryDistributionMetric::Child(_c) => panic!(
+                "Cannot get test value for {:?} in non-parent process!",
+                self
+            ),
+        }
+    }
+}
+
+impl MemoryDistributionMetricImpl {
+    pub fn new(meta: CommonMetricData, memory_unit: MemoryUnit) -> Self {
+        let inner = glean_core::metrics::MemoryDistributionMetric::new(meta, memory_unit);
+        Self { inner }
+    }
+
+    pub fn accumulate(&self, sample: u64) {
+        crate::with_glean(|glean| self.inner.accumulate(glean, sample))
+    }
+
     pub fn test_get_value(&self, storage_name: &str) -> Option<DistributionData> {
         crate::with_glean(move |glean| self.inner.test_get_value(glean, storage_name))
     }

toolkit/components/glean/api/tests/memory_distribution.rs

@@ -5,6 +5,7 @@
 mod common;
 use common::*;
 
+use fog::ipc;
 use fog::private::{CommonMetricData, Lifetime, MemoryDistributionMetric, MemoryUnit};
 
 #[test]
@@ -32,3 +33,40 @@ fn smoke_test_memory_distribution() {
     assert_eq!(0, metric_data.values[&44376]);
     assert_eq!(43008, metric_data.sum);
 }
+
+#[test]
+fn memory_distribution_child() {
+    let _lock = lock_test();
+    let _t = setup_glean(None);
+
+    let store_names: Vec<String> = vec!["store1".into()];
+    let meta = CommonMetricData {
+        name: "memory_distribution_metric".into(),
+        category: "telemetry".into(),
+        send_in_pings: store_names.clone(),
+        disabled: false,
+        lifetime: Lifetime::Ping,
+        ..Default::default()
+    };
+    let parent_metric = MemoryDistributionMetric::new(meta.clone(), MemoryUnit::Kilobyte);
+
+    parent_metric.accumulate(42);
+
+    {
+        // scope for need_ipc RAII
+        let _raii = ipc::test_set_need_ipc(true);
+        let child_metric = MemoryDistributionMetric::new(meta.clone(), MemoryUnit::Kilobyte);
+
+        child_metric.accumulate(13 * 9);
+    }
+
+    let metric_data = parent_metric.test_get_value("store1").unwrap();
+    assert_eq!(1, metric_data.values[&42494]);
+    assert_eq!(0, metric_data.values[&44376]);
+    assert_eq!(43008, metric_data.sum);
+    // TODO: implement replay. See bug 1646165.
+    // For now, let's ensure there's something in the buffer and replay doesn't error.
+    let buf = ipc::take_buf().unwrap();
+    assert!(buf.len() > 0);
+    assert!(ipc::replay_from_buf(&buf).is_ok());
+}