Trying zipfian distro.

datastores/gossip_kv/Cargo.toml

@@ -19,6 +19,7 @@ lazy_static = "1.5.0"
 notify = { version = "6.1.1", default-features = false, features = ["macos_kqueue"] }
 prometheus = "0.13.4"
 rand = "0.8.5"
+rand_distr = "0.4.3"
 serde = "1.0.203"
 serde_json = "1.0.117"
 shlex = "1.3.0"

datastores/gossip_kv/kv/server.rs

@@ -1,7 +1,7 @@
 use std::collections::{HashMap, HashSet};
 use std::fmt::Debug;
 use std::hash::Hash;
-
+use std::iter::Zip;
 use hydroflow::futures::{Sink, Stream};
 use hydroflow::hydroflow_syntax;
 use hydroflow::itertools::Itertools;
@@ -15,6 +15,7 @@ use lazy_static::lazy_static;
 use prometheus::{register_int_counter, IntCounter};
 use rand::seq::IteratorRandom;
 use rand::thread_rng;
+use rand_distr::{Distribution, Zipf};
 use serde::de::DeserializeOwned;
 use serde::{Deserialize, Serialize};
 use tracing::{info, trace};
@@ -102,6 +103,9 @@ where
     let member_id_5 = my_member_id.clone();
     let member_id_6 = my_member_id.clone();
 
+    let zipf = Zipf::new(1_000_000, 4.0).unwrap();
+    let mut rng = thread_rng();
+
     hydroflow_syntax! {
 
         on_start = initialize() -> tee();
@@ -148,8 +152,8 @@ where
             })
             -> writes;
 
-        simulated_puts = repeat_fn(2000, || {
-            upsert_row(Clock::new(100), rand::random(), "value".to_string())
+        simulated_puts = repeat_fn(20000, move || {
+            upsert_row(Clock::new(100), zipf.sample(&mut rng) as u64, "value".to_string())
         })
             -> inspect (|_| {
                 SETS_COUNTER.inc();