GenericFileSystem Buffered Copy

Hi all. Love fsspec.

I'm trying to use GenericFileSystem like this:

import fsspec
import fsspec.generic

fs = fsspec.url_to_fs("sftp://username@host")[0]

fsspec.generic.rsync(
    "sftp:///stuff", # only the path necessary here. the username and host and stuff is discarded. just the protocol and path are used
    "gs://bucket/dir1/dir2",  # don't add the trailing slash, unless you want double slashes in your path!
    inst_kwargs={"default_method": "current"},
)

This will currently use the fsspec.generic.GenericFileSystem._copy method which creates a temp file on disk(by default) e.g. sftp -> local -> gs. This is undesirable for my use case.

Assumption 1

When looking at GenericFileSystem, there is a buffering implementation in fsspec.generic.GenericFileSystem._cp_file however, I don't think that method will ever be called because _copy has been implemented(unless it is added to _copy).

Force _cp_file to be used

If I remove _copy(rename to __copy) then _cp_file is in fact called, but there is a problem. The error is that open_async is not implemented by either of the filesystems in my example(sshfs.spec.SSHFileSystem and gcsfs.core.GCSFileSystem)

Assumption 2

This is very confusing to me because both SSHFileSystem and GCSFileSystem both extend fsspec.asyn.AsyncFileSystem but neither of them implement open_async. So, in GenericFileSystem._cp_file when the if hasattr(fs, "open_async") checks are done, they return true because they technically have that attribute/method, but it is not implemented.

Force sync open

If a sync open is forced in GenericFileSystem._cp_file then another error

NotImplementedError: Calling sync() from within a running loop

is found because we are in an async context but trying to call a sync method.

Question

What to do here?

1. Implement my own rsync that doesn't try to be generic? (kind of already did this, heavily based on `GenericFileSystem
2. figure out how to do non-async in an async context, for now, and incur the performance "penalty"
3. implement open_async in both SSHFileSystem and GCSFileSystem

Start with fsspec.generic.GenericFileSystem._copy and _cp_file, then trace how rsync selects the copy path for SSHFileSystem and GCSFileSystem. Check the open_async implementations and the sync() call behavior in an async context. Done should define and verify a supported buffered-copy path that avoids an unnecessary local temporary file.