Measure space allocated for data

Author: Cecca

Cargo.lock

@@ -20,6 +20,7 @@ bincode = "1.0"
 packed_simd = { version = "0.3.4", package = "packed_simd_2" }
 lazy_static = "1.3"
 statrs = "0.11"
+deepsize = "0.2"
 
 [dev-dependencies]
 probabilistic-collections = "0.5"

danny-base/Cargo.toml

@@ -3,6 +3,7 @@ use crate::types::*;
 use rand::distributions::{Distribution, Normal, Uniform};
 use rand::Rng;
 use std::clone::Clone;
+use deepsize::DeepSizeOf;
 
 pub trait LSHFunction {
     type Input;
@@ -18,7 +19,7 @@ pub trait LSHFunction {
     }
 }
 
-#[derive(Clone, Abomonation, Debug, Hash)]
+#[derive(Clone, Abomonation, Debug, Hash, DeepSizeOf)]
 pub struct TensorPool {
     left: Vec<u16>,
     right: Vec<u16>,

danny-base/src/lsh.rs

@@ -1,3 +1,4 @@
+use deepsize::DeepSizeOf;
 use crate::lsh::*;
 use crate::types::*;
 use packed_simd::u64x2;
@@ -15,7 +16,7 @@ pub trait Sketcher {
     fn sketch(&self, v: &Self::Input) -> Self::Output;
 }
 
-pub trait BitBasedSketch: Clone + Copy {
+pub trait BitBasedSketch: Clone + Copy + DeepSizeOf {
     fn different_bits(&self, other: &Self) -> u32;
     fn same_bits(&self, other: &Self) -> u32;
     fn num_bits(&self) -> usize;
@@ -32,7 +33,7 @@ pub trait FromCosine: Sized {
 }
 
 /// A 0-bits sketch
-#[derive(Debug, Clone, Copy, Abomonation, Hash, Eq, PartialEq)]
+#[derive(Debug, Clone, Copy, Abomonation, Hash, Eq, PartialEq, DeepSizeOf)]
 pub struct Sketch0;
 
 impl BitBasedSketch for Sketch0 {
@@ -84,7 +85,7 @@ where
     }
 }
 
-#[derive(Debug, Clone, Copy, Abomonation, Hash, Eq, PartialEq)]
+#[derive(Debug, Clone, Copy, Abomonation, Hash, Eq, PartialEq, DeepSizeOf)]
 pub struct Sketch64 {
     data: u64,
 }
@@ -142,7 +143,7 @@ where
     }
 }
 
-#[derive(Debug, Clone, Copy, Abomonation, Hash, Eq, PartialEq)]
+#[derive(Debug, Clone, Copy, Abomonation, Hash, Eq, PartialEq, DeepSizeOf)]
 pub struct Sketch128 {
     data: [u64; 2],
 }
@@ -270,7 +271,7 @@ where
     }
 }
 
-#[derive(Debug, Clone, Copy, Abomonation, Hash, Eq, PartialEq)]
+#[derive(Debug, Clone, Copy, Abomonation, Hash, Eq, PartialEq, DeepSizeOf)]
 pub struct Sketch256 {
     data: [u64; 4],
 }
@@ -357,7 +358,7 @@ where
     }
 }
 
-#[derive(Debug, Clone, Copy, Abomonation, Hash, Eq, PartialEq)]
+#[derive(Debug, Clone, Copy, Abomonation, Hash, Eq, PartialEq, DeepSizeOf)]
 pub struct Sketch512 {
     data: [u64; 8],
 }

danny-base/src/sketch.rs

@@ -6,8 +6,9 @@ use std::collections::BTreeSet;
 use std::collections::HashMap;
 use std::fmt::Debug;
 use std::hash::Hash;
+use deepsize::DeepSizeOf;
 
-#[derive(Clone, Default, Eq, Ord, Hash, PartialEq, PartialOrd, Abomonation, Copy, Debug)]
+#[derive(Clone, Default, Eq, Ord, Hash, PartialEq, PartialOrd, Abomonation, Copy, Debug, DeepSizeOf)]
 pub struct ElementId(pub u32);
 
 impl Into<u64> for ElementId {
@@ -16,7 +17,7 @@ impl Into<u64> for ElementId {
     }
 }
 
-#[derive(Serialize, Deserialize, Clone, Abomonation, Default)]
+#[derive(Serialize, Deserialize, Clone, Abomonation, Default, DeepSizeOf)]
 pub struct Vector {
     data: Vec<f32>,
 }
@@ -68,7 +69,7 @@ impl Vector {
     }
 }
 
-#[derive(Abomonation, Serialize, Deserialize, Clone, Debug, Default)]
+#[derive(Abomonation, Serialize, Deserialize, Clone, Debug, Default, DeepSizeOf)]
 pub struct BagOfWords {
     pub universe: u32,
     words: Vec<u32>,

danny-base/src/types.rs

@@ -44,6 +44,7 @@ packed_simd = { version = "0.3.4", package = "packed_simd_2" }
 progress-logger = "0.1.1"
 sha2 = "0.8"
 pprof = {version = "0.4.1", features = ["flamegraph", "protobuf"]}
+deepsize = "0.2"
 
 [build-dependencies]
 vergen = "3.1"

danny/Cargo.toml

@@ -1,4 +1,5 @@
 use crate::operators::Route;
+use deepsize::DeepSizeOf;
 
 /// Utilities to compute the (self) cartesian product of a stream.
 ///
@@ -47,7 +48,7 @@ impl SelfCartesian {
     }
 }
 
-#[derive(Debug, Clone, Copy, Eq, PartialEq, Ord, PartialOrd, Hash, Abomonation)]
+#[derive(Debug, Clone, Copy, Eq, PartialEq, Ord, PartialOrd, Hash, Abomonation, DeepSizeOf)]
 pub enum Marker {
     Left,
     Right,
@@ -75,7 +76,7 @@ impl Marker {
     }
 }
 
-#[derive(Debug, Clone, Copy, Eq, PartialEq, Ord, PartialOrd, Hash, Abomonation)]
+#[derive(Debug, Clone, Copy, Eq, PartialEq, Ord, PartialOrd, Hash, Abomonation, DeepSizeOf)]
 pub struct CartesianKey(pub u8, pub u8);
 
 impl CartesianKey {

danny/src/cartesian.rs

@@ -399,7 +399,6 @@ impl Hosts {
 pub fn parse_hosts(arg: &str) -> Result<Hosts, String> {
     use std::fs::File;
     use std::io::{BufRead, BufReader};
-    use std::path::PathBuf;
 
     let path = PathBuf::from(arg);
     if path.is_file() {

danny/src/config.rs

@@ -1,6 +1,8 @@
+use crate::sysmonitor::DATASTRUCTURES_BYTES;
 use crate::cartesian::*;
 use crate::logging::*;
 use crate::operators::*;
+use deepsize::DeepSizeOf;
 use channels::pact::Pipeline;
 use std::cell::RefCell;
 use std::collections::HashMap;
@@ -68,8 +70,8 @@ impl<G, K, V> Join<G, K, V> for Stream<G, (K, V)>
 where
     G: Scope,
     G::Timestamp: ToStepId,
-    K: KeyData + Ord + std::fmt::Debug,
-    V: ExchangeData + KeyPayload,
+    K: KeyData + Ord + std::fmt::Debug + DeepSizeOf,
+    V: ExchangeData + KeyPayload + DeepSizeOf,
 {
     fn self_join_map<F, I, O>(&self, balance: Balance, mut f: F) -> Stream<G, O>
     where
@@ -108,6 +110,8 @@ where
 
                 input.for_each(|t, data| {
                     let data = data.replace(Vec::new());
+                    DATASTRUCTURES_BYTES
+                        .fetch_add(data.deep_size_of(), std::sync::atomic::Ordering::SeqCst);
                     for (k, v) in data {
                         stash
                             .borrow_mut()