Update dependencies, deprecations and clippy

Cargo.toml

@@ -23,7 +23,7 @@ profiler = []
 
 [dependencies]
 # for logging to the console
-log = { version = "0.4.20", features = ["release_max_level_info"] }
+log = { version = "0.4.21", features = ["release_max_level_info"] }
 
 # used by the network for storing type-generic structures
 typemap_rev = "0.3.0"
@@ -36,31 +36,31 @@ derivative = "2.2.0"
 
 # serialization library used for reading the config file (yaml) and serializing
 # the messages on the network
-serde = { version = "1.0.193", features = ["derive"] }
-serde_yaml = "0.9.28"
-serde_json = "1.0.108"
+serde = { version = "1.0.197", features = ["derive"] }
+serde_yaml = "0.9.32"
+serde_json = "1.0.114"
 bincode = "1.3.3"
 
 # handy Result type
-anyhow = "1.0.76"
-thiserror = "1.0.51"
+anyhow = "1.0.81"
+thiserror = "1.0.58"
 
 # handy iterators functions
-itertools = "0.12.0"
+itertools = "0.12.1"
 
 # for storing non-mutable static variables with non-trivial initialization
 once_cell = "1.19.0"
 
 # used for spawning the remote workers
 ssh2 = { version = "0.9.4", optional = true }
-whoami = { version = "1.4.1", optional = true }
+whoami = { version = "1.5.1", optional = true }
 shell-escape = { version = "0.1.5", optional = true }
-clap = { version = "4.4.11", features = ["derive"], optional = true }
+clap = { version = "4.5.2", features = ["derive"], optional = true }
 sha2 = { version = "0.10.8", optional = true }
-base64 = { version = "0.21.5", optional = true }
+base64 = { version = "0.22.0", optional = true }
 
 # channel implementation
-crossbeam-channel = { version = "0.5.9", optional = true }
+crossbeam-channel = { version = "0.5.12", optional = true }
 flume = { version = "0.11.0", optional = true }
 
 # used for csv file source
@@ -70,26 +70,26 @@ csv = "1.3.0"
 lazy-init = "0.5.1"
 
 # Faster monotonic clock using libc's CLOCK_MONOTONIC_COARSE
-coarsetime = "0.1.33"
+coarsetime = "0.1.34"
 
-tokio = { version = "1.35.1", features = ["rt"], default-features = false, optional = true }
-futures = { version = "0.3.29", optional = true }
+tokio = { version = "1.36.0", features = ["rt"], default-features = false, optional = true }
+futures = { version = "0.3.30", optional = true }
 
 parking_lot = "0.12.1"
 
 wyhash = "0.5.0"
 fxhash = "0.2.1"
 glidesort = "0.1.2"
-indexmap = "2.1.0"
+indexmap = "2.2.5"
 tracing = { version = "0.1.40", features = ["log"] }
-quick_cache = "0.4.0"
+quick_cache = "0.4.1"
 dashmap = "5.5.3"
 
 [dev-dependencies]
 # for the tests
-env_logger = "0.10.1"
+env_logger = "0.11.3"
 rand = { version = "0.8.5", features = ["small_rng"] }
-tempfile = "3.8.1"
+tempfile = "3.10.1"
 criterion = { version = "0.5.1", features = ["html_reports"] }
 fake = "2.9.2"
 mimalloc = { version = "0.1.39", default-features = false }
@@ -98,10 +98,10 @@ tracing-subscriber = "0.3.18"
 micrometer = { version = "0.2.7", features = ["enable"]}
 
 # for the examples
-regex = "1.10.2"
+regex = "1.10.3"
 
 # used in the benchmarks
-crossbeam-channel = "0.5.9"
+crossbeam-channel = "0.5.12"
 flume = "0.11.0"
 
 kstring = { version = "2.0.0", features = ["serde"] }

README.md

@@ -24,8 +24,8 @@ use noir_compute::prelude::*;
 fn main() {
     // Convenience method to parse deployment config from CLI arguments
     let (config, args) = EnvironmentConfig::from_args();
+    config.spawn_remote_workers();
     let mut env = StreamEnvironment::new(config);
-    env.spawn_remote_workers();
 
     let result = env
         // Open and read file line by line in parallel

examples/car_accidents.rs

@@ -312,8 +312,8 @@ fn main() {
         _ => panic!("share_source must be either `true` or `false`, not `{share_source}`"),
     };
 
+    config.spawn_remote_workers();
     let mut env = StreamEnvironment::new(config);
-    env.spawn_remote_workers();
 
     let (query1, query2, query3) = if share_source {
         let source = CsvSource::<Accident>::new(path).has_headers(true);

examples/collatz.rs

@@ -12,10 +12,9 @@ fn main() {
     let limit: u64 = args[0].parse().unwrap();
     let num_iter = 1000;
 
+    config.spawn_remote_workers();
     let mut env = StreamEnvironment::new(config);
 
-    env.spawn_remote_workers();
-
     let output = env
         .stream_par_iter(1..limit)
         .batch_mode(BatchMode::fixed(1024))

examples/connected_components.rs

@@ -37,10 +37,9 @@ fn main() {
     let path_vertices = &args[2];
     let path_edges = &args[3];
 
+    config.spawn_remote_workers();
     let mut env = StreamEnvironment::new(config);
 
-    env.spawn_remote_workers();
-
     let vertices_source = CsvSource::<u64>::new(path_vertices).has_headers(false);
     let edges_source = CsvSource::<(u64, u64)>::new(path_edges)
         .delimiter(b',')

examples/kmeans.rs

@@ -115,10 +115,9 @@ fn main() {
     let num_iters: usize = args[1].parse().expect("Invalid number of iterations");
     let path = &args[2];
 
+    config.spawn_remote_workers();
     let mut env = StreamEnvironment::new(config);
 
-    env.spawn_remote_workers();
-
     let centroids = read_centroids(path, num_centroids);
     assert_eq!(centroids.len(), num_centroids);
     let initial_state = State::new(centroids);

examples/nexmark-latency.rs

@@ -202,8 +202,8 @@ fn main() {
     let args = Args::parse_from(std::iter::once("nexmark-latency".into()).chain(args));
     let q = &args.q[..];
 
+    config.spawn_remote_workers();
     let mut env = StreamEnvironment::new(config);
-    env.spawn_remote_workers();
     micrometer::start();
 
     match q {

examples/nexmark.rs

@@ -470,8 +470,8 @@ fn main() {
     }
     let n: usize = args[0].parse().unwrap();
     let q = &args[1][..];
+    config.spawn_remote_workers();
     let mut env = StreamEnvironment::new(config);
-    env.spawn_remote_workers();
 
     match q {
         "0" => query0(events(&mut env, n)),

examples/pagerank.rs

@@ -18,10 +18,9 @@ fn main() {
     let path_pages = &args[2];
     let path_links = &args[3];
 
+    config.spawn_remote_workers();
     let mut env = StreamEnvironment::new(config);
 
-    env.spawn_remote_workers();
-
     let pages_source = CsvSource::<u64>::new(path_pages).has_headers(false);
     let links_source = CsvSource::<(u64, u64)>::new(path_links).has_headers(false);
 

examples/pagerank_stateful.rs

@@ -24,10 +24,9 @@ fn main() {
     let path_pages = &args[2];
     let path_links = &args[3];
 
+    config.spawn_remote_workers();
     let mut env = StreamEnvironment::new(config);
 
-    env.spawn_remote_workers();
-
     let q = Instant::now();
     let links = BufReader::new(File::open(path_links).unwrap());
     let adjacency_list: HashMap<u64, Vec<u64>, BuildHasherDefault<fxhash::FxHasher>> =

examples/pagerank_timely.rs

@@ -18,8 +18,8 @@ fn main() {
     let edges: u64 = args[1].parse().unwrap();
     let max_iter: usize = 10000;
 
+    config.spawn_remote_workers();
     let mut env = StreamEnvironment::new(config);
-    env.spawn_remote_workers();
 
     let source = env
         .stream_par_iter(move |index, peers| {

examples/rolling_top_words.rs

@@ -167,8 +167,8 @@ fn main() {
     let k = 4;
 
     let (config, _args) = EnvironmentConfig::from_args();
+    config.spawn_remote_workers();
     let mut env = StreamEnvironment::new(config);
-    env.spawn_remote_workers();
 
     let source = ParallelIteratorSource::new(TopicSource::new);
     env.stream(source)

examples/rolling_top_words_e2e.rs

@@ -113,8 +113,8 @@ fn main() {
     let (config, args) = EnvironmentConfig::from_args();
     let limit: u64 = args[0].parse().expect("Invalid number of events");
 
+    config.spawn_remote_workers();
     let mut env = StreamEnvironment::new(config);
-    env.spawn_remote_workers();
 
     let source =
         ParallelIteratorSource::new(move |id, instances| TopicSource::new(id, instances, limit));

examples/simple.rs

@@ -2,8 +2,8 @@ use noir_compute::prelude::*;
 
 fn main() {
     let (config, _args) = EnvironmentConfig::from_args();
+    config.spawn_remote_workers();
     let mut env = StreamEnvironment::new(config);
-    env.spawn_remote_workers();
 
     let source = IteratorSource::new(0..100);
     let result = env

examples/transitive_closure.rs

@@ -13,10 +13,9 @@ fn main() {
     let num_iterations: usize = args[0].parse().expect("Invalid number of iterations");
     let path_edges = &args[1];
 
+    config.spawn_remote_workers();
     let mut env = StreamEnvironment::new(config);
 
-    env.spawn_remote_workers();
-
     let edges_source = CsvSource::<(u64, u64)>::new(path_edges)
         .delimiter(b',')
         .has_headers(false);

examples/triangles_fold.rs

@@ -12,10 +12,9 @@ fn main() {
     }
     let path = &args[0];
 
+    config.spawn_remote_workers();
     let mut env = StreamEnvironment::new(config);
 
-    env.spawn_remote_workers();
-
     let source = CsvSource::<(u32, u32)>::new(path).has_headers(false);
     let mut edges = env
         .stream(source)

examples/triangles_rich_map.rs

@@ -14,10 +14,9 @@ fn main() {
     }
     let path = &args[0];
 
+    config.spawn_remote_workers();
     let mut env = StreamEnvironment::new(config);
 
-    env.spawn_remote_workers();
-
     let source = CsvSource::<(u32, u32)>::new(path).has_headers(false);
     let mut edges = env
         .stream(source)

examples/wordcount.rs

@@ -21,10 +21,9 @@ fn main() {
     }
     let path = &args[0];
 
+    config.spawn_remote_workers();
     let mut env = StreamEnvironment::new(config);
 
-    env.spawn_remote_workers();
-
     let source = FileSource::new(path);
     let tokenizer = Tokenizer::new();
     let result = env
@@ -53,10 +52,9 @@ async fn main() {
     }
     let path = &args[0];
 
+    config.spawn_remote_workers();
     let mut env = StreamEnvironment::new(config);
 
-    env.spawn_remote_workers();
-
     let source = FileSource::new(path);
     let tokenizer = Tokenizer::new();
     let result = env

examples/wordcount_assoc.rs

@@ -17,10 +17,9 @@ fn main() {
     }
     let path = &args[0];
 
+    config.spawn_remote_workers();
     let mut env = StreamEnvironment::new(config);
 
-    env.spawn_remote_workers();
-
     let source = FileSource::new(path);
     let tokenizer = Tokenizer::new();
     let result = env
@@ -57,10 +56,9 @@ async fn main() {
     }
     let path = &args[0];
 
+    config.spawn_remote_workers();
     let mut env = StreamEnvironment::new(config);
 
-    env.spawn_remote_workers();
-
     let source = FileSource::new(path);
     let tokenizer = Tokenizer::new();
     let result = env

examples/wordcount_opt.rs

@@ -16,10 +16,9 @@ fn main() {
     }
     let path = &args[0];
 
+    config.spawn_remote_workers();
     let mut env = StreamEnvironment::new(config);
 
-    env.spawn_remote_workers();
-
     let result = env
         .stream_file(path)
         .batch_mode(BatchMode::fixed(1024))

examples/wordcount_windowed.rs

@@ -14,10 +14,9 @@ fn main() {
     }
     let path = &args[0];
 
+    config.spawn_remote_workers();
     let mut env = StreamEnvironment::new(config);
 
-    env.spawn_remote_workers();
-
     let source = FileSource::new(path);
     let tokenizer = Tokenizer::new();
     env.stream(source)

src/lib.rs

@@ -25,8 +25,8 @@ use noir_compute::prelude::*;
 fn main() {
     // Convenience method to parse deployment config from CLI arguments
     let (config, args) = EnvironmentConfig::from_args();
+        config.spawn_remote_workers();
     let mut env = StreamEnvironment::new(config);
-    env.spawn_remote_workers();
 
     let result = env
         // Open and read file line by line in parallel
@@ -124,7 +124,7 @@ use serde::{Deserialize, Serialize};
 pub use block::structure;
 pub use block::BatchMode;
 pub use block::Replication;
-pub use block::{GroupHasherBuilder, group_by_hash};
+pub use block::{group_by_hash, GroupHasherBuilder};
 pub use config::EnvironmentConfig;
 pub use environment::StreamEnvironment;
 pub use operator::iteration::IterationStateHandle;