# Using dill to pickle anything

ipyparallel doesn’t do much in the way of serialization. It has custom zero-copy handling of numpy arrays, but other than that, it doesn’t do anything other than the bare minimum to make basic interactively defined functions and classes sendable.

There are a few projects that extend pickle to make just about anything sendable, and one of these is dill. Another is cloudpickle.

To install dill:

    pip install dill

First, as always, we create a task function, this time with a closure

    def make_closure(a):
        """make a weird function with a closure on an open file, and return it"""
        import os

        f = open('/tmp/dilltest', 'a')

        def has_closure(b):
            product = a * b
            f.write(f"{os.getpid()}: {product:g}\n")
            f.flush()
            return product

        return has_closure

    !rm -f /tmp/dilltest
    closed = make_closure(5)

    closed(2)

    10

    !cat /tmp/dilltest

    25810: 10

    import pickle

Without help, pickle can’t deal with closures

    pickle.dumps(closed)

    ---------------------------------------------------------------------------
    AttributeError                            Traceback (most recent call last)
    Cell In[6], line 1
    ----> 1 pickle.dumps(closed)

    AttributeError: Can't pickle local object 'make_closure.<locals>.has_closure'

But after we import dill, magic happens

    import dill

    dill.dumps(closed)[:64] + b'...'

    b'\x80\x04\x95E\x03\x00\x00\x00\x00\x00\x00\x8c\ndill._dill\x94\x8c\x10_create_function\x94\x93\x94(h\x00\x8c\x0c_create_code\x94\x93...'

So from now on, pretty much everything is pickleable.

## Using dill in IPython Parallel

As usual, we start by creating our Client and View

    import ipyparallel as ipp

    cluster = ipp.Cluster(n=2)
    cluster.start_cluster_sync()
    rc = cluster.connect_client_sync()
    rc.wait_for_engines(n=2)
    view = rc.load_balanced_view()

    Starting 2 engines with <class 'ipyparallel.cluster.launcher.LocalEngineSetLauncher'>

Now let’s try sending our function with a closure:

    view.apply_sync(closed, 3)

    ---------------------------------------------------------------------------
    TypeError                                 Traceback (most recent call last)
    Cell In[10], line 1
    ----> 1 view.apply_sync(closed, 3)

    File ~/dev/ip/parallel/ipyparallel/client/view.py:228, in View.apply_sync(self, _View__ipp_f, *args, **kwargs)
        224 def apply_sync(self, __ipp_f, *args, **kwargs):
        225     """calls ``f(*args, **kwargs)`` on remote engines in a blocking manner,
        226     returning the result.
        227     """
    --> 228     return self._really_apply(__ipp_f, args, kwargs, block=True)

    File ~/.virtualenvs/ipp7-stable/lib/python3.11/site-packages/decorator.py:232, in decorate.<locals>.fun(*args, **kw)
        230 if not kwsyntax:
        231     args, kw = fix(args, kw, sig)
    --> 232 return caller(func, *(extras + args), **kw)

    File ~/dev/ip/parallel/ipyparallel/client/view.py:55, in sync_results(f, self, *args, **kwargs)
         53 self._in_sync_results = True
         54 try:
    ---> 55     ret = f(self, *args, **kwargs)
         56 finally:
         57     self._in_sync_results = False

    File ~/.virtualenvs/ipp7-stable/lib/python3.11/site-packages/decorator.py:232, in decorate.<locals>.fun(*args, **kw)
        230 if not kwsyntax:
        231     args, kw = fix(args, kw, sig)
    --> 232 return caller(func, *(extras + args), **kw)

    File ~/dev/ip/parallel/ipyparallel/client/view.py:39, in save_ids(f, self, *args, **kwargs)
         37 n_previous = len(self.client.history)
         38 try:
    ---> 39     ret = f(self, *args, **kwargs)
         40 finally:
         41     nmsgs = len(self.client.history) - n_previous

    File ~/dev/ip/parallel/ipyparallel/client/view.py:1364, in LoadBalancedView._really_apply(self, f, args, kwargs, block, track, after, follow, timeout, targets, retries)
       1359 follow = self._render_dependency(follow)
       1360 metadata = dict(
       1361     after=after, follow=follow, timeout=timeout, targets=idents, retries=retries
       1362 )
    -> 1364 future = self.client.send_apply_request(
       1365     self._socket, f, args, kwargs, track=track, metadata=metadata
       1366 )
       1368 ar = AsyncResult(
       1369     self.client,
       1370     future,
       (...)
       1373     owner=True,
       1374 )
       1375 if block:

    File ~/dev/ip/parallel/ipyparallel/client/client.py:1954, in Client.send_apply_request(self, socket, f, args, kwargs, metadata, track, ident, message_future_hook)
       1951 if not isinstance(metadata, dict):
       1952     raise TypeError(f"metadata must be dict, not {type(metadata)}")
    -> 1954 bufs = serialize.pack_apply_message(
       1955     f,
       1956     args,
       1957     kwargs,
       1958     buffer_threshold=self.session.buffer_threshold,
       1959     item_threshold=self.session.item_threshold,
       1960 )
       1962 future = self._send(
       1963     socket,
       1964     "apply_request",
       (...)
       1970     message_future_hook=message_future_hook,
       1971 )
       1972 msg_id = future.msg_id

    File ~/dev/ip/parallel/ipyparallel/serialize/serialize.py:182, in pack_apply_message(f, args, kwargs, buffer_threshold, item_threshold)
        174 kwarg_bufs = list(
        175     chain.from_iterable(
        176         serialize_object(kwargs[key], buffer_threshold, item_threshold)
        177         for key in kw_keys
        178     )
        179 )
        181 info = dict(nargs=len(args), narg_bufs=len(arg_bufs), kw_keys=kw_keys)
    --> 182 msg = serialize_object(f)
        183 msg.append(pickle.dumps(info, PICKLE_PROTOCOL))
        184 msg.extend(arg_bufs)

    File ~/dev/ip/parallel/ipyparallel/serialize/serialize.py:113, in serialize_object(obj, buffer_threshold, item_threshold)
        110     cobj = can(obj)
        111     buffers.extend(_extract_buffers(cobj, buffer_threshold))
    --> 113 buffers.insert(0, pickle.dumps(cobj, PICKLE_PROTOCOL))
        114 return buffers

    TypeError: cannot pickle '_io.TextIOWrapper' object

Oops, no dice. For IPython to work with dill, there are one or two more steps. IPython will do these for you if you call `DirectView.use_dill`:

    rc[:].use_dill()

    <AsyncResult(use_dill): pending>

Now let’s try again

    view.apply_sync(closed, 3)

    15

    !cat /tmp/dilltest

    25810: 10
    26717: 15

Yay! Now we can use dill to allow ipyparallel to send anything.

And that’s it! We can send closures, open file handles, and other previously non-pickleables to our engines.

Let’s give it a try now:

    remote_closure = view.apply_sync(make_closure, 4)
    remote_closure(5)

    20

    !cat /tmp/dilltest

    25810: 10
    26717: 15
    25810: 20

But wait, there’s more!

At this point, we can send/recv all kinds of stuff

    def outer(a):
        def inner(b):
            def inner_again(c):
                return c * b * a

            return inner_again

        return inner

So outer returns a function with a closure, which returns a function with a closure.

Now, we can resolve the first closure on the engine, the second here, and the third on a different engine, after passing through a lambda we define here and call there, just for good measure.

    view.apply_sync(lambda f: f(3), view.apply_sync(outer, 1)(2))

    6

And for good measure, let’s test that normal execution still works:

    %%px
    foo = 5

    print(rc[:]['foo'])
    rc[:]['bar'] = lambda: 2 * foo
    rc[:].apply_sync(ipp.Reference('bar'))

    [5, 5]

    [10, 10]

And test that the `@interactive` decorator works

    %%file testdill.py
    import ipyparallel as ipp

    @ipp.interactive
    class C:
        a = 5

    @ipp.interactive
    class D(C):
        b = 10

    @ipp.interactive
    def foo(a):
        return a * b

    Writing testdill.py

    import testdill

    v = rc[-1]
    v['D'] = testdill.D
    d = v.apply_sync(lambda: D())  # noqa: F821
    print(d.a, d.b)

    5 10

    v['b'] = 10
    v.apply_sync(testdill.foo, 5)

    50