Thoughts about multi-file transfer APIs

[arogozhnikov]

Hi Kyle,
very interesting project!

I've ran some minimal non-async test for downloads (from R2, which is S3-compatible).

                  obstore    s3fs
20 x 100 bytes    2.85 s     2.65 s
20 x 1e6 bytes    4.12 s     3.36 s

In this regime obstore was somewhat slower than s3fs. I initially thought this could be because s3fs authorizes in service only once, but it seems there is difference in download speed as well. As a side comment, overhead for uploading a single small file is large in both cases.

So, why wouldn't I just use async?

Well, there are a couple of points with async:

1. inconvenience: one needs to know if you're already in an event loop, and otherwise create one. Every piece of code should deal with this. That's on python, but still should be mentioned
2. load with asyncs is unpredictable, that's a blocker. My common usecase is to upload/download multiple files at once. And sometimes it is a couple of large files, sometimes many small files, sometimes many large files.
   Running many files in parallel would eat up CPU and saturate outbound channel (happened several times), so there should be some global switch for this operation to use e.g. no more than 12 cores (as in your defaults). But as I understand (correct me if I'm wrong here), every call of obstore operates independently and spawns additional threads in case of multipart upload. So total number of threads can go 12 x n_files currenlty uploaded.

So, my wish would be to have something like

put_many(tuples_of_src_tgt, ...)
get_many(tuples_of_src_tgt, ...)

That would automatically allocate a pool of (12 or what user asks) threads and optimally use them until all objects/chunks are uploaded / downloaded.

LMK if that's too much to ask 😆

[kylebarron]

Interesting; we haven't done any benchmarking of the obstore synchronous API yet. I'm curious if it's the same speed for asyncio.run() around the async API. As it stands, the sync API is just tokio making blocking calls (like asyncio.run) from the Rust side.

Back to the core question around put_many and get_many, I don't think that's likely to happen natively in obstore. I think obstore should stay minimal and as close as possible to the primitives provided by the object_store crate.

However, I plan to publish an 0.1 version of obspec this week, which is a set of protocols that define a sync or async object store API. Ideally, then, people can write utilities for obstore using obspec protocols that work both for obstore and any future API that implements obspec (or a wrapper).

So you could have obspec-pool that wraps an obspec object and implements put_many and get_many.

[arogozhnikov]

As it stands, the sync API is just tokio making blocking calls (like asyncio.run) from the Rust side.

yup, so they are basically independent for each python call. And uncontrollable parallelism is a problem unless you have infinite bandwidth (channel either gets uncontrollably saturated).

So you could have obspec-pool that wraps

Wrappers unfortunately can't fix total control over parallelism - if they could, I'd already have one :) This should be handled within the lib (thus the request)

[kylebarron]

2. Running many files in parallel would eat up CPU and saturate outbound channel

I'm not sure what you mean here. Obstore isn't really doing any CPU work other than essential HTTP overhead.

2. every call of obstore operates independently and spawns additional threads in case of multipart upload

This is not true. Obstore creates threads once and only once at startup/import to power the underlying tokio event loop which communicates with the Python event loop. Then it never again does anything with threads. In #424 we'd like to present an API for users to be able to customize this runtime, but regardless, obstore will never create more threads.

The multipart upload allows up to max_concurrency number of concurrent tasks, but those are not threads.

As it stands, the sync API is just tokio making blocking calls (like asyncio.run) from the Rust side.

yup, so they are basically independent for each python call. And uncontrollable parallelism is a problem unless you have infinite bandwidth (channel either gets uncontrollably saturated).

I don't understand what you mean. You can control the parallelism in the ThreadPoolExecutor.

[arogozhnikov]

but regardless, obstore will never create more threads.

thanks for explanation!

I don't understand what you mean. You can control the parallelism in the ThreadPoolExecutor.

Let me try with an example sketch:

def upload_files(files, targets):
    # Use a ThreadPoolExecutor with 16 threads.
    # Each thread calls obstore’s *sync* upload API.

Each sync call runs its own Rust async loop, which has an internal parallelism of 12. So, uploading 20 large files could result in up to 16 × 12 chunks being uploaded simultaneously. Is that correct?

If instead I used a local asyncio loop and tried to upload ~1000 files, how much parallelism would I get? Does it vary depending on file size?

[kylebarron]

Each sync call runs its own Rust async loop

No, each sync call runs on the same Rust async loop.

Here even if you use 16 threads in Python you'll still have the same number of constant threads on the Rust side, which is right now hard-coded by new_multi_thread in the upstream pyo3-async-runtimes library. This defaults to the number of cores on the machine, but you can set it with the TOKIO_WORKER_THREADS env var. See https://docs.rs/tokio/latest/tokio/runtime/struct.Builder.html#method.worker_threads.

There's only one Rust async loop per process. So you'd have to use multiple processes to get more threads on the Rust side, but tokio doesn't work with fork-based processing anyhow. I haven't tried multiprocessing with obstore at all.

If instead I used a local asyncio loop and tried to upload ~1000 files, how much parallelism would I get? Does it vary depending on file size?

In terms of parallelism you'd still have the same number of threads on the Rust side. But it would likely be a lot more efficient on the Python side because each Rust task is so lightweight that you could drive all of those Rust cores with just one Python core running the Python event loop.

The max_concurrency parameter just determines the number of async tasks to use for uploading at the same time.

[arogozhnikov]

awesome, thanks so much for the pointers! I'll need more digging

[arogozhnikov]

I've tested solution based on ThreadPool. Serial code is slightly faster than multi-threaded. I've tested it when uploading many small files.

    for local, remote in local_remote:
        store.put(path=remote, file=local, mode="create", max_concurrency=4)

vs threads:

    with ThreadPoolExecutor(max_workers=5) as executor:
        futures: list = [
            executor.submit(
                partial(
                    store.put,
                    path=remote,
                    file=local,
                    mode="create",
                    max_concurrency=4,
                )
            )
            for local, remote in local_remote
        ]
        futures_wait(futures)  # Blocks until all are done

[arogozhnikov]

async is fast though.
I guess GIL isn't released in these functions

[kylebarron]

async is fast though. I guess GIL isn't released in these functions

GIL should be released in all sync functions. It accidentally wasn't being released in put functions and was fixed in #451. Use latest main if you want to test obstore uploads with the GIL released.

I'd like to get a release out next week

[arogozhnikov]

thank you @kylebarron , that would be great.

One more potentially relevant bug (appeared using obstore 0.6.0 + async API):

Fatal Python error: PyGILState_Release: thread state 0x7618a07014e0 must be current when releasing
0: Python runtime state: finalizing (tstate=0x0000000000a691d8)

[kylebarron]

One more potentially relevant bug (appeared using obstore 0.6.0 + async API):

Fatal Python error: PyGILState_Release: thread state 0x7618a07014e0 must be current when releasing
0: Python runtime state: finalizing (tstate=0x0000000000a691d8)

Would you be able to make a reproducible example and post it in a new issue?