update jsonpath library to support async compute

Cargo.toml

@@ -18,12 +18,16 @@ license = "MIT"
 travis-ci = { repository = "freestrings/jsonpath", branch = "master" }
 
 [dependencies]
+futures = "0.3"
 log = "0.4"
 serde = { version = "1.0", features = ["derive"] }
 serde_json = { version = "1.0", features = ["preserve_order"] }
 
 [dev-dependencies]
+criterion = { version = "0.5.1", features = ["html_reports", "async_tokio"] }
 env_logger = "0.8"
+tokio = { version = "1.33.0", features = ["macros"] }
+tokio-test = "0.4.3"
 
 [lib]
 name = "jsonpath_lib"
@@ -33,3 +37,11 @@ crate-type = ["cdylib", "rlib"]
 #[profile.release]
 #debug = true
 #lto = false
+
+[[bench]]
+name = "async"
+harness = false
+
+[[bench]]
+name = "sync_mut"
+harness = false

benches/async.rs

@@ -0,0 +1,167 @@
+extern crate jsonpath_lib as jsonpath;
+#[macro_use]
+extern crate serde_json;
+
+use std::{
+    pin::Pin,
+    sync::{Arc, Mutex},
+    task::{Context, Poll},
+};
+
+use common::read_json;
+use criterion::{criterion_group, criterion_main, BenchmarkId};
+use futures::Future;
+use jsonpath::{MultiJsonSelectorMutWithMetadata, PathParserWithMetadata};
+use serde_json::Value;
+
+mod common;
+
+#[derive(Clone)]
+struct ValueFuture<T> {
+    inner: Arc<Mutex<Option<T>>>,
+}
+
+impl<T> ValueFuture<T> {
+    fn new() -> Self {
+        ValueFuture {
+            inner: Arc::new(Mutex::new(None)),
+        }
+    }
+
+    fn set_value(&self, value: T) {
+        let mut inner = self.inner.lock().unwrap();
+        *inner = Some(value);
+    }
+}
+
+impl<T: Clone> Future for ValueFuture<T> {
+    type Output = T;
+
+    fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
+        let inner = self.inner.lock().unwrap();
+        if let Some(value) = inner.as_ref() {
+            Poll::Ready(value.clone())
+        } else {
+            // This future isn't ready yet, so we'll notify the context when it is.
+            cx.waker().wake_by_ref();
+            Poll::Pending
+        }
+    }
+}
+
+struct MutationRequest {
+    bags: Mutex<Vec<Field>>,
+}
+
+impl MutationRequest {
+    fn new() -> Self {
+        Self {
+            bags: Mutex::new(Vec::new()),
+        }
+    }
+
+    fn new_field(&self, input: Value) -> Field {
+        let bag = Field::new(input);
+        self.bags.lock().unwrap().push(bag.clone());
+        bag
+    }
+
+    async fn send_request(&self) {
+        let mut bags = self.bags.lock().unwrap();
+        for bag in bags.iter_mut() {
+            bag.value.set_value(bag.input.take().unwrap());
+        }
+    }
+}
+
+#[derive(Clone)]
+struct Field {
+    input: Option<Value>,
+    value: ValueFuture<Value>,
+}
+
+impl Field {
+    fn new(input: Value) -> Self {
+        Self {
+            input: Some(input),
+            value: ValueFuture::new(),
+        }
+    }
+
+    pub fn value(self) -> ValueFuture<Value> {
+        self.value
+    }
+}
+
+async fn async_run(mut selector_mut: MultiJsonSelectorMutWithMetadata<'_, &str>, json: Value) {
+    let mut_request = Arc::new(MutationRequest::new());
+
+    let result_futures = selector_mut
+        .replace_with_async(json, |v, _| {
+            let bag: Field = mut_request.new_field(v);
+
+            Box::pin(async move {
+                let val = bag.value().await;
+                Some(val)
+            })
+        })
+        .unwrap();
+
+    mut_request.send_request().await;
+
+    let _result = result_futures.await.unwrap();
+}
+
+fn setup_async_benchmark(c: &mut criterion::Criterion) {
+    let t1_json = read_json("./benchmark/example.json");
+    let t1_parser = PathParserWithMetadata::compile("$.store..price", "one").unwrap();
+    let t1_parser_two = PathParserWithMetadata::compile("$.store..author", "two").unwrap();
+    let t1_selector_mut =
+        MultiJsonSelectorMutWithMetadata::new_multi_parser(vec![t1_parser, t1_parser_two]);
+
+    // let big_array = read_json("./benchmark/big_array.json");
+    let t2_json = read_json("./benchmark/big_example.json");
+    let t2_parser = PathParserWithMetadata::compile("$.store.book[*].author", "one").unwrap();
+    let t2_parser_two = PathParserWithMetadata::compile("$.store.author", "two").unwrap();
+    let t2_selector_mut =
+        MultiJsonSelectorMutWithMetadata::new_multi_parser(vec![t2_parser, t2_parser_two]);
+
+    let runtime = tokio::runtime::Builder::new_current_thread()
+        .build()
+        .unwrap();
+
+    c.bench_with_input(
+        BenchmarkId::new("async_selector_mut", "Json"),
+        &(t1_selector_mut, t1_json),
+        |b, (s, v)| {
+            // Insert a call to `to_async` to convert the bencher to async mode.
+            // The timing loops are the same as with the normal bencher.
+            b.to_async(&runtime).iter_batched(
+                || (s.clone(), v.clone()),
+                |(s, v)| async {
+                    async_run(s, v).await;
+                },
+                criterion::BatchSize::SmallInput,
+            );
+        },
+    );
+
+    c.bench_with_input(
+        BenchmarkId::new("async_selector_mut", "BigJson"),
+        &(t2_selector_mut, t2_json),
+        |b, (s, v)| {
+            // Insert a call to `to_async` to convert the bencher to async mode.
+            // The timing loops are the same as with the normal bencher.
+            b.to_async(&runtime).iter_batched(
+                || (s.clone(), v.clone()),
+                |(s, v)| async {
+                    async_run(s, v).await;
+                },
+                criterion::BatchSize::LargeInput,
+            );
+        },
+    );
+}
+
+criterion_group!(benches, setup_async_benchmark);
+criterion_main!(benches);

benches/common.rs

@@ -0,0 +1,52 @@
+extern crate env_logger;
+extern crate jsonpath_lib as jsonpath;
+extern crate serde_json;
+
+use std::io::Read;
+
+use serde_json::Value;
+
+use self::jsonpath::{JsonSelector, PathParser};
+
+#[allow(dead_code)]
+pub fn setup() {
+    let _ = env_logger::try_init();
+}
+
+#[allow(dead_code)]
+pub fn read_json(path: &str) -> Value {
+    let mut f = std::fs::File::open(path).unwrap();
+    let mut contents = String::new();
+    f.read_to_string(&mut contents).unwrap();
+    serde_json::from_str(&contents).unwrap()
+}
+
+#[allow(dead_code)]
+pub fn read_contents(path: &str) -> String {
+    let mut f = std::fs::File::open(path).unwrap();
+    let mut contents = String::new();
+    f.read_to_string(&mut contents).unwrap();
+    contents
+}
+
+#[allow(dead_code)]
+pub fn select_and_then_compare(path: &str, json: Value, target: Value) {
+    let parser = PathParser::compile(path).unwrap();
+    let mut selector = JsonSelector::new(parser);
+    let result = selector.value(&json).select_as::<Value>().unwrap();
+    assert_eq!(
+        result,
+        match target {
+            Value::Array(vec) => vec,
+            _ => panic!("Give me the Array!"),
+        },
+        "{}",
+        path
+    );
+}
+
+#[allow(dead_code)]
+pub fn compare_result(result: Vec<&Value>, target: Value) {
+    let result = serde_json::to_value(result).unwrap();
+    assert_eq!(result, target);
+}

benches/sync_mut.rs

@@ -0,0 +1,113 @@
+extern crate jsonpath_lib as jsonpath;
+extern crate serde_json;
+
+use common::read_json;
+use criterion::{criterion_group, criterion_main, BenchmarkId};
+
+use jsonpath::{JsonSelectorMut, PathParser};
+use serde_json::Value;
+
+mod common;
+
+fn selector_mut(mut selector_mut: JsonSelectorMut, json: Value) -> Value {
+    let mut nums = Vec::new();
+    let result = selector_mut
+        .value(json)
+        .replace_with(&mut |v| {
+            if let Value::Number(n) = v {
+                nums.push(n.as_f64().unwrap());
+            }
+            Ok(Some(Value::String("a".to_string())))
+        })
+        .unwrap()
+        .take()
+        .unwrap();
+
+    result
+}
+
+fn setup_async_benchmark(c: &mut criterion::Criterion) {
+    let t1_json = read_json("./benchmark/example.json");
+    let t1_parser = PathParser::compile("$.store..price").unwrap();
+    let t1_selector_mut = JsonSelectorMut::new(t1_parser.clone());
+    let t1_selector_mut_two = JsonSelectorMut::new(t1_parser);
+
+    let t2_json = read_json("./benchmark/big_example.json");
+    let t2_parser = PathParser::compile("$.store.book[*].author").unwrap();
+    let t2_parser_two = PathParser::compile("$.store.author").unwrap();
+    let t2_selector_mut = JsonSelectorMut::new(t2_parser);
+    let t2_selector_mut_two = JsonSelectorMut::new(t2_parser_two);
+
+    let runtime = tokio::runtime::Builder::new_current_thread()
+        .build()
+        .unwrap();
+
+    c.bench_with_input(
+        BenchmarkId::new("selector_mut", "Json"),
+        &(t1_selector_mut.clone(), t1_json.clone()),
+        |b, (s, v)| {
+            // Insert a call to `to_async` to convert the bencher to async mode.
+            // The timing loops are the same as with the normal bencher.
+            b.to_async(&runtime).iter_batched(
+                || (s.clone(), v.clone()),
+                |(s, v)| async {
+                    selector_mut(s, v);
+                },
+                criterion::BatchSize::SmallInput,
+            );
+        },
+    );
+
+    c.bench_with_input(
+        BenchmarkId::new("selector_mut", "BigJson"),
+        &(t2_selector_mut.clone(), t2_json.clone()),
+        |b, (s, v)| {
+            // Insert a call to `to_async` to convert the bencher to async mode.
+            // The timing loops are the same as with the normal bencher.
+            b.to_async(&runtime).iter_batched(
+                || (s.clone(), v.clone()),
+                |(s, v)| async {
+                    selector_mut(s, v);
+                },
+                criterion::BatchSize::LargeInput,
+            );
+        },
+    );
+
+    c.bench_with_input(
+        BenchmarkId::new("double_selector_mut", "Json"),
+        &(t1_selector_mut, t1_selector_mut_two, t1_json),
+        |b, (s, s2, v)| {
+            // Insert a call to `to_async` to convert the bencher to async mode.
+            // The timing loops are the same as with the normal bencher.
+            b.to_async(&runtime).iter_batched(
+                || (s.clone(), s2.clone(), v.clone()),
+                |(s, s2, v)| async {
+                    let v = selector_mut(s, v);
+                    let _ = selector_mut(s2, v);
+                },
+                criterion::BatchSize::SmallInput,
+            );
+        },
+    );
+
+    c.bench_with_input(
+        BenchmarkId::new("double_selector_mut", "BigJson"),
+        &(t2_selector_mut, t2_selector_mut_two, t2_json),
+        |b, (s, s2, v)| {
+            // Insert a call to `to_async` to convert the bencher to async mode.
+            // The timing loops are the same as with the normal bencher.
+            b.to_async(&runtime).iter_batched(
+                || (s.clone(), s2.clone(), v.clone()),
+                |(s, s2, v)| async {
+                    let v = selector_mut(s, v);
+                    let _ = selector_mut(s2, v);
+                },
+                criterion::BatchSize::LargeInput,
+            );
+        },
+    );
+}
+
+criterion_group!(benches, setup_async_benchmark);
+criterion_main!(benches);