Repository: dask/dask
File: dask/base.py (Python)

Code before:
                        dict,
                        [
                            [f.name, _unpack(getattr(expr, f.name))]
                            for f in dataclasses.fields(expr)
                        ],
                    ),
                )
            elif is_namedtuple_instance(expr):
                tsk = (typ, *[_unpack(i) for i in expr])
            else:
                return expr

        repack_dsk[tok] = tsk
        return tok

    out = uuid.uuid4().hex
    repack_dsk[out] = (tuple, [_unpack(i) for i in args])

    def repack(results):
        dsk = repack_dsk.copy()
        dsk[collections_token] = quote(results)
        return simple_get(dsk, out)

    return collections, repack


def optimize(*args, traverse=True, **kwargs):
    """Optimize several dask collections at once.

    Returns equivalent dask collections that all share the same merged and
    optimized underlying graph. This can be useful if converting multiple
    collections to delayed objects, or to manually apply the optimizations at
    strategic points.

    Note that in most cases you shouldn't need to call this method directly.

    Parameters
    ----------
    *args : objects
        Any number of objects. If a dask object, its graph is optimized and
        merged with all those of all other dask objects before returning an
        equivalent dask collection. Non-dask arguments are passed through
        unchanged.
    traverse : bool, optional
        By default dask traverses builtin python collections looking for dask
        objects passed to ``optimize``. For large collections this can be
        expensive. If none of the arguments contain any dask objects, set
        ``traverse=False`` to avoid doing this traversal.
    optimizations : list of callables, optional
        Additional optimization passes to perform.
    **kwargs

Review comment (suggestion):
I still haven'f fully figured out what's going on. Even without this clear, the test shows that the collection is released after all but not all the futures... with this clear everything works as expected.
I suspect that the `repack` closure is holding on to a reference to the local frame which keeps the `collections` list alive. I think the `repack` closure is the thing that gets GCed which then releases the graph. That was nasty...

Code after:
                            [f.name, _unpack(getattr(expr, f.name))]
                            for f in dataclasses.fields(expr)
                        ],
                    ),
                )
            elif is_namedtuple_instance(expr):
                tsk = (typ, *[_unpack(i) for i in expr])
            else:
                return expr

        repack_dsk[tok] = tsk
        return tok

    out = uuid.uuid4().hex
    repack_dsk[out] = (tuple, [_unpack(i) for i in args])

    def repack(results):
        dsk = repack_dsk.copy()
        dsk[collections_token] = quote(results)
        return simple_get(dsk, out)

    # The original `collections` is kept alive by the closure
    # This causes the collection to be only freed by the garbage collector
    collections2 = list(collections)
    collections.clear()
    return collections2, repack


def optimize(*args, traverse=True, **kwargs):
    """Optimize several dask collections at once.

    Returns equivalent dask collections that all share the same merged and
    optimized underlying graph. This can be useful if converting multiple
    collections to delayed objects, or to manually apply the optimizations at
    strategic points.

    Note that in most cases you shouldn't need to call this method directly.

    Parameters
    ----------
    *args : objects
        Any number of objects. If a dask object, its graph is optimized and
        merged with all those of all other dask objects before returning an
        equivalent dask collection. Non-dask arguments are passed through
        unchanged.
    traverse : bool, optional
        By default dask traverses builtin python collections looking for dask
        objects passed to ``optimize``. For large collections this can be
        expensive. If none of the arguments contain any dask objects, set
        ``traverse=False`` to avoid doing this traversal.
    optimizations : list of callables, optional