Serialising a File will also serialise the cache which can grow very large

[phofl]

We've run into this when using Xarray together with Dask. The default way of calling this is like this at the moment:

file_list = []
model = "ACCESS-CM2"
variable = "hurs"
data_dir = f"s3://nex-gddp-cmip6/NEX-GDDP-CMIP6/{model}/historical/r1i1p1f1/{variable}/*.nc"
file_list += [f"s3://{path}" for path in s3fs.glob(data_dir)]

# Opening the files here
files = [s3fs.open(f) for f in file_list]

ds = xr.open_mfdataset(
    files,
    engine="h5netcdf",
    combine="nested",
    concat_dim="time",
    data_vars="minimal",
    coords="minimal",
    compat="override",
    parallel=True,
)

The files are accessed initially to get the meta information before they are serialised. The initial access populates the cache with a lot of data.

This triggers a very large cache for 4 of the 130 files which is pretty bad when serialising things.

Serialising the cache doesn't seem like a great idea generally and specifically for remote file systems. Was this decision made intentionally or is this rather something that hasn't been a problem so far?

Ideally, we could purge the cache before things are serialised in fsspec / the inherited libraries. Is this something you would consider? I'd be happy to put up a PR if there is agreement about this.

[martindurant]

You are right, AbstractBufferedFile could do with a __reduce__ method to control what gets serialised. Certainly the cache should not be serialised, and indeed probably the file should be re-opened from fresh on each deserialise (perhaps with a seek to the correct place). Write-mode files should not be serialisable at all.

Note that OpenFile instances are designed exactly to encapsulate this kind of information without caches and other state - these are what should really be passed around.

Additionally, I don't really understand exactly what you are pickling - the xarray object itself? I don't know that such a use is really anticipated.

[phofl]

Additionally, I don't really understand exactly what you are pickling - the xarray object itself? I don't know that such a use is really anticipated.

No, not xarray objects themselves. Xarray extracts some metadata for netcdf files with the opened file and then this object is put into the graph (that happens in open_mfdataset fwiw). This causes the cache to be populated. I haven't dug through everything in the API yet, so can't tell you exactly when and where this happens

[dcherian]

OpenFile instances

These don't appear to fit the open_mfdataset pattern in the OP.

[martindurant]

Would someone like to write a reasonable reduce function for AbstractBufferedFile?

[phofl]

Yep, I would take a look at this

[jrbourbeau]

Btw, just following up here. If I add a small computation like

ds.hurs.mean(dim=["lon", "lat"]).compute()

to @phofl's original example, the graph is now much smaller with the changes in #1753. With the latest fsspec release, the graph is 213.65 MiB and with the update in PR the graph goes down to 12.40 MiB. That's still larger than I'd like, but I'll take the 17x reduction in size

[lbesnard]

I think the problem is still present with the latest main. I don't have a reproducible code yet, but basically this happens only when my local machine running a coiled cluster to create zarr objects from NetCDF opened with s3fs fails (something as simple as the local machine running out of battery) during the xarray.to_zarr() process.

After this failure, I'm trying to continue writing adding zarr object. I first, have to overwrite the data in some specific regions, and again, there is an increase of memory and bandwith usage of the local machine running a remote cluster as it happened during open_mfdataset example at the top of this issue

region={'time':slice(1120, 1125)}
time_dimension_name='time'
amount_to_pad=0
ds.isel(**{time_dimension_name: indexes}).drop_vars(
                self.vars_to_drop_no_common_dimension, errors="ignore"
            ).pad(**{time_dimension_name: (0, amount_to_pad)}).chunk().compute().to_zarr(
                self.store,
                write_empty_chunks=false,
                region=region,
                compute=True,
                consolidated=True,
                safe_chunks=True,
                mode="r+",
            )

This behaviour happens when i add the .compute(). But then if I don't add this option, I get other errors such as ** ValueError: Specified zarr chunks encoding['chunks']=(5, 500, 500) for variable named 'chl_oci' would overlap multiple dask chunks which wouldn't happen if my original to_zarr() process didn't stop and leading to a "corrupted" zarr dataset.

I'm very aware there are probably many different issues in my message but I find it hard to know which package (s3fs/xarray/dask...) has a bug.

[martindurant]

Debugging dask within xarray is very hard! If you can show that the cache is being serialised using fsspec alone, we can fix it!

[lbesnard]

@martindurant I'm no expert and only notice this when using a combination of remote cluster/dask/s3fs/xarray. I could try to do this locally without dask/remote cluster, but then I have no idea how to spot that the cache is being serialised as I would be running everything on my local machine anyway

[lbesnard]

@martindurant , I created this gist previously to highlight the serialisation issue. https://gist.github.com/lbesnard/97bdf0b4af9fa340e8ef47aa20b3cc93

I installed the latest commit of this repo

fsspec                        2024.10.0.post20+g4a6d573
fsspec-reference-maker        0.0.4

I still have the same behaviour where data is sent back from the remote cluster to my local machine.

If i add default_fill_cache=None such as

        s3_fs = s3fs.S3FileSystem(anon=True, default_fill_cache=None)

then the behaviour is as expected.

However, I still have the same issue in the to_zarr() as highlighted yesterday

[martindurant]

Perhaps testing len(pickle.dumps(file)) before and after some reading?