Q: python multiprocessing pool abort if value I am developping a script where I create objects at random but I don't want duplicates. They are stored and each time I create a new one I check it against the already existing ones. As I want to do it for large numbers of objects, I am now trying to parallelize it but so far with no success. I tried some solutions found on the web (mainly here in fact) but still not working.
The idea is that I launch a pool and map my function to it. When a process finds a match, it sets a value to 1. This value is readable by all processes, they can write it using a lock and I need it at the end to return. Thus I made a Lock and a Value, so that all processes can read the value (hence the lock=False) and check if a match has been found in another process. Then I tried something different with an Event and checking if it is set but this is still not working... I then tried to raise a special Exception but still no success in making the code to a success.
Please, I prefer to program OOP, so I will avoid until my last resource to define a global variable as I regard them as unsure (personal opinion).
Here is a MWE, I replaced my complicated objects by int and my stored objects by a range(10000) to help you understand.
#!/usr/bin/env python3

import multiprocessing as muproc


def ParallelCheck(me):
    print(" Testing {}...".format(me))
    #manager = muproc.Manager()
    #lock = manager.Lock()
    lock = muproc.Lock()
    back = muproc.Value("i", 0, lock=False)
    ParChild = ParallelChild(me, lock, back)
    with muproc.Pool() as pool:
        try:
            pool.map(ParChild.run, range(10000))
        except AbortPool:
            pool.terminate()
            print("pool")
    return back.value


def Computation(me, neighbour):
    return me == neighbour



class ParallelChild(object):
    def __init__(self, me, lock, back):
        self.abort = muproc.Event()
        self.lock = lock
        self.me = me
        self.back = back

    def run(self, neighbour):
        print("run")
        if self.abort.is_set():
            print("Aborting")
            pass
        else:
            if Computation(self.me, neighbour):
                self.lock.acquire()
                self.abort.set()
                self.back.value = 1
                print("GOTCHA")
                self.lock.release()
                raise AbortPool
            else:
                print("...")


class AbortPool(Exception):
    #pass
    def __init__(self):
        ## Just to check
        print("AbortPool raised!")



if __name__ == "__main__":
    values = [12000, 13, 7]
    for v in values:
        print("value={} match={}".format(v, ParallelCheck(v)))

Now it yields a RunTimeError:
me@stigepc4$ python3 mwe.py 
 Testing 12000...
Traceback (most recent call last):
  File "mwe.py", line 63, in <module>
    print("value={} match={}".format(v, ParallelCheck(v)))
  File "mwe.py", line 16, in ParallelCheck
    pool.map(ParChild.run, range(10000))
  File "/usr/lib/python3.4/multiprocessing/pool.py", line 260, in map
    return self._map_async(func, iterable, mapstar, chunksize).get()
  File "/usr/lib/python3.4/multiprocessing/pool.py", line 599, in get
    raise self._value
  File "/usr/lib/python3.4/multiprocessing/pool.py", line 383, in _handle_tasks
    put(task)
  File "/usr/lib/python3.4/multiprocessing/connection.py", line 206, in send
    self._send_bytes(ForkingPickler.dumps(obj))
  File "/usr/lib/python3.4/multiprocessing/reduction.py", line 50, in dumps
    cls(buf, protocol).dump(obj)
  File "/usr/lib/python3.4/multiprocessing/sharedctypes.py", line 128, in reduce_ctype
    assert_spawning(obj)
  File "/usr/lib/python3.4/multiprocessing/context.py", line 347, in assert_spawning
    ' through inheritance' % type(obj).__name__
RuntimeError: c_int objects should only be shared between processes through inheritance

I guess it has something to do with the Lock (though the commented Manager but this did not work any better) or with the Value but now idea how to get rid of it...
Edit
As I continue to try to change my code to work the way I want it to, I realize I did not mention what my main problem was. My real difficulty is to have all the process in the pool stop if a match is found. This is what I need so that running parallel is better than serial. For now I can have an event to tell the child run if a match was already found or not, but it keeps looping through the data, even if I raise an exception...
Edit 2
Simply put, I have the following...
for o in objects:
    if too_close(o, existing_objects):
        return 1
return 0

...that I want to distribute among CPUs in something like...
for o in objects:
    if too_close(o, some_existing_objects):
        return 1 and abort other processes
return 0


A: By looking for an answer, I have complicated my script a little too much.
I tried to start with something close to the original doc of
the multiprocessing module.
Then not succeeding I looked for a way to fix it and added some stuff.
I am no expert in python multiprocessing, but after trying out things for a while,
the only way I found to abort the pool.map at the first match is to use an event
so that all processes know it has happen, then they all throw a special exception
to abort themselves.
I can get rid of the Value and the Lock, they are useless for my case.
But my way of doing the things might not be very efficient.
Spawning those processes will cost an amount of computational time,
and each process will copy to its own memory the data it needs to run.
I tried to spawn less process, but each with less data and with
a dataset they will iterate themselves (not letting the pool handling this part).
So I can choose which data goes to which process.
In my example, I split the range(10000) into e.g. 4 processes
each having a range of 2500.
I only want to know if there was a match, thus I can simplify further.
I can set that when a match is found, the event is set and the function returns so that it stops.
The other process test the state of the event, and as soon as it is set they return to stop themselves.
Now back in the main process, at the end I only take a look at the event
(not forgetting to clear it at the beginning, of course).
If it is set, a match was found and it is as simple as that.
The downside is that I must declare the multiprocessing.Event global...
Otherwise when the processes are spawned, each child process will make a copy of it
and they will not be able to communicate between them and with the main process.
But as bj0 already mentioned, it might not be better to parallelize this problem...
After implementing both methods, I compared them to the serial problem and here are my results
for a given case with the same machine:


*

*Serial: 7s

*With a pool: 910s

*With 3 processes having each its own dataset: 97s


So nothing better here... I'll stick to my serial implementation and look for other ways to accelerate things, like other methods than full random...
Here is the last working version of my MWE:
#!/usr/bin/env python3
import multiprocessing as muproc

def ParallelCheck(me):
    print(" Testing {}...".format(me))
    global abort
    abort.clear()
    ParChild = ParallelChild(me)
    jobs = []
    N = 4
    for i in range(N):
        jobs.append(muproc.Process(target = ParChild.run, args=(range(i * 2500, (i+1) * 2500),)))
    for p in jobs:
        p.start()
    for p in jobs:
        p.join()
    if abort.is_set():
        print("MATCH FOUND")
        return 1
    else:
        print(" no match...")
        return 0


def Computation(me, neighbour):
    return me == neighbour


class ParallelChild(object):
    def __init__(self, me):
        self.me = me

    def run(self, neighbours):
        global abort
        for neighbour in neighbours:
            print("{} vs {} by {}".format(self.me, neighbour, self.CurProc()))
            if abort.is_set():
                print("Aborting {}".format(self.CurProc()))
                return 0
            else:
                if Computation(self.me, neighbour):
                    abort.set()
                    print("GOTCHA {}".format(self.CurProc()))
                    return 1

    def CurProc(self):
        return muproc.current_process()._identity[0]



if __name__ == "__main__":
    abort = muproc.Event()
    values = [12000, 130, 7]
    for v in values:
        print("value={} match={}".format(v, ParallelCheck(v)))

