Repository navigation
[16.0][FIX] fs_storage: close dropped SFTP connections - #680
danielduqma wants to merge 3 commits into
Conversation
|
All commit authors signed the Contributor License Agreement. |
Assisted-by: Claude Opus 5.5
a6b8de6 to
05ce516
Compare
| # fsspec filesystems have no close method: only SFTP holds a connection | ||
| client = getattr(fs, "client", None) | ||
| if client is not None and hasattr(client, "get_transport"): | ||
| try: | ||
| fs.ftp.close() | ||
| client.close() | ||
| except Exception: | ||
| _logger.warning("Unable to close the SFTP connection", exc_info=True) |
There was a problem hiding this comment.
Si fs.ftp.close() falla, salta al except y client.close() no llega a ejecutarse, así que la conexión SSH quedaría abierta, y si se comprueba uno por uno tipo:
| # fsspec filesystems have no close method: only SFTP holds a connection | |
| client = getattr(fs, "client", None) | |
| if client is not None and hasattr(client, "get_transport"): | |
| try: | |
| fs.ftp.close() | |
| client.close() | |
| except Exception: | |
| _logger.warning("Unable to close the SFTP connection", exc_info=True) | |
| # fsspec filesystems have no close method: only SFTP holds a connection | |
| client = getattr(fs, "client", None) | |
| if client is not None and hasattr(client, "get_transport"): | |
| for conn in (client, fs.ftp): | |
| try: | |
| conn.close() | |
| except Exception: | |
| _logger.warning("Unable to close the SFTP connection", exc_info=True) |
There was a problem hiding this comment.
Good catch: closing the channel sends a message over the socket, and paramiko raises EOFError when the socket is broken, which is the usual case after a failed check. The client was then left open, with its transport thread.
Fixed in 87f153c: the channel and the client are closed in separate try blocks, channel first. test_drop_filesystem_closes_client_when_channel_fails covers it.
| self._check_connection(self.__fs, self.check_connection_method) | ||
| except Exception as e: | ||
| self.__fs.clear_instance_cache() | ||
| _last_filesystems.pop(self._last_filesystem_key(), None) |
There was a problem hiding this comment.
El pop borra del registro lo que haya para este storage, sin comprobar que sea la conexión de este recordset (self.__fs). Lo mejor seria borrar solo si lo registrado es la conexión que se descarta, no?
key = self._last_filesystem_key() if _last_filesystems.get(key) is self.__fs: del _last_filesystems[key]
There was a problem hiding this comment.
Agreed, a recordset holding a filesystem already replaced by another one could forget the current one.
The suggested check alone would break rooted_dir storages, though: the registry keeps the wrapped SFTP filesystem, while self.__fs is the rooted_dir wrapper, so the entry would never be removed and the broken connection would be neither closed nor replaced.
Fixed in 7982163 with _forget_filesystem, which removes the entry only when it is the failing filesystem or the one it wraps, compared by identity (fsspec filesystems with the same options compare equal). Covered by test_failed_check_on_replaced_filesystem_keeps_current_one and test_failed_check_reconnects_with_directory_path; the latter fails with the suggested check.
There was a problem hiding this comment.
The change looks great to me, thanks
07be92b to
7982163
Compare
7982163 to
93bc858
Compare
fsspec caches its filesystems and never closes them. An SFTP filesystem keeps a paramiko transport and its thread alive until the process dies, so every instance that fs_storage stops using leaks a connection: - when the connection check failed, the whole instance cache of the class was cleared without closing anything, and the next access opened a new connection. With a rooted_dir storage only the wrapper was dropped, so the broken SFTP instance kept coming back; - when the options of a storage changed, the instance built with the old ones stayed cached. fs_storage now remembers the filesystem each storage uses, per process and thread, like the fsspec cache does. A failed check forgets it and drops the failing instance; a change of options drops the previous one. Dropping removes the instance from the cache and closes its connection, unless another storage with the same options still uses it. The cache entry is only removed when it is the dropped instance: a new instance built with the same options gets the same token, and removing it would open a new connection on every access. Likewise, a failed check only forgets the failing filesystem, or the one its rooted_dir wraps, since another recordset may have replaced it meanwhile. The SFTP channel and the client are closed separately: closing the channel raises when the socket is broken, which is the usual reason for a failed check, and the client still has to be closed to stop the transport thread. Assisted-by: Claude Opus 5.5
93bc858 to
ef3db93
Compare
| token = getattr(fs, "_fs_token_", None) | ||
| if type(fs)._cache.get(token) is fs: | ||
| del type(fs)._cache[token] |
There was a problem hiding this comment.
You're dealing with fsspec cache internal.... Could make the code at risk in case of deep change into the fsspec lib.
There was a problem hiding this comment.
Agreed: fsspec has no public API to evict a single instance, only
clear_instance_cache(), which drops every instance of the class without
closing any (that is what leaked). And the internals do move: recent versions
guard _cache with _instantiation_lock.
Done in 63655e9: fs_storage no longer relies on the fsspec instance cache. It
builds its filesystems with skip_instance_cache=True and keeps the one of
each storage per process and thread, building a new one when the options change
or the check fails, and closing the previous one. The only side effect is that
two storages with identical options no longer share a connection.
lmignon
left a comment
lmignon
left a comment
There was a problem hiding this comment.
Thank you for the fix. Just a little comment and a thought.
| # fsspec filesystems have no close method: only SFTP holds a connection | ||
| client = getattr(fs, "client", None) | ||
| if client is not None and hasattr(client, "get_transport"): | ||
| # Closing the channel may fail on a broken connection, and the | ||
| # client still has to be closed to stop the transport thread | ||
| for connection in (fs.ftp, client): | ||
| try: | ||
| connection.close() | ||
| except Exception: | ||
| _logger.warning( | ||
| "Unable to close the SFTP connection", exc_info=True | ||
| ) |
There was a problem hiding this comment.
I would prefer to move this code into a dedicated method _on_filesystem_dropped since it's specific to a client protocol.... and call here: self._on_filesystem_dropped(fs)
There was a problem hiding this comment.
Done in 63655e9: _close_filesystem calls self._on_filesystem_dropped(fs)
for the filesystem and the one a rooted_dir wraps, and the SFTP closing lives
there.
790725b to
63655e9
Compare
Removing a single instance from the fsspec instance cache relied on its internals (_cache, _fs_token_), which change between fsspec versions. fs_storage now builds its filesystems with skip_instance_cache=True and keeps the one of each storage per process and thread, with the options it was built with. It is reused while the options do not change; a new one is built when they do or when the connection check fails, and the previous one is closed. Each storage has its own filesystem, so none is shared with another storage and closing it is always safe. When the options already ask for skip_instance_cache, the caller wants a filesystem of its own and closes it itself, so fs_storage builds it without keeping it or closing the one of the storage. Releasing the connection depends on the protocol, so it moves to the _on_filesystem_dropped hook, which closes SFTP connections and can be overridden for other protocols. Assisted-by: Claude Opus 5.5
63655e9 to
250958b
Compare
fsspec never reuses some filesystems (cachable = False), like FTP ones, whose connections do not survive being idle: it builds a new one on each call. fs_storage kept them anyway in its own cache, so an FTP connection closed by the server was reused by the next job, which failed with BrokenPipeError or EOFError. The connection check did not catch it, since FTP listings are served from the fsspec listings cache. fs_storage now builds a new filesystem on each access for those classes, as it did before keeping its own cache. Assisted-by: Claude Opus 5.5
fsspec caches its filesystems and never closes them. An SFTP filesystem keeps a paramiko transport, and its thread, alive until the process dies. Every SFTP instance that
fs_storagestops using therefore leaks a connection. On long-lived workers this ends withRuntimeError: can't start new thread.Where the connections leak
fscallsclear_instance_cache(), which empties the cache of the class without closing anything, so the next access opens a new connection. With adirectory_path, only therooted_dirwrapper is cleared, and the broken SFTP instance keeps coming back from the cache.Fix
fs_storageremembers which filesystem each storage uses, keyed by process, thread, database and storage. The key matches the fsspec token, so it is safe across worker forks.rooted_dirstorage, the wrapped SFTP filesystem is dropped as well.Connections are still reused across requests and jobs of a worker, as before.
Measured
Live paramiko transports left after each scenario, against a local SFTP server, with fsspec 2026.9.0 and paramiko 5.0.0:
directory_path, 10 accessesdirectory_path, 10 accessesThis change was developed with the assistance of Claude Opus 5.5 (
Assisted-bytrailer in the commit).FL-556-10794