How to properly close resources for async filesystems?

[lobis]

I am using fsspec to concurrently fetch data. This data is being sent into a queue and processed as soon as it's recieved. For this I am using the async api (_cat_file, etc.).

I first began by providing my own loop to fsspec but it led to some race conditions that were difficult to fix. Later I found out fsspec provides its own loop via fsspec.asyn.get_loop so I used this instead. I understand that fsspec creates a daemon thread that manages this loop and properly closes all resources when the main thread finishes.

Below is some code that interfaces with fsspec in way similar to my application. It works but there are some warnings. These warnings sometimes cause the pytest suite to fail so there is something that needs fixing, I just can't find it.

Traceback (most recent call last):
  File "C:\Miniconda3\envs\test\lib\asyncio\sslproto.py", line 320, in __del__
    _warn(f"unclosed transport {self!r}", ResourceWarning, source=self)
ResourceWarning: unclosed transport <asyncio.sslproto._SSLProtocolTransport object at 0x0000018C9D4668C0>

(full log at https://github.com/scikit-hep/uproot5/actions/runs/6616614189/job/17971248151#step:7:880). In this particular case it fails for fsspec_xrootd but I've seen this same error for s3fs.

In the code below if I don't try to close the loop, there are no warnings. However this is not the case in my other code (scikit-hep/uproot5#999) so that is the motivation to learn how these resources are managed.

Thanks!

import fsspec
import fsspec.asyn
import asyncio

url = "http://example.com"


def main():
    loop = fsspec.asyn.get_loop()
    fs = fsspec.filesystem("http")

    futures = []
    ranges = [(0, 100), (50, 55), (200, 400)]
    for start, stop in ranges:
        coroutine = fs._cat_file(url, start=start, end=stop)
        future = asyncio.run_coroutine_threadsafe(coroutine, loop)
        futures.append(future)

    # block until all futures are done
    for future in futures:
        future.result()

    # it appears there are no running tasks, but is this really the case?
    tasks = asyncio.all_tasks(loop)
    assert len(tasks) == 0

    # How do I properly close all resources?

    asyncio.run_coroutine_threadsafe(fs._session.close(), loop).result()
    asyncio.run_coroutine_threadsafe(loop.shutdown_asyncgens(), loop).result()
    loop.call_soon_threadsafe(loop.stop)

    # this does not properly close resources and raises a warning


if __name__ == "__main__":
    main()

[martindurant]

The problem that you are facing is not that there are coroutines waiting to run, but that the showdown code run in the main thread at garbage collection time ideally needs the loop to still be running. I think loop.shutdown_asyncgens() would prevent this.

In any case, it is anticipated that you would either use a filesystem instance in async or blocking mode, not both. If in async more, you would be much better creating the instance within an async context, not what you do here.

async def make_fs():
    return fsspec.filesystem("http")

fs = asyncio.run_coroutine_threadsafe(make_fs, loop).result()

An alternative would be to specifically provide the loop in the constructor (filesystem("http", loop=loop)), but I don't recommend it.

It should be noted that fs.cat_ranges provides what you want in a single step:

starts, stops = zip(*ranges)
out = fs.cat_ranges([url] * len(ranges), starts, stops)

(which also has an async variant).

[lobis]

It should be noted that fs.cat_ranges provides what you want in a single step:

The reason I am not using this method is that I would like to further process the data as soon as it arrives and not wait for all chunks to arrive before processing. I think I could also do this using the fsspec callbacks but I haven't tried it yet. This would be easier to do if our code was fully async but this is not the case yet.

Here is a more detailed code that illustrates this (Python 3.8):

import fsspec
import fsspec.asyn
import asyncio
from queue import Queue
import time
from concurrent.futures import ThreadPoolExecutor

url = "http://example.com"

def process_chunk(chunk):
    time.sleep(1)
    return chunk


def main():
    loop = fsspec.asyn.get_loop()

    async def make_fs():
        return fsspec.filesystem("http")

    # fs = fsspec.filesystem("http")
    fs = asyncio.run_coroutine_threadsafe(make_fs(), loop).result()

    chunks = Queue()
    futures = []
    ranges = [(0, 100), (50, 55), (200, 400)]
    for start, stop in ranges:
        # maybe do this?
        async def coroutine_function():
            fs = await make_fs()
            return await fs._cat_file(url, start=start, end=stop)

        coroutine = fs._cat_file(url, start=start, end=stop)
        # coroutine = coroutine_function()
        future = asyncio.run_coroutine_threadsafe(coroutine, loop)

        def callback(chunks: Queue):
            def f(future):
                chunks.put(future.result())

            return f

        future.add_done_callback(callback(chunks))
        futures.append(future)

    futures_processing = []
    with ThreadPoolExecutor(max_workers=4) as executor:
        n = len(ranges)
        while not n == 0:
            # chunks can be processed as soon as they arrive (potentially in a thread pool)
            chunk = chunks.get()
            future = executor.submit(process_chunk, chunk)
            futures_processing.append(future)
            n = n - 1
            print(n, len(chunk))

    for future in futures_processing:
        future.result()

    asyncio.run_coroutine_threadsafe(loop.shutdown_asyncgens(), loop)
    loop.call_soon_threadsafe(loop.stop)
    loop.close()
l

if __name__ == "__main__":
    time_start = time.time()
    main()
    time_elapsed = time.time() - time_start
    print(f"elapsed time: {time_elapsed:.2f} seconds")

However the shutdown still does not work as I would expect. I think should be able to explicitly close the loop...

[martindurant]

    loop.call_soon_threadsafe(loop.stop)
    loop.close()

I don't suppose the loop ever runs and has the chance to terminate.

You should probably call fs.close_session() at some point, ideally within a coroutine (and make sure it actually runs).

[lobis]

You should probably call fs.close_session() at some point, ideally within a coroutine (and make sure it actually runs).

I tried to call this but it doesn't seem to make a difference. I wish there was some (blocking) method to shut everything fsspec related down. Sadly I'm not sure the code I provided in this discussion fully reproduces my issue since it only happens intermitently (and in python <= 3.10): when running the full test suite some random (unrelated to fsspec) test will fail after running other fsspec test.

[martindurant]

some random (unrelated to fsspec) test will fail after running other fsspec test.

So maybe it's a fixture thing?

[lobis]

some random (unrelated to fsspec) test will fail after running other fsspec test.

So maybe it's a fixture thing?

This post is motivated by an error on the CI for this PR: scikit-hep/uproot5#1012 (this other PR is where the error originally appeared and has some useful information too). I tried to come up with a standalone reproducer: I thought if I learn how to properly close all resources then the error would disappear. However it's been difficult to provide a representative reproducer code.

After many attempts to figure it out, I think this error is only specific to s3fs but I'm not 100% sure. When I disable the tests involving s3fs, the exception disappears consistently. I would bet that s3fs is not properly closing something but I couldn't find out what. I also couldn't reproduce the error on a brand new repository even if I used the exact same code on a new pytest suite so at this point I don't think this error is ever going to show up but still I would like to fix it (so we can add s3fs to out test suite).

I would say the problem is somewhere here: https://github.com/fsspec/s3fs/blob/main/s3fs/core.py#L541-L560 but I couldn't find it.

• The error only shows up after running a pytest test: it can show up in a different test but is reproducible in a single test if you open and close many connections such as in the code below.
• Here is a full traceback of the error: feat: use only loop executor for fsspec source scikit-hep/uproot5#999 (comment).
• I suspect it's related to s3fs because of this "minimal" code I wrote to reproduce it:

def test_open_fsspec_s3_issue():
    fs, path = fsspec.core.url_to_fs("s3://pivarski-princeton/pythia_ppZee_run17emb.picoDst.root")
    # error won't trigger if using the commented filesystem, but will trigger with the filesystem above
    # fs, path = fsspec.core.url_to_fs("github://scikit-hep:scikit-hep-testdata@v0.4.33/src/skhep_testdata/data/uproot-issue121.root")
    # fs, path = fsspec.core.url_to_fs("https://github.com/scikit-hep/scikit-hep-testdata/raw/main/src/skhep_testdata/data/uproot-issue121.root")
    data = fs.cat_file(path, start=0, end=100)

    url = "s3://pivarski-princeton/pythia_ppZee_run17emb.picoDst.root:PicoDst"

    # it will trigger at some iteration
    for handler in [
        uproot.source.s3.S3Source,
        uproot.source.s3.S3Source,
        uproot.source.s3.S3Source,
        uproot.source.s3.S3Source,
        uproot.source.s3.S3Source,
        uproot.source.s3.S3Source,
    ]:
        with uproot.open(
                url,
                anon=True,
                handler=handler,
        ) as f:
            data = f["Event/Event.mEventId"].array(library="np")
            assert len(data) == 8004

In this code fsspec is not used behind the scenes as it's using a different handler (uproot.source.s3.S3Source as opposed to uproot.source.fsspec.FSSpecSource) with uses minio. fsspec is only being called explitcitly in a semingly unrelated way. I tried multiple variations of this code to reproduce the error and if I only use uproot.source.s3.S3Source or just use fsspec then the error does not appear, it must be a combination of both.

[martindurant]

Your S3Source is single-threaded, I think.

S3FileSystem and HTTPFileSystem both use similar thread/loop and close methods, so if the latter is OK but not the former, then the difference is subtle.

Note that S3FileSystem.close_session will only get called at interpreter shutdown time. If you want to run it earlier in your code, you would do

fs.clear_instance_cache()
del fs

(which should ensure that the thread and loop are both still alive)

[lobis]

The S3Source is actually multi-threaded in the code above, but I made it single threaded in the code below (blocking sequential requests in the main thread) and it made no difference.

I added the changes you suggested (also tried with and without gc.collect()) but it didn't make a difference:

def test_open_fsspec_s3_issue():
    fs, path = fsspec.core.url_to_fs("s3://pivarski-princeton/pythia_ppZee_run17emb.picoDst.root")
    data = fs.cat_file(path, start=0, end=100)
    fs.clear_instance_cache()
    del fs

    import gc
    gc.collect()
    
    url = "s3://pivarski-princeton/pythia_ppZee_run17emb.picoDst.root:PicoDst"

    for handler in [
        uproot.source.s3.S3Source,
        uproot.source.s3.S3Source,
        uproot.source.s3.S3Source,
        uproot.source.s3.S3Source,
        uproot.source.s3.S3Source,
        uproot.source.s3.S3Source,
    ]:
        with uproot.open(
                url,
                anon=True,
                handler=handler,
                use_threads=False, # this is new!
        ) as f:
            data = f["Event/Event.mEventId"].array(library="np")
            assert len(data) == 8004

[martindurant]

I don't know :|

You could further try operating on

fsspec.asyn.loop[0]
fsspec.asyn.iothread[0]

but I don't really know what you might do with them. You should at least put logging into the close_session method to make sure it is indeed getting called when expected, and if yes, a little trial and error to see what you can achieve there.