scatter failed when putting into storage function that closed the class containing the Dask Future

[anmyachev]

Describe the issue:

Interesting detail: distributed.protocol.serialize can serialize a function, but scatter cannot. Based on this, it seems that the scatter should work in this case too.

Minimal Complete Verifiable Example:

import distributed
from distributed import Client, default_client

if __name__ == '__main__':
    client = Client()

    class Custom:
        @classmethod
        def _construct(cls, data):
            return cls(data)
        def __init__(self, data):
            self.data = data
        def __reduce__(self):
            client = default_client()
            return self._construct, (client.gather(self.data),)

    data = Custom(client.scatter(6))
    def normal_function():
        return data.data

    print(distributed.protocol.deserialize(*distributed.protocol.serialize(normal_function))())  # works
    print(client.scatter(normal_function)())  # TypeError: ('Could not serialize object of type function', '<function normal_function at 0x00000207B4EE20D0>')

Anything else we need to know?:

Environment:

• Distributed version: 2023.12.1
• Python version: 3.9.18
• Operating System: Windows 11
• Install method (conda, pip, source): conda