pickles break while "multiprocess"-ing dots

[sergpolly]

more of a docs/reminder rather than an issue

1. While trying to split call-dots into modular steps, returned to scroing_step by @nvictus in call-dots.
   Testing on "big"-data yielded somewhat familiar multiprocessing/pickle error:

multiprocess.pool.MaybeEncodingError: Error sending result: '[ 
...
[dump of ~500-1,000 pd.DataFrame-s, ~500,000 rows by 30 columns each ]
...
Reason: 'error("'i' format requires -2147483648 <= number <= 2147483647",)'

2. For the same input parameters it breaks here: https://github.com/mirnylab/cooltools/blob/441a84ab6c1efd3bcd29de6cfd6ee78551873478/cooltools/cli/call_dots.py#L302 , but not here: https://github.com/mirnylab/cooltools/blob/441a84ab6c1efd3bcd29de6cfd6ee78551873478/cooltools/cli/call_dots.py#L404 , as these objects are slices of histograms and are way smaller than 500,000X30
3. @Phlya observed same or similar issue even for https://github.com/mirnylab/cooltools/blob/441a84ab6c1efd3bcd29de6cfd6ee78551873478/cooltools/cli/call_dots.py#L404 or https://github.com/mirnylab/cooltools/blob/441a84ab6c1efd3bcd29de6cfd6ee78551873478/cooltools/cli/call_dots.py#L482 while running "modern" call-dots instance that didn't use @nvictus -'s scoring_step. I could not find a corresponding issue anywhere.
4. apparently pickle is calculating total number of elements - looks like it does columns*rows by the number of dataframes, otherwise math does not work out (>=2147483647 ). Is it indeed the case @nvictus @mimakaev @golobor ? what if it were to be a bunch of string-s of total length >2bln ?! https://stackoverflow.com/questions/47776486/python-struct-error-i-format-requires-2147483648-number-2147483647 - indeed says something about calculating elements in each objects ...
5. I'll work around this BS for testing etc purposes but we should address it eventually - dask ? or at least multipro-something that is using dill - @nvictus ?
6. Also, looks like it was finally fixed bpo-17560: Too small type for struct.pack/unpack in mutliprocessing.Connection python/cpython#10305 ?! can anyone more knowledgeable confirm @nvictus @mimakaev @golobor ?

[nvictus]

This is caused by multiprocessing's default use of pickle protocol v3 for transferring data between processes, which encodes its payload size as a signed 32-bit integer i. This was supposed to be changed in 3.7, maybe it won't be until 3.8. The better solution would be to register custom serializers -- it doesn't look like that patch is ready yet, but there does seem to be some undocumented workaround.

multiprocess is a fork of multiprocessing that replaces pickle with dill, but in my experience it doesn't seem to be able to override the protocol version, which is likely a bug similar to this one.

In any case, we should also look into shedding some weight by scrutinizing how much data is getting serialized and sent to worker processes:

chunks = map_(job, tiles, **map_kwargs)

I doubt that the individual tiles can be made lighter, but there seems to be heavy stuff going into in the job partial function. e.g. the expected dataframe (which is probably the culprit) could be passed as a file path and reloaded on the worker's side, rather than pickled and sent to the worker over a socket.

[sergpolly]

• looks's like Python 3.8 https://bugs.python.org/issue17560 ?!

In any case, we should also look into shedding some weight by scrutinizing how much data is getting serialized and sent to worker processes:

so, input counts as well ?!
reading a file with expected ~1,000 times ? i.e. is it as many times as there are chunks or as many as there are workers ? Apparently I have no idea how multiprocess(ing) works ...
Maybe we could to_hdf in each process ... or there is no way to preserve order that way ?

apparently pickle is calculating total number of elements - looks like it does columns*rows by the number of dataframes, otherwise math does not work out

it is number of bytes, right ? not columns*row by whatever. it's 2GB limit on what is being sent ...

this turns out to be a bit more complicated/delicate thing.

[nvictus]

reading a file with expected ~1,000 times ?

Well, not all at once! Plus, I doubt it will be less efficient than pickling + IPC + unpickling 1000 times.
Though, you're right that this only needs to be done once for each worker, because workers persist between jobs. There isn't a clean API for doing this with the Pool interface, though it can be hacked using global variables.

it is number of bytes, right ?

Yup.

Another possibility to consider is the concurrent.futures.ProcessPoolExecutor API.

[nvictus]

The undocumented patch seems to work on 3.6

import multiprocessing as mp

class ForkingPickler4(mp.reduction.ForkingPickler):
    def __init__(self, *args):
        if len(args) > 1:
            args[1] = 4
        else:
            args.append(4)
        super().__init__(*args)

    @classmethod
    def dumps(cls, obj, protocol=4):
        print("USING VERSION 4!!!")
        return mp.reduction.ForkingPickler.dumps(obj, protocol)


class Pickle4Reducer(mp.reduction.AbstractReducer):
    ForkingPickler = ForkingPickler4
    register = ForkingPickler4.register
    def dump(self, obj, file, protocol=4):
        ForkingPickler4(file, protocol).dump(obj)
 

ctx = mp.get_context()
ctx.reducer = Pickle4Reducer()


def foo(x):
    print(x)

with mp.Pool(3) as pool:
    pool.map(foo, ['a', 'b', 'c'])

USING VERSION 4!!!
USING VERSION 4!!!
USING VERSION 4!!!
a
USING VERSION 4!!!
b
USING VERSION 4!!!
c
USING VERSION 4!!!
USING VERSION 4!!!
USING VERSION 4!!!
USING VERSION 4!!!
USING VERSION 4!!!
USING VERSION 4!!!

[sergpolly]

It's not about efficiency, though, - just to overcome 2GB limitation, right ?

I found another weak spot in the current implementation #51 - need to stop annotating chunks, and overall trim them down.

2GB is still too small for us, as we might be passing ~kernelsXsize_of_cooler in a normal/typical(?) situtation, probably even smaller, as we usually process a fraction of the heatmap (~x mbases around diagonal) - but we still need that patch or something like that

I'd personally focus on optimizations - trimming dataframes down, because multiprocess(ing) business is harder for me - it also seems secondary now after we rediscovered #51

[sergpolly]

this bites again ... strikes back ...
in the one-pass-dots implementation everything breaks because we're trying to return pieces of the final DataFrame with the enrichment/adjust-expected scores for the surveyed pixels and that could be a lot of data: https://github.com/mirnylab/cooltools/blob/969e766fcf596916179bd4a8872f54d77636b6c8/cooltools/dotfinder.py#L1201

Observations:

• using smaller tiles does not help.
• asking for more workers - helps! - i.e. splitting large # of tiles over larger Pool ...

From that I can understand that:
it is the # of bytes to be serialized per worker - that matters - i.e. breaks during pickling/dilling ...
not the # of bytes per tile ... https://github.com/mirnylab/cooltools/blob/969e766fcf596916179bd4a8872f54d77636b6c8/cooltools/dotfinder.py#L1245

I'll describe other thoughts about the whole one-pass-dots situation in #51 - it seems unsustainable to me right now

[Mittmich]

I stumbled upon this issue when doing pileups and for me, it was solved by updating to python 3.8 (python/cpython#10305). Would that be possible for this use case as well?

[golobor]

@sergpolly - is this still an issue?