One of the main reasons for doing this is that the current version of cloudpickle that we vendor does not properly serialize keyword only arguments in functions.

Test FAILed.
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/Ray-PRB/16813/
Test FAILed.

jenkins, retest this please

Jenkins retest this please

Test FAILed.
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/Ray-PRB/16825/
Test FAILed.

Test FAILed.
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/Ray-PRB/16827/
Test FAILed.

Here's one easy patch to the original cloudpickle:

cloudpipe/cloudpickle@b9dc17f

    def save_codeobject(self, obj):
        """
        Save a code object
        """
        if PY3:  # pragma: no branch
            if hasattr(obj, "co_posonlyargcount"):  # pragma: no branch
                args = (
                    obj.co_argcount, obj.co_posonlyargcount,
                    obj.co_kwonlyargcount, obj.co_nlocals, obj.co_stacksize,
                    obj.co_flags, obj.co_code, obj.co_consts, obj.co_names,
                    obj.co_varnames, obj.co_filename, obj.co_name,
                    obj.co_firstlineno, obj.co_lnotab, obj.co_freevars,
                    obj.co_cellvars
                )
            else:
                args = (
                    obj.co_argcount, obj.co_kwonlyargcount, obj.co_nlocals,
                    obj.co_stacksize, obj.co_flags, obj.co_code, obj.co_consts,
                    obj.co_names, obj.co_varnames, obj.co_filename,
                    obj.co_name, obj.co_firstlineno, obj.co_lnotab,
                    obj.co_freevars, obj.co_cellvars
                )
        else:
            args = (
                obj.co_argcount, obj.co_nlocals, obj.co_stacksize, obj.co_flags, obj.co_code,
                obj.co_consts, obj.co_names, obj.co_varnames, obj.co_filename, obj.co_name,
                obj.co_firstlineno, obj.co_lnotab, obj.co_freevars, obj.co_cellvars
            )
        self.save_reduce(types.CodeType, args, obj=obj)

@suquark we should update the cloudpickle entirely rather than monkey patch it.

Test FAILed.
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/Ray-PRB/16884/
Test FAILed.

Test PASSed.
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/Ray-PRB/16887/
Test PASSed.

@ericl Do you foresee anything else breaking? It's important to merge this so @suquark can move forward with #5611

Thanks @richardliaw, I think we should merge it and try it out. This behavior of reusing existing classes instead of creating a new one seems dangerous to me (e.g., this means that mutations to the class will affect the deserialized object, right?).

@suquark doesn't cloudpickle have an option to turn off this behavior?

Test FAILed.
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/Ray-PRB/16894/
Test FAILed.

Test FAILed.
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/Ray-PRB/16895/
Test FAILed.

jenkins retest this please

@richardliaw just to clarify, you are saying that cloudpickle's behavior has changed between the previous version and this one, right?

Yeah

Test PASSed.
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/Ray-PRB/16910/
Test PASSed.

@robertnishihara I don't think cloudpickle has such options. This is essential for cloudpickle to work correctly with dynamic class between remote processes.

For example,

def main():
    def DynamicClass:
        pass

    @ray.remote
    def remote_function():
        return DynamicClass()

    # register a serializer (assume broadcast to all workers)
    cloudpickle.dispatch[DynamicClass] = ....

    ray.get(remote_function.remote())

In this case, if cloudpickle doesn't keep the existing dynamic class, then when it serializes the DynamicClass() object (the return value of the remote function), it will try to look up the dispatch table for the correct serializer. However, the returned object's class has been changed, because after DynamicClass itself was serialized and deserialized, Python will not think it is the original type (because types are compared by their pointers), even in this case their qualname and content match. With a different type, the correct serializer will not likely to be triggered. So cloudpickle creates a class_tracker_id for each dynamic class, and will reuse the class for instances of dynamic classes.

Pyarrow also did in the same ways but with class_id. So the risk could be low.