Merge pull request #98 from constellation-rs/bench

Performance improvements & benchmarks
constellation-rs · Jul 30, 2020 · e118942 · e118942
2 parents 59937f0 + 6672dd7
commit e118942
Show file tree

Hide file tree

Showing 63 changed files with 1,335 additions and 395 deletions.
diff --git a/Cargo.toml b/Cargo.toml
@@ -2,7 +2,7 @@
 
 [package]
 name = "amadeus"
-version = "0.3.6"
+version = "0.3.7"
 license = "Apache-2.0"
 authors = ["Alec Mocatta <[email protected]>"]
 categories = ["concurrency", "science", "database", "parser-implementations", "text-processing"]
@@ -23,30 +23,31 @@ azure-devops = { project = "alecmocatta/amadeus", pipeline = "tests", build = "2
 maintenance = { status = "actively-developed" }
 
 [features]
-constellation = ["constellation-rs", "serde_traitobject"]
+constellation = ["bincode", "constellation-rs", "serde_traitobject"]
 aws = ["amadeus-aws"]
 commoncrawl = ["amadeus-commoncrawl"]
 parquet = ["amadeus-parquet", "amadeus-derive/parquet"]
 postgres = ["amadeus-postgres", "amadeus-derive/postgres"]
 csv = ["amadeus-serde", "amadeus-derive/serde"]
 json = ["amadeus-serde", "amadeus-derive/serde"]
+bench = ["serde-csv", "once_cell", "arrow-parquet", "rayon"]
 
 [package.metadata.docs.rs]
 features = ["constellation", "aws", "commoncrawl", "parquet", "postgres", "csv", "json"]
 
 [dependencies]
-amadeus-core = { version = "=0.3.6", path = "amadeus-core" }
-amadeus-derive = { version = "=0.3.6", path = "amadeus-derive" }
-amadeus-types = { version = "=0.3.6", path = "amadeus-types" }
-amadeus-aws = { version = "=0.3.6", path = "amadeus-aws", optional = true }
-amadeus-commoncrawl = { version = "=0.3.6", path = "amadeus-commoncrawl", optional = true }
-amadeus-parquet = { version = "=0.3.6", path = "amadeus-parquet", optional = true }
-amadeus-postgres = { version = "=0.3.6", path = "amadeus-postgres", optional = true }
-amadeus-serde = { version = "=0.3.6", path = "amadeus-serde", optional = true }
+amadeus-core = { version = "=0.3.7", path = "amadeus-core" }
+amadeus-derive = { version = "=0.3.7", path = "amadeus-derive" }
+amadeus-types = { version = "=0.3.7", path = "amadeus-types" }
+amadeus-aws = { version = "=0.3.7", path = "amadeus-aws", optional = true }
+amadeus-commoncrawl = { version = "=0.3.7", path = "amadeus-commoncrawl", optional = true }
+amadeus-parquet = { version = "=0.3.7", path = "amadeus-parquet", optional = true }
+amadeus-postgres = { version = "=0.3.7", path = "amadeus-postgres", optional = true }
+amadeus-serde = { version = "=0.3.7", path = "amadeus-serde", optional = true }
 async-channel = "1.1"
+bincode = { version = "1.3", optional = true }
 constellation-rs = { version = "0.2.0-alpha.2", default-features = false, optional = true }
 derive-new = "0.5"
-doc-comment = "0.3"
 futures = "0.3"
 num_cpus = "1.13"
 pin-project = "0.4"
@@ -55,12 +56,19 @@ serde_closure = "0.3"
 serde_traitobject = { version = "0.2", optional = true }
 tokio = { version = "0.2", features = ["rt-threaded", "rt-util", "blocking"] }
 
+# Move to dev-dependencies once fixed: https://github.com/rust-lang/cargo/issues/1596
+arrow-parquet = { package = "parquet", version = "1.0", default-features = false, features = ["brotli", "flate2", "lz4", "snap"], optional = true }
+once_cell = { version = "1.4", optional = true }
+rayon = { version = "1.3", optional = true }
+serde-csv = { package = "csv", version = "1.0", optional = true }
+
 [target.'cfg(target_arch = "wasm32")'.dependencies]
 wasm-bindgen = "0.2"
 wasm-bindgen-futures = "0.4"
 web-sys = { version = "0.3", features = ["Blob", "Performance", "Response", "Window"] }
 
 [dev-dependencies]
+doc-comment = "0.3"
 either = { version = "1.5", features = ["serde"] }
 rand = "0.7"
 serde_json = "1.0"
@@ -164,3 +172,15 @@ name = "postgres_dist"
 harness = false
 required-features = ["postgres"]
 test = false # TODO set up postgres on CI
+
+[[bench]]
+name = "csv"
+required-features = ["bench", "csv"]
+
+[[bench]]
+name = "in_memory"
+required-features = ["bench"]
+
+[[bench]]
+name = "parquet"
+required-features = ["bench", "parquet"]
diff --git a/amadeus-aws/Cargo.toml b/amadeus-aws/Cargo.toml
@@ -1,6 +1,6 @@
 [package]
 name = "amadeus-aws"
-version = "0.3.6"
+version = "0.3.7"
 license = "Apache-2.0"
 authors = ["Alec Mocatta <[email protected]>"]
 categories = ["concurrency", "science", "database", "parser-implementations", "text-processing"]
@@ -19,8 +19,8 @@ azure-devops = { project = "alecmocatta/amadeus", pipeline = "tests", build = "2
 maintenance = { status = "actively-developed" }
 
 [dependencies]
-amadeus-core = { version = "=0.3.6", path = "../amadeus-core" }
-amadeus-types = { version = "=0.3.6", path = "../amadeus-types" }
+amadeus-core = { version = "=0.3.7", path = "../amadeus-core" }
+amadeus-types = { version = "=0.3.7", path = "../amadeus-types" }
 async-compression = { version = "0.3.3", features = ["gzip", "futures-bufread"] }
 async-trait = "0.1"
 chrono = { version = "0.4", default-features = false }

diff --git a/amadeus-aws/src/lib.rs b/amadeus-aws/src/lib.rs
@@ -6,7 +6,7 @@
 //!
 //! This is a support crate of [Amadeus](https://github.com/constellation-rs/amadeus) and is not intended to be used directly. These types are re-exposed in [`amadeus::source`](https://docs.rs/amadeus/0.3/amadeus/source/index.html).
 
-#![doc(html_root_url = "https://docs.rs/amadeus-aws/0.3.6")]
+#![doc(html_root_url = "https://docs.rs/amadeus-aws/0.3.7")]
 #![cfg_attr(nightly, feature(type_alias_impl_trait))]
 #![warn(
 	// missing_copy_implementations,

diff --git a/amadeus-commoncrawl/Cargo.toml b/amadeus-commoncrawl/Cargo.toml
@@ -1,6 +1,6 @@
 [package]
 name = "amadeus-commoncrawl"
-version = "0.3.6"
+version = "0.3.7"
 license = "MIT OR Apache-2.0"
 authors = ["Stephen Becker IV <[email protected]>", "Alec Mocatta <[email protected]>"]
 categories = ["concurrency", "science", "database", "parser-implementations", "text-processing"]
@@ -19,8 +19,8 @@ azure-devops = { project = "alecmocatta/amadeus", pipeline = "tests", build = "2
 maintenance = { status = "actively-developed" }
 
 [dependencies]
-amadeus-core = { version = "=0.3.6", path = "../amadeus-core" }
-amadeus-types = { version = "=0.3.6", path = "../amadeus-types" }
+amadeus-core = { version = "=0.3.7", path = "../amadeus-core" }
+amadeus-types = { version = "=0.3.7", path = "../amadeus-types" }
 async-compression = { version = "0.3.3", features = ["gzip", "futures-bufread"] }
 futures = "0.3"
 nom = "4.2.3"

diff --git a/amadeus-commoncrawl/src/lib.rs b/amadeus-commoncrawl/src/lib.rs
@@ -6,7 +6,7 @@
 //!
 //! This is a support crate of [Amadeus](https://github.com/constellation-rs/amadeus) and is not intended to be used directly. These types are re-exposed in [`amadeus::source`](https://docs.rs/amadeus/0.3/amadeus/source/index.html).
 
-#![doc(html_root_url = "https://docs.rs/amadeus-commoncrawl/0.3.6")]
+#![doc(html_root_url = "https://docs.rs/amadeus-commoncrawl/0.3.7")]
 #![cfg_attr(nightly, feature(type_alias_impl_trait))]
 #![warn(
 	// missing_copy_implementations,

diff --git a/amadeus-core/Cargo.toml b/amadeus-core/Cargo.toml
@@ -1,6 +1,6 @@
 [package]
 name = "amadeus-core"
-version = "0.3.6"
+version = "0.3.7"
 license = "Apache-2.0"
 authors = ["Alec Mocatta <[email protected]>"]
 categories = ["concurrency", "science", "database", "parser-implementations", "text-processing"]
@@ -24,6 +24,7 @@ derive-new = "0.5"
 educe = "0.4"
 either = { version = "1.5", features = ["serde"] }
 futures = "0.3"
+indexmap = { version = "1.5", features = ["serde-1"] }
 itertools = "0.9"
 owned_chars = "0.3"
 pin-project = "0.4"

diff --git a/amadeus-core/src/into_par_stream.rs b/amadeus-core/src/into_par_stream.rs
@@ -13,12 +13,14 @@ impl_par_dist_rename! {
 		fn into_par_stream(self) -> Self::ParStream
 		where
 			Self: Sized;
-		fn dist_stream_mut(&mut self) -> <&mut Self as IntoParallelStream>::ParStream
+		#[inline(always)]
+		fn par_stream_mut(&mut self) -> <&mut Self as IntoParallelStream>::ParStream
 		where
 			for<'a> &'a mut Self: IntoParallelStream,
 		{
 			<&mut Self as IntoParallelStream>::into_par_stream(self)
 		}
+		#[inline(always)]
 		fn par_stream(&self) -> <&Self as IntoParallelStream>::ParStream
 		where
 			for<'a> &'a Self: IntoParallelStream,
@@ -31,6 +33,7 @@ impl_par_dist_rename! {
 		type ParStream = Self;
 		type Item = <Self as ParallelStream>::Item;
 
+		#[inline(always)]
 		fn into_par_stream(self) -> Self::ParStream
 		where
 			Self: Sized,

diff --git a/amadeus-core/src/into_par_stream/collections.rs b/amadeus-core/src/into_par_stream/collections.rs
@@ -14,6 +14,7 @@ impl<'a, 'b, I: Iterator<Item = (&'a A, &'b B)>, A: Clone + 'a, B: Clone + 'b> I
 {
 	type Item = (A, B);
 
+	#[inline(always)]
 	fn next(&mut self) -> Option<Self::Item> {
 		self.0.next().map(|(a, b)| (a.clone(), b.clone()))
 	}
@@ -27,6 +28,7 @@ impl_par_dist_rename! {
 		type ParStream = IterParStream<vec::IntoIter<T>>;
 		type Item = T;
 
+		#[inline(always)]
 		fn into_par_stream(self) -> Self::ParStream
 		where
 			Self: Sized,
@@ -41,6 +43,7 @@ impl_par_dist_rename! {
 		type ParStream = IterParStream<iter::Cloned<slice::Iter<'a, T>>>;
 		type Item = T;
 
+		#[inline(always)]
 		fn into_par_stream(self) -> Self::ParStream
 		where
 			Self: Sized,
@@ -56,6 +59,7 @@ impl_par_dist_rename! {
 		type ParStream = IterParStream<vec_deque::IntoIter<T>>;
 		type Item = T;
 
+		#[inline(always)]
 		fn into_par_stream(self) -> Self::ParStream
 		where
 			Self: Sized,
@@ -70,6 +74,7 @@ impl_par_dist_rename! {
 		type ParStream = IterParStream<iter::Cloned<vec_deque::Iter<'a, T>>>;
 		type Item = T;
 
+		#[inline(always)]
 		fn into_par_stream(self) -> Self::ParStream
 		where
 			Self: Sized,
@@ -85,6 +90,7 @@ impl_par_dist_rename! {
 		type ParStream = IterParStream<binary_heap::IntoIter<T>>;
 		type Item = T;
 
+		#[inline(always)]
 		fn into_par_stream(self) -> Self::ParStream
 		where
 			Self: Sized,
@@ -99,6 +105,7 @@ impl_par_dist_rename! {
 		type ParStream = IterParStream<iter::Cloned<binary_heap::Iter<'a, T>>>;
 		type Item = T;
 
+		#[inline(always)]
 		fn into_par_stream(self) -> Self::ParStream
 		where
 			Self: Sized,
@@ -114,6 +121,7 @@ impl_par_dist_rename! {
 		type ParStream = IterParStream<linked_list::IntoIter<T>>;
 		type Item = T;
 
+		#[inline(always)]
 		fn into_par_stream(self) -> Self::ParStream
 		where
 			Self: Sized,
@@ -128,6 +136,7 @@ impl_par_dist_rename! {
 		type ParStream = IterParStream<iter::Cloned<linked_list::Iter<'a, T>>>;
 		type Item = T;
 
+		#[inline(always)]
 		fn into_par_stream(self) -> Self::ParStream
 		where
 			Self: Sized,
@@ -144,6 +153,7 @@ impl_par_dist_rename! {
 		type ParStream = IterParStream<hash_set::IntoIter<T>>;
 		type Item = T;
 
+		#[inline(always)]
 		fn into_par_stream(self) -> Self::ParStream
 		where
 			Self: Sized,
@@ -159,6 +169,7 @@ impl_par_dist_rename! {
 		type ParStream = IterParStream<iter::Cloned<hash_set::Iter<'a, T>>>;
 		type Item = T;
 
+		#[inline(always)]
 		fn into_par_stream(self) -> Self::ParStream
 		where
 			Self: Sized,
@@ -176,6 +187,7 @@ impl_par_dist_rename! {
 		type ParStream = IterParStream<hash_map::IntoIter<K, V>>;
 		type Item = (K, V);
 
+		#[inline(always)]
 		fn into_par_stream(self) -> Self::ParStream
 		where
 			Self: Sized,
@@ -192,6 +204,7 @@ impl_par_dist_rename! {
 		type ParStream = IterParStream<TupleCloned<hash_map::Iter<'a, K, V>>>;
 		type Item = (K, V);
 
+		#[inline(always)]
 		fn into_par_stream(self) -> Self::ParStream
 		where
 			Self: Sized,
@@ -207,6 +220,7 @@ impl_par_dist_rename! {
 		type ParStream = IterParStream<btree_set::IntoIter<T>>;
 		type Item = T;
 
+		#[inline(always)]
 		fn into_par_stream(self) -> Self::ParStream
 		where
 			Self: Sized,
@@ -221,6 +235,7 @@ impl_par_dist_rename! {
 		type ParStream = IterParStream<iter::Cloned<btree_set::Iter<'a, T>>>;
 		type Item = T;
 
+		#[inline(always)]
 		fn into_par_stream(self) -> Self::ParStream
 		where
 			Self: Sized,
@@ -237,6 +252,7 @@ impl_par_dist_rename! {
 		type ParStream = IterParStream<btree_map::IntoIter<K, V>>;
 		type Item = (K, V);
 
+		#[inline(always)]
 		fn into_par_stream(self) -> Self::ParStream
 		where
 			Self: Sized,
@@ -252,6 +268,7 @@ impl_par_dist_rename! {
 		type ParStream = IterParStream<TupleCloned<btree_map::Iter<'a, K, V>>>;
 		type Item = (K, V);
 
+		#[inline(always)]
 		fn into_par_stream(self) -> Self::ParStream
 		where
 			Self: Sized,
@@ -264,6 +281,7 @@ impl_par_dist_rename! {
 		type ParStream = IterParStream<IntoChars>;
 		type Item = char;
 
+		#[inline(always)]
 		fn into_par_stream(self) -> Self::ParStream
 		where
 			Self: Sized,
@@ -275,6 +293,7 @@ impl_par_dist_rename! {
 		type ParStream = IterParStream<str::Chars<'a>>;
 		type Item = char;
 
+		#[inline(always)]
 		fn into_par_stream(self) -> Self::ParStream
 		where
 			Self: Sized,
@@ -290,6 +309,7 @@ impl_par_dist_rename! {
 		type ParStream = IterParStream<option::IntoIter<T>>;
 		type Item = T;
 
+		#[inline(always)]
 		fn into_par_stream(self) -> Self::ParStream
 		where
 			Self: Sized,
@@ -304,6 +324,7 @@ impl_par_dist_rename! {
 		type ParStream = IterParStream<iter::Cloned<option::Iter<'a, T>>>;
 		type Item = T;
 
+		#[inline(always)]
 		fn into_par_stream(self) -> Self::ParStream
 		where
 			Self: Sized,
@@ -319,6 +340,7 @@ impl_par_dist_rename! {
 		type ParStream = IterParStream<result::IntoIter<T>>;
 		type Item = T;
 
+		#[inline(always)]
 		fn into_par_stream(self) -> Self::ParStream
 		where
 			Self: Sized,
@@ -333,6 +355,7 @@ impl_par_dist_rename! {
 		type ParStream = IterParStream<iter::Cloned<result::Iter<'a, T>>>;
 		type Item = T;
 
+		#[inline(always)]
 		fn into_par_stream(self) -> Self::ParStream
 		where
 			Self: Sized,