s3: s3fs migration

[isidentical]

Roadmap;

• A basic S3FileSystem class like the rest of providers
• Passing the bare minimum tests (apparently moto doesn't work well with aiobotocore when used in wrapper mode, support moto client wrappers aio-libs/aiobotocore#755, so we probably would have to rewrite the test remote using the server mode)
• Porting ~/.aws/config parsing
• Permission/grant mangement
• Port functional/unit tests (and drop the redundant ones)
• etag-preserving server side copy implementation (possibly by overriding the s3fs's copy method while preserving the underlying logic) (copy: implement preserve_etag mode fsspec/s3fs#441)
• support exception unrolling ( Prevent recursion on error translation  fsspec/s3fs#443)
• s3fs doesn't support 3.6 (py3.6 >= support fsspec/s3fs#445) [perhaps we can have a fork for 3.6 only and have it on pypi until we also drop the support, EOL is in 9 months]
• try to setup dvc bench
• end to end performance testing
• Collecting issues regarding s3 on the bug tracker and checking whether they still apply or not

Future;

• should we continue to wrap boto-based exceptions (e.g NoBucketFound etc), using FileNotExistError/PermissionError will offer better handling, deferred.
• Look into whether we can preserve / port the walk_files(prefix=) optimization, .find() can do it in theory though due to some limitations and my limited understanding over that part, this is also deferred (and possibly should be considered in s3: performance implications of s3fs #5969)

[isidentical]

It seems like the naive approach on s3fs (using the FSSpecWrapper without any additions) is about 1.5-3x slower than the current master of simple push/pull/status

Some stats regarding API calls;

       master         s3fs
GET     160            103
PUT      81             89
HEAD    132            278
POST      2              3
        375            473

Looking at the possible optimizations 👀

[isidentical]

Here are some new numbers (real timings instead of request amounts), it appears to be that pull of small files is extremely slow (25x);
https://gist.github.com/isidentical/e0c33a5db41aca10509fcb5c95489247

[isidentical]

Here is the stats for master / s3fs / s3fs via using put_file/get_file APIs

	master	s3fs	s3fs with download/upload APIs
push (single big)	8.30 min	12 min	9.20 min
push (10k small)	5.40 min	24 min	15.20 min
pull (single big)	1.34 min	10.50 min	3 min
pull (10k small)	11.40 min	~4 hours	~1.5 hours

It seems like the next biggest issue is the concurrency. This might be the result of us using a lot of threads + fsspec wrapping up each call to an async function (literally every function) with a per-thread lock 👀
https://github.com/intake/filesystem_spec/blob/9a5d56f221873aa5a3acc606cbabf07590f7be6d/fsspec/asyn.py#L37-L50

[isidentical]

I've re-implemented the push/pull logic in async way to see how much it affects, and it appears to be that this made the pull only 2x slower (compared to 9x slower in the previous version) though the push is now 4x slower (instead of 3x);

	master	s3fs	s3fs with download/upload APIs	ASYNC s3fs with download/upload APIs
push (single big)	8.30 min	12 min	9.20 min	9.15 min
push (10k small)	5.40 min	24 min	15.20 min	~22.30 min
pull (single big)	1.34 min	10.50 min	3 min	2.30 min
pull (10k small)	11.40 min	~4 hours	~1.5 hours	22 min

I am not sure about how much time we want to allocate on this, though I'll give it a shot on implementing a way that combines multiprocessing (not threading) + async and try to see with that. It appears fsspec instances care about serialisability but I am not sure about the rest of the dvc code (@efiop mentioned some parts are not serializable, yet IIRC). We might need to create a filesystem per process to avoid some potential problems.

[efiop]

@isidentical Another option might be to for now convert our current implementation to comply with fsspec, so that we could play around with swapping it for s3fs a bit later, after we fully migrate the rest of fs/*.py to fsspec.

[dberenbaum]

Would this be a good use case to consider merging to a non-master branch to avoid needing to convert the current implementation?

[isidentical]

We decided to put s3fs migration on hold, and use a temporary sync fs like our current S3FileSystem for the time being. There are multiple issues present, including py3.6 being not supported, pull/push on big files are 0.10x~ slow, pull/push on directories with huge amounts of small files are 2-3x slow (at best, with an async pull/push system).

[isidentical]

With #5683's approach now push/pull are the same (as well as status / gc), will do the same tests with partial data and copy on big files...
https://gist.github.com/isidentical/c2c97798681ed574525017b25009cec8

[martindurant]

Can you show me the code for non-async s3fs up/downloads, to see if there's any obvious improvement I can see?

[isidentical]

Can you show me the code for non-async s3fs up/downloads, to see if there's any obvious improvement I can see?

The code was actually just a wrapper around fs.put_file/fs.get_file and then we basically submit those calls to the thread pool (func below refers to one of put_file/get_file)
https://github.com/iterative/dvc/blob/4f8cfc025b429e2e0c31c3152f380940a3b6318b/dvc/remote/base.py#L382-L389

and the async version of that code looked like this;

    def _create_async_tasks(self, executor, jobs, func, pbar, file_plans):
        fails = 0
        jobs *= 4
        file_plan_iterator = iter(file_plans)

        def create_taskset(amount):
            return [
                func(from_info, to_info, name)
                for from_info, to_info, name, _ in itertools.islice(
                    file_plan_iterator, amount
                )
            ]

        async def runner(func=func, file_plans=file_plans):
            fails = 0
            tasks = create_taskset(jobs)
            while tasks:
                fails += sum(await asyncio.gather(*tasks))
                pbar.update(len(tasks))
                tasks = create_taskset(jobs)
            return fails

        fails = asyncio.run(runner())
        self.fs.invalidate_async_fs()
        return fails

(invalidate_async_fs just removes the cached fsspec instance, since the event loop is closed we need to re _connect with it)

[martindurant]

OK, so the standard get and put will actually work on whole lists of files, using asyncio.gather internally. You should not need to submit anything to an executor, nor throttle the number of jobs

from_list, to_list = get_all_files()  # from your iterator of create_taskset
s3.put(from_list, to_list)

Similarly, when in an async context, you can

await s3._put(from_list, to_list)