Parallel processing with `jaq_json`

[I-Al-Istannen]

Hey,

I am currently using the Rust API for trying to process slightly larger datasets (a few thousand json files with a few hundred MB total) with a few queries (100+). As this problem is basically embarrassingly parallel, I parallelized my program over the queries, with worker threads each polling queries from a queue, running the query over all inputs and then collecting the results. This leads to the following flow:

1. Read all input files and queries
2. Parse all input files to json (jaq_json's Val)
3. Clone this into each worker thread, so they have the inputs lying around
4. In each worker:
   1. Pick a query to run from the queue
   2. Clone your inputs, run the query against it, keep the result
   3. Loop
5. Collect the results from all queries and return it

This breaks completely at step 3 and 5, as Val is not Send due to the Rc and can not be passed to a different thread. Even if I move the json strings over, and re-parse everything in each worker thread (which takes quite a while…), this problem persists: I can not get the results back from the worker thread!

My first solution was defining my own JSON type that I could convert from and to jaq_json's Val type, but the conversion and dropping costs of that approach lead to an order of magnitude slower execution.
My current solution basically copies jaq_json and replaces every Rc with Arc.

Could I talk you into defining a

#[cfg(feature = "sync-val")]
type Rc = Arc
#[cfg(not(feature = "sync-val"))]
type Rc = Rc

or something similiar to allow users to define this? As cargo features are additive, users might pay a small penalty if any crate in their graph uses parallel processing, but that sounds acceptable to me.

I can PR a feature like that if you want.

[01mf02]

Hi @I-Al-Istannen, thanks for explaining your workflow so well! I am sympathetic to changes that help with using Val in a multi-threaded setting.

Your #[cfg(...)] approach seems like a viable route towards this. I was thinking alternatively to make Val abstract over the type of pointer à la PointerFamily, but I think that this might become a parameter fest quite fast.

Do you know of any other projects that have a similar build-time feature, in order to get inspiration for the feature name?

[I-Al-Istannen]

Yea, I also thought about a generic for this but I think it's going to be a lot more annoying than just configuring it once. The only thing it really buys you is flexibility, and I think that's not really needed. I doubt people are running jaq in-process and have two libraries that depend on parallel and single-threaded execution. The other problem with a generic is that you break every single user in existence and force them have pointless boilerplate, which is likely unpopular.

Do you know of any other projects that have a similar build-time feature, in order to get inspiration for the feature name?

I don't, but searching surfaces https://docs.rs/maybe-sync/latest/maybe_sync/, which seems abandoned but for the same usecase.

---

And good job with the library by the way :) My rust impl takes ~1s (of which 200-300ms are just drop) to crunch through the data — the python impl (single threaded, but still) takes 5 minutes 30s. A single query ran over all files takes >3s.
This means that my rust program can execute all queries over all files three times while Python is busy executing one query across all files using the jq pypi crate and a python loop.

[01mf02]

Yea, I also thought about a generic for this but I think it's going to be a lot more annoying than just configuring it once. The only thing it really buys you is flexibility, and I think that's not really needed. I doubt people are running jaq in-process and have two libraries that depend on parallel and single-threaded execution. The other problem with a generic is that you break every single user in existence and force them have pointless boilerplate, which is likely unpopular.

Yes, I agree with you.

Do you know of any other projects that have a similar build-time feature, in order to get inspiration for the feature name?

I don't, but searching surfaces https://docs.rs/maybe-sync/latest/maybe_sync/, which seems abandoned but for the same usecase.

OK, let me do it.

$ git clone https://github.com/rust-lang/crates.io-index
$ cd crates.io-index/
$ find -mindepth 2 -type f -not -path '*/[@.]*' | xargs cat | jaq 'reduce (inputs | .features | keys[]) as $k ({}; .[$k] += 1)' > ../features.json
$ jaq 'to_entries | sort_by(.value) | from_entries' ../features.json

This gives us a list of all features together with their frequency in today's crates.io index (revision 73827c9cc6f403b6e0872fefa1ec708d56e56448).
Here are the highest-ranking entries:

  "sync": 4693,
  "testing": 4950,
  "static": 4970,
  "rt": 5288,
  "try-runtime": 5350,
  "tls": 5409,
  "server": 5427,
  "vendored-openssl": 5550,
  "enable_reqwest_rustls": 5714,
  "enable_reqwest": 5720,
  "runtime-benchmarks": 5810,
  "parallel": 5851,
  "cli": 6029,
  "test": 6342,
  "wasm": 6707,
  "json": 6871,
  "sqlite": 7345,
  "derive": 7392,
  "client": 7416,
  "all": 7548,
  "async": 7974,
  "serde": 8471,
  "concurrent": 9012,
  "default-https-client": 9559,
  "unstable": 10285,
  "btree-map": 11145,
  "vec-u8": 11145,
  "hash-map": 11150,
  "bytes": 11552,
  "debug": 13814,
  "gated-tests": 15229,
  "nightly": 16760,
  "alloc": 17236,
  "full": 20915,
  "native-tls": 22537,
  "behavior-version-latest": 35628,
  "test-util": 42507,
  "rt-tokio": 48378,
  "rustls": 61400,
  "std": 64157,
  "default": 439863

So "sync" is pretty high up there. "arc" has only 270 entries, "send" 233. So I think that going for the feature "sync" is a good way to go.

Please just wait with a PR until I have merged #284. I hope that it finally happens today. :)

And good job with the library by the way :) My rust impl takes ~1s (of which 200-300ms are just drop) to crunch through the data — the python impl (single threaded, but still) takes 5 minutes 30s. A single query ran over all files takes >3s. This means that my rust program can execute all queries over all files three times while Python is busy executing one query across all files using the jq pypi crate and a python loop.

Wow, that sounds amazing. Thanks for sharing your story here, it is always very encouraging for me to read that my focus on high performance enables some real-world advantage!

However, to be fair to jq, I'm wondering where the Python version leaves the performance at the table ... do you reuse the compiled query for all files in the Python version, or do you recompile each query for each file?

[I-Al-Istannen]

This gives us a list of all features together with their frequency in today's crates.io index (revision 73827c9cc6f403b6e0872fefa1ec708d56e56448).

That's certainly a way :D Neat! Has the same answer, even.

Please just wait with a PR until I have merged #284. I hope that it finally happens today. :)

No worries, I can live with my copy-paste workaround for a good while :P

Wow, that sounds amazing. Thanks for sharing your story here, it is always very encouraging for me to read that my focus on high performance enables some real-world advantage!

However, to be fair to jq, I'm wondering where the Python version leaves the performance at the table ... do you reuse the compiled query for all files in the Python version, or do you recompile each query for each file?

The program part for that is quite simple and I believe what the docs recommend.

def exec_query(input_files: list[str], query: str) -> list[str]:
    compiled = jq.compile(query)

    results = []

    for inp in input_files:
        single = compiled.input_text(inp).all()
        if len(single) == 1 and isinstance(single, list):
            single = single[0]
        results.append(single)

    return results

and results in the following flamegraph

The rust program has the benefit of sharing its data structures with jaq now, which likely helps. It first parses all files in parallel into my sync Val version, then (relatively) cheaply clones that for each new query. The python version needs to parse it from scratch every time, as I can't share my python dict with jq — I need to pass over the json string. But even though, I think the non-parallel json parsing bumps up the runtime by a few hundred ms and is mostly offset by the rust program already printing things instead of just storing them in a variable. Maybe the python wrapper uses the jq API in a particularly inefficient way, though the flamegraph looks okay.

[01mf02]

Just wanted to give you a heads up, I've merged #284. So starting from now, I'll be happy to accept a PR for your sync feature in jaq-json.

[I-Al-Istannen]

Thank you! I will have a look in the evening, if I can manage :)

[I-Al-Istannen]

I have opened a PR :)

Slightly off-topic follows.

---

Oof, that was quite painful. Your main branch has a few invasive changes compared to the last released version and still lacks documentation for all of it, as far as I can see :)

The one problem I could not solve was getting anything out of an Exn in my code @01mf02. It has a private field and the get method is private too. It now appears in the result of filter runs as ValX and I don't see any way to get things out of it? I can't even unwrap it to crash, it is just an opaque handle? Is there any chance you could provide a way to get an Error from it? Otherwise I will just stringify it using Debug and wrap it in my own error, I guess. Previously I would Display my way out of your Error type to get a nice human-readable error message.

This part of the diff feels a bit cumbersome (and was quite annoying to get right, with all the types being deeply nested and type inference completely giving up). Copying over the custom DataKind and things to get it to run with anything also feels like there could be a better solution. Maybe you want to give some advice for these things in the docs.rs example when you release the new version :)

-    let defs = jaq_std::defs()
-        .chain(super::data::defs())
-        .collect::<Vec<_>>();
+    let defs = jaq_std::defs().chain(jaq_json::defs()).collect::<Vec<_>>();
     let loader = Loader::new(defs);
     let modules = loader.load(&arena, query.clone());
     let modules = errors_to_whatever(modules)?;
+    let funs = jaq_std::funs::<DataKind>().chain(jaq_json::funs());
     let filter = Compiler::default()
-        .with_funs(jaq_std::funs().chain(super::data::base_funs()))
+        .with_funs(funs)
         .compile(modules);
     let filter = errors_to_whatever(filter)?;
-    let remaining = RcIter::new(empty());
-    let ctx = Ctx::new([], &remaining);
+    let iter = empty();
+    let iter = Box::new(iter) as Box<dyn Iterator<Item = _>>;
+    let iter = &RcIter::new(iter);
+    let data = Data::new(&filter.lut, iter);
+    let ctx = Ctx::<DataKind>::new(&data, Vars::new(empty()));

[TheIndifferent]

What a perfect timing, I was just dealing with exactly the same issue!

One small question to that: does this change make Filter compliant with Send + Sync so that the filter could be compiled once per service and reused by worker threads?

Also, just for me to understand, are there any downsides to moving to Arc completely and not maintaining Rc as a separate feature? In my head that would reduce maintenance cost without having any significant downsides...

[I-Al-Istannen]

@TheIndifferent Isn't Filter already Send + Sync? This works and docs.rs says the same:

    fn send_sync<T: Send + Sync + Sized>(f: T) {}
    send_sync(filter.clone());

Also, just for me to understand, are there any downsides to moving to Arc completely and not maintaining Rc as a separate feature? In my head that would reduce maintenance cost without having any significant downsides...

Yes! Arc forces rust to use atomic operations to update the reference count, even if only one thread ever accesses it. This is not free, I would expect a few percent of runtime difference but that depends on too many variables to tell. You can benchmark it on your machine, if you want :) Do note that atomic operations also inhibit the compiler's ability to reorder load-store operations and code flow, so you ideally need to test that with a real-world example.

The maintenance cost now is not too burdensome, though. It is just one conditional use Arc as Rc statement.

[I-Al-Istannen]

Ah and btw @TheIndifferent, if you are satisfied that makes two happy users — enough to close this issue als completed, I think :P

[01mf02]

Oof, that was quite painful. Your main branch has a few invasive changes compared to the last released version and still lacks documentation for all of it, as far as I can see :)

I'm sorry to hear that this transition was painful for you. I tried to make things as simple as I could, yet not too simple (following Albert Einstein).
However, I tried to document everything. Did you look at cargo doc --open?

The one problem I could not solve was getting anything out of an Exn in my code @01mf02. It has a private field and the get method is private too. It now appears in the result of filter runs as ValX and I don't see any way to get things out of it? I can't even unwrap it to crash, it is just an opaque handle? Is there any chance you could provide a way to get an Error from it? Otherwise I will just stringify it using Debug and wrap it in my own error, I guess. Previously I would Display my way out of your Error type to get a nice human-readable error message.

For now, you should be able to use the new jaq_core::unwrap_valr function to get a ValR (and thus an Error) out of a ValX. It's also what jaq (the CLI) does.
But perhaps that function could be more discoverable. I could also make Exn::get_err public and add a note about unwrap_valr there. What do you think?

This part of the diff feels a bit cumbersome (and was quite annoying to get right, with all the types being deeply nested and type inference completely giving up). Copying over the custom DataKind and things to get it to run with anything also feels like there could be a better solution. Maybe you want to give some advice for these things in the docs.rs example when you release the new version :)

First, I see that you turbofish the DataKind twice --- are both of these really necessary?

And yes, I also have found type inference around DataT to be ... lacking. However, I have not really been able to find a simpler solution that makes it possible to pass arbitrary data to native filters, which is an important new feature (that I already use for the new repl filter).

I am also aware that copy-pasting DataKind is awkward. However, what is the alternative? I could define something like JustLutAndInputs (analogous to jaq_core::data::JustLut) and expose it, but this does obviously not compose so well with other data. If you have any ideas here, I would be really happy to hear them.

For your concrete use case, if you do not use inputs, perhaps JustLut would be sufficient?

As for giving advice in the documentation: Is the main doctest in lib.rs (minimal example) and the jaq::filter module (fully worked example) not sufficient? Should I perhaps point out these two examples more prominently? If you have some good idea how to document the transition from 2.0 -> 3.0, I am happy to accept a PR. :)

Thanks for your feedback so far, it helps me a lot!

[I-Al-Istannen]

I'm sorry to hear that this transition was painful for you. I tried to make things as simple as I could, yet not too simple (following Albert Einstein).
However, I tried to document everything. Did you look at cargo doc --open?

I did, but apparently not at the right one. I either had docs.rs open (where there is no release yet), it seems, or some outdated version locally. That's on me, it would have probably been a lot smoother with it :)

For your concrete use case, if you do not use inputs, perhaps JustLut would be sufficient?

Yea, that is exactly what I expected to be exported. I just didn't find it, probably due to looking at the wrong docs 🙃 The implementers for Data show JustLut, which is good. Maybe a comment at the Ctx::new method would be nice, but if it is at least linked as an implementing struct that's already something.

First, I see that you turbofish the DataKind twice --- are both of these really necessary?

No, one is enough there. It was just a bit delicate and I didn't try minimizing them after various combinations didn't work.

I am also aware that copy-pasting DataKind is awkward. However, what is the alternative? I could define something like JustLutAndInputs (analogous to jaq_core::data::JustLut) and expose it, but this does obviously not compose so well with other data. If you have any ideas here, I would be really happy to hear them.

If I actually use all the features, having my own context data is okay, I think. It was mostly annoying because I use none of it, making it purely noise :P If I want my own custom data type, passing around a Lut as well is fine imho.

As for giving advice in the documentation: Is the main doctest in lib.rs (minimal example) and the jaq::filter module (fully worked example) not sufficient? Should I perhaps point out these two examples more prominently? If you have some good idea how to document the transition from 2.0 -> 3.0, I am happy to accept a PR. :)

The doctest is nice (if it is up to date, which it is). I have looked at the filter, but only after digging through the source code. Having that called out next to the doctext example would be nice, I think.

For now, you should be able to use the new jaq_core::unwrap_valr function to get a ValR (and thus an Error) out of a ValX. It's also what jaq (the CLI) does.
But perhaps that function could be more discoverable. I could also make Exn::get_err public and add a note about unwrap_valr there. What do you think?

Making Exn::get_err public sounds good, but adding unwrap_valr to the docs of Exn and ValX would likely already help a lot. Then you don't have to expose methods you might not want in your API. I could have found it via the updated doctest example, which I didn't see, but that's not the place I would be looking for it.

Apart from that, the errors when the Lut doesn't match aren't the most intuitive. I first thought the Lut was populated by the context and created a new one to pass into the context. This obviously lead to fun crashes with index-out-of-bounds errors for the lut lookups. That one is on me for not thinking through the execution model enough (or looking at the updated doctest I didn't have). But I wonder if you might want to forego the Default impl or add a comment that the user is likely being a dumbass. Losing the impl is bordering on excessive though, if the user is stupid enough to shoot themselves in the foot, you can't stop them.

---

I think with a few more comments:

• on Exn, ValX linking to unwrap_valr
• on the doctest linking to the filter
• on the Ctx::new linking to JustLut
• maybe on Lut::default telling the user to likely get lost / deleting that impl

things are fine, really. I just had the wrong docs open, which didn't exactly help xD

[01mf02]

@I-Al-Istannen, thanks for your excellent suggestions and for your experience report about funny Lut crashes. That is very valuable feedback to me, and I made a PR to make Lut more failsafe to use (#335).

Are you motivated to improve the docs the way you suggest?

[I-Al-Istannen]

I can update the docs, but I won't have time until (realistically) Sunday :)

[01mf02]

I can update the docs, but I won't have time until (realistically) Sunday :)

Don't worry --- take your time. :)