From 5d5b7a4031994e897cd5b4e298bc20341887b816 Mon Sep 17 00:00:00 2001 From: Pablo Galindo Date: Thu, 18 Oct 2018 22:18:11 +0100 Subject: [PATCH 01/32] bpo-28053: Complete and fix custom reducers in multiprocessing. --- Doc/library/multiprocessing.rst | 71 ++++++++++++++++++++++++ Lib/multiprocessing/connection.py | 17 +++--- Lib/multiprocessing/context.py | 6 +- Lib/multiprocessing/forkserver.py | 7 ++- Lib/multiprocessing/heap.py | 7 ++- Lib/multiprocessing/managers.py | 7 ++- Lib/multiprocessing/popen_forkserver.py | 9 +-- Lib/multiprocessing/popen_spawn_posix.py | 7 ++- Lib/multiprocessing/popen_spawn_win32.py | 9 +-- Lib/multiprocessing/queues.py | 9 ++- Lib/multiprocessing/resource_sharer.py | 6 +- Lib/multiprocessing/spawn.py | 14 +++-- Lib/test/_test_multiprocessing.py | 56 +++++++++++++++++++ 13 files changed, 179 insertions(+), 46 deletions(-) diff --git a/Doc/library/multiprocessing.rst b/Doc/library/multiprocessing.rst index 492f94c30017f0..e797b65162c9c9 100644 --- a/Doc/library/multiprocessing.rst +++ b/Doc/library/multiprocessing.rst @@ -1193,6 +1193,77 @@ For example: the data in the pipe is likely to become corrupted, because it may become impossible to be sure where the message boundaries lie. +Custom Reduction +~~~~~~~~~~~~~~~~ + +.. currentmodule:: multiprocessing + +Several primitives of the :mod:`multiprocessing` module such as +:class:`multiprocessing.Queue`, :class:`multiprocessing.connection.Listener` or +:class:`multiprocessing.connection.Server` need to serialize and deserialize Python +objects to communicate between processes. Sometimes is useful to control what +serialization is to be used for the transport of data for supporting communication with +different versions of Python, use more performant 3rd party tools or custom strategies. + +For this pourpose a set of hooks is available to provide alternate implementations of +the reduction mechanism: + +.. currentmodule:: multiprocessing.reduction + +.. class:: AbstractReducer() + + Abstract base class for use in implementing a Reduction class suitable for use + in replacing the standard reduction mechanism used in multiprocessing. + + .. class:: ForkingPickler + + A subclass of :class:`pickle.Pickler` to be used by multiprocessing. + + .. function:: dump(obj, file, protocol=None) + + Replacement for pickle.dump() to be used by multiprocessing. + + .. function:: register(cls, type, reduce) + + Register a reduce function for a type to be used by multiprocessing. + +.. currentmodule:: multiprocessing + +.. function:: set_reducer(reduction) + + Sets a reduction class to be used for serialization and deserialization by the module + primitive internals. **reduction** must be a subclass of + :class:`multiprocessing.reduction.AbstractReducer`. + +.. function:: get_reducer() + + Gets the current reduction class in use by the module's primitive internals. + +For example, substituting the reducer class to use the :mod:`pickle` protocol +version 2 to be able to communicate with a Python 2.x programs.:: + + import multiprocessing + from multiprocessing.reduction import AbstractReducer, ForkingPickler + import pickle + + class ForkingPickler2(ForkingPickler): + @classmethod + def dumps(cls, obj, pickle_protocol=2): + return super().dumps(obj, protocol=pickle_protocol) + + loads = pickle.loads + + + def dump(obj, file, protocol=2): + return ForkingPickler2(file, protocol).dump(obj) + + + class Pickle2Reducer(AbstractReducer): + ForkingPickler = ForkingPickler2 + register = ForkingPickler2.register + dump = dump + + multiprocessing.set_reducer(Pickle2Reducer) Synchronization primitives ~~~~~~~~~~~~~~~~~~~~~~~~~~ diff --git a/Lib/multiprocessing/connection.py b/Lib/multiprocessing/connection.py index c9f995e5fa7f57..2aefb87617103f 100644 --- a/Lib/multiprocessing/connection.py +++ b/Lib/multiprocessing/connection.py @@ -23,8 +23,7 @@ from . import util from . import AuthenticationError, BufferTooShort -from .context import reduction -_ForkingPickler = reduction.ForkingPickler +from . import context try: import _winapi @@ -203,7 +202,7 @@ def send(self, obj): """Send a (picklable) object""" self._check_closed() self._check_writable() - self._send_bytes(_ForkingPickler.dumps(obj)) + self._send_bytes(context.reduction.ForkingPickler.dumps(obj)) def recv_bytes(self, maxlength=None): """ @@ -248,7 +247,7 @@ def recv(self): self._check_closed() self._check_readable() buf = self._recv_bytes() - return _ForkingPickler.loads(buf.getbuffer()) + return context.reduction.ForkingPickler.loads(buf.getbuffer()) def poll(self, timeout=0.0): """Whether there is any input available to be read""" @@ -950,23 +949,23 @@ def reduce_connection(conn): def rebuild_connection(ds, readable, writable): sock = ds.detach() return Connection(sock.detach(), readable, writable) - reduction.register(Connection, reduce_connection) + context.reduction.register(Connection, reduce_connection) def reduce_pipe_connection(conn): access = ((_winapi.FILE_GENERIC_READ if conn.readable else 0) | (_winapi.FILE_GENERIC_WRITE if conn.writable else 0)) - dh = reduction.DupHandle(conn.fileno(), access) + dh = context.reduction.DupHandle(conn.fileno(), access) return rebuild_pipe_connection, (dh, conn.readable, conn.writable) def rebuild_pipe_connection(dh, readable, writable): handle = dh.detach() return PipeConnection(handle, readable, writable) - reduction.register(PipeConnection, reduce_pipe_connection) + context.reduction.register(PipeConnection, reduce_pipe_connection) else: def reduce_connection(conn): - df = reduction.DupFd(conn.fileno()) + df = context.reduction.DupFd(conn.fileno()) return rebuild_connection, (df, conn.readable, conn.writable) def rebuild_connection(df, readable, writable): fd = df.detach() return Connection(fd, readable, writable) - reduction.register(Connection, reduce_connection) + context.reduction.register(Connection, reduce_connection) diff --git a/Lib/multiprocessing/context.py b/Lib/multiprocessing/context.py index 5f8e0f0cd46585..4e66b056a5d2ff 100644 --- a/Lib/multiprocessing/context.py +++ b/Lib/multiprocessing/context.py @@ -200,14 +200,12 @@ def get_start_method(self, allow_none=False): def set_start_method(self, method, force=False): raise ValueError('cannot set start method of concrete context') - @property - def reducer(self): + def get_reducer(self): '''Controls how objects will be reduced to a form that can be shared with other processes.''' return globals().get('reduction') - @reducer.setter - def reducer(self, reduction): + def set_reducer(self, reduction): globals()['reduction'] = reduction def _check_available(self): diff --git a/Lib/multiprocessing/forkserver.py b/Lib/multiprocessing/forkserver.py index 87ebef6588acd3..a644f072dc2981 100644 --- a/Lib/multiprocessing/forkserver.py +++ b/Lib/multiprocessing/forkserver.py @@ -10,8 +10,9 @@ from . import connection from . import process -from .context import reduction from . import resource_tracker +from . import context +from .context import reduction from . import spawn from . import util @@ -91,7 +92,7 @@ def connect_to_new_process(self, fds): resource_tracker.getfd()] allfds += fds try: - reduction.sendfds(client, allfds) + context.reduction.sendfds(client, allfds) return parent_r, parent_w except: os.close(parent_r) @@ -259,7 +260,7 @@ def sigchld_handler(*_unused): # Incoming fork request with listener.accept()[0] as s: # Receive fds from client - fds = reduction.recvfds(s, MAXFDS_TO_SEND + 1) + fds = context.reduction.recvfds(s, MAXFDS_TO_SEND + 1) if len(fds) > MAXFDS_TO_SEND: raise RuntimeError( "Too many ({0:n}) fds to send".format( diff --git a/Lib/multiprocessing/heap.py b/Lib/multiprocessing/heap.py index 6217dfe12689b3..5e05cad4fed2d1 100644 --- a/Lib/multiprocessing/heap.py +++ b/Lib/multiprocessing/heap.py @@ -15,7 +15,8 @@ import tempfile import threading -from .context import reduction, assert_spawning +from . import context +from .context import assert_spawning from . import util __all__ = ['BufferWrapper'] @@ -101,12 +102,12 @@ def reduce_arena(a): if a.fd == -1: raise ValueError('Arena is unpicklable because ' 'forking was enabled when it was created') - return rebuild_arena, (a.size, reduction.DupFd(a.fd)) + return rebuild_arena, (a.size, context.reduction.DupFd(a.fd)) def rebuild_arena(size, dupfd): return Arena(size, dupfd.detach()) - reduction.register(Arena, reduce_arena) + context.reduction.register(Arena, reduce_arena) # # Class allowing allocation of chunks of memory from arenas diff --git a/Lib/multiprocessing/managers.py b/Lib/multiprocessing/managers.py index 1f9c2daa25d97f..5f5aa76e82f455 100644 --- a/Lib/multiprocessing/managers.py +++ b/Lib/multiprocessing/managers.py @@ -27,7 +27,8 @@ from traceback import format_exc from . import connection -from .context import reduction, get_spawning_popen, ProcessError +from . import context +from .context import get_spawning_popen, ProcessError from . import pool from . import process from . import util @@ -44,14 +45,14 @@ def reduce_array(a): return array.array, (a.typecode, a.tobytes()) -reduction.register(array.array, reduce_array) +context.reduction.register(array.array, reduce_array) view_types = [type(getattr({}, name)()) for name in ('items','keys','values')] if view_types[0] is not list: # only needed in Py3.0 def rebuild_as_list(obj): return list, (list(obj),) for view_type in view_types: - reduction.register(view_type, rebuild_as_list) + context.reduction.register(view_type, rebuild_as_list) # # Type for identifying shared objects diff --git a/Lib/multiprocessing/popen_forkserver.py b/Lib/multiprocessing/popen_forkserver.py index a56eb9bf11080b..93c9a17cbc1db0 100644 --- a/Lib/multiprocessing/popen_forkserver.py +++ b/Lib/multiprocessing/popen_forkserver.py @@ -1,8 +1,9 @@ import io import os -from .context import reduction, set_spawning_popen -if not reduction.HAVE_SEND_HANDLE: +from . import context +from .context import set_spawning_popen +if not context.reduction.HAVE_SEND_HANDLE: raise ImportError('No support for sending fds between processes') from . import forkserver from . import popen_fork @@ -43,8 +44,8 @@ def _launch(self, process_obj): buf = io.BytesIO() set_spawning_popen(self) try: - reduction.dump(prep_data, buf) - reduction.dump(process_obj, buf) + context.reduction.dump(prep_data, buf) + context.reduction.dump(process_obj, buf) finally: set_spawning_popen(None) diff --git a/Lib/multiprocessing/popen_spawn_posix.py b/Lib/multiprocessing/popen_spawn_posix.py index 24b8634523e5f2..a3572c68c1d501 100644 --- a/Lib/multiprocessing/popen_spawn_posix.py +++ b/Lib/multiprocessing/popen_spawn_posix.py @@ -1,7 +1,8 @@ import io import os -from .context import reduction, set_spawning_popen +from . import context +from .context import set_spawning_popen from . import popen_fork from . import spawn from . import util @@ -43,8 +44,8 @@ def _launch(self, process_obj): fp = io.BytesIO() set_spawning_popen(self) try: - reduction.dump(prep_data, fp) - reduction.dump(process_obj, fp) + context.reduction.dump(prep_data, fp) + context.reduction.dump(process_obj, fp) finally: set_spawning_popen(None) diff --git a/Lib/multiprocessing/popen_spawn_win32.py b/Lib/multiprocessing/popen_spawn_win32.py index 9c4098d0fa4f1e..4086f294870115 100644 --- a/Lib/multiprocessing/popen_spawn_win32.py +++ b/Lib/multiprocessing/popen_spawn_win32.py @@ -4,7 +4,8 @@ import sys import _winapi -from .context import reduction, get_spawning_popen, set_spawning_popen +from . import context +from .context import get_spawning_popen, set_spawning_popen from . import spawn from . import util @@ -89,14 +90,14 @@ def __init__(self, process_obj): # send information to child set_spawning_popen(self) try: - reduction.dump(prep_data, to_child) - reduction.dump(process_obj, to_child) + context.reduction.dump(prep_data, to_child) + context.reduction.dump(process_obj, to_child) finally: set_spawning_popen(None) def duplicate_for_child(self, handle): assert self is get_spawning_popen() - return reduction.duplicate(handle, self.sentinel) + return context.reduction.duplicate(handle, self.sentinel) def wait(self, timeout=None): if self.returncode is None: diff --git a/Lib/multiprocessing/queues.py b/Lib/multiprocessing/queues.py index d112db2cd981d1..46b41a99bcd896 100644 --- a/Lib/multiprocessing/queues.py +++ b/Lib/multiprocessing/queues.py @@ -23,7 +23,6 @@ from . import connection from . import context -_ForkingPickler = context.reduction.ForkingPickler from .util import debug, info, Finalize, register_after_fork, is_exiting @@ -113,7 +112,7 @@ def get(self, block=True, timeout=None): finally: self._rlock.release() # unserialize the data after having released the lock - return _ForkingPickler.loads(res) + return context.reduction.ForkingPickler.loads(res) def qsize(self): # Raises NotImplementedError on Mac OSX because of broken sem_getvalue() @@ -236,7 +235,7 @@ def _feed(buffer, notempty, send_bytes, writelock, close, ignore_epipe, return # serialize the data before acquiring the lock - obj = _ForkingPickler.dumps(obj) + obj = context.reduction.ForkingPickler.dumps(obj) if wacquire is None: send_bytes(obj) else: @@ -355,11 +354,11 @@ def get(self): with self._rlock: res = self._reader.recv_bytes() # unserialize the data after having released the lock - return _ForkingPickler.loads(res) + return context.reduction.ForkingPickler.loads(res) def put(self, obj): # serialize the data before acquiring the lock - obj = _ForkingPickler.dumps(obj) + obj = context.reduction.ForkingPickler.dumps(obj) if self._wlock is None: # writes to a message oriented win32 pipe are atomic self._writer.send_bytes(obj) diff --git a/Lib/multiprocessing/resource_sharer.py b/Lib/multiprocessing/resource_sharer.py index 8d5c9900f69fed..45f20ed0d9e5d4 100644 --- a/Lib/multiprocessing/resource_sharer.py +++ b/Lib/multiprocessing/resource_sharer.py @@ -15,7 +15,7 @@ import threading from . import process -from .context import reduction +from . import context from . import util __all__ = ['stop'] @@ -47,7 +47,7 @@ class DupFd(object): def __init__(self, fd): new_fd = os.dup(fd) def send(conn, pid): - reduction.send_handle(conn, new_fd, pid) + context.reduction.send_handle(conn, new_fd, pid) def close(): os.close(new_fd) self._id = _resource_sharer.register(send, close) @@ -55,7 +55,7 @@ def close(): def detach(self): '''Get the fd. This should only be called once.''' with _resource_sharer.get_connection(self._id) as conn: - return reduction.recv_handle(conn) + return context.reduction.recv_handle(conn) class _ResourceSharer(object): diff --git a/Lib/multiprocessing/spawn.py b/Lib/multiprocessing/spawn.py index 7cc129e2610761..81ab3f49e61100 100644 --- a/Lib/multiprocessing/spawn.py +++ b/Lib/multiprocessing/spawn.py @@ -15,7 +15,7 @@ from . import get_start_method, set_start_method from . import process -from .context import reduction +from . import context from . import util __all__ = ['_main', 'freeze_support', 'set_executable', 'get_executable', @@ -104,8 +104,12 @@ def spawn_main(pipe_handle, parent_pid=None, tracker_fd=None): False, parent_pid) else: source_process = None - new_handle = reduction.duplicate(pipe_handle, - source_process=source_process) + try: + new_handle = context.reduction.duplicate(pipe_handle, + source_process=source_process) + finally: + if source_process is not None: + _winapi.CloseHandle(source_process) fd = msvcrt.open_osfhandle(new_handle, os.O_RDONLY) parent_sentinel = source_process else: @@ -121,9 +125,9 @@ def _main(fd, parent_sentinel): with os.fdopen(fd, 'rb', closefd=True) as from_parent: process.current_process()._inheriting = True try: - preparation_data = reduction.pickle.load(from_parent) + preparation_data = context.reduction.pickle.load(from_parent) prepare(preparation_data) - self = reduction.pickle.load(from_parent) + self = context.reduction.pickle.load(from_parent) finally: del process.current_process()._inheriting return self._bootstrap(parent_sentinel) diff --git a/Lib/test/_test_multiprocessing.py b/Lib/test/_test_multiprocessing.py index 4e48cd45e14ca7..55dc7144bf9be0 100644 --- a/Lib/test/_test_multiprocessing.py +++ b/Lib/test/_test_multiprocessing.py @@ -34,6 +34,7 @@ test.support.import_module('multiprocessing.synchronize') import threading +import multiprocessing import multiprocessing.connection import multiprocessing.dummy import multiprocessing.heap @@ -5473,6 +5474,61 @@ def test__all__(self): # Just make sure names in blacklist are excluded support.check__all__(self, multiprocessing, extra=multiprocessing.__all__, blacklist=['SUBDEBUG', 'SUBWARNING']) + + +class _TestCustomReducer(BaseTestCase): + + ALLOWED_TYPES = ('processes',) + + def setUp(self): + self.original_reducer = multiprocessing.get_reducer() + self.spy = unittest.mock.MagicMock(spec_set=self.original_reducer, + wraps=self.original_reducer) + multiprocessing.set_reducer(self.spy) + + + def tearDown(self): + multiprocessing.set_reducer(self.original_reducer) + + @unittest.skipUnless(HAS_REDUCTION, "test needs multiprocessing.reduction") + def test_connection_custom_reducer(self): + with self.connection.Listener() as l: + with self.connection.Client(l.address) as c: + with l.accept() as d: + c.send(1729) + self.assertEqual(d.recv(), 1729) + + if self.TYPE == 'processes': + self.assertRaises(OSError, l.accept) + + self.assertEqual(self.spy.ForkingPickler.dumps.call_count, 1) + self.assertEqual(self.spy.ForkingPickler.loads.call_count, 1) + + @classmethod + def _put_and_get_in_queue(cls, queue, parent_can_continue): + parent_can_continue.set() + queue.put("Something") + queue.get() + + @unittest.skipUnless(HAS_REDUCTION, "test needs multiprocessing.reduction") + def test_queue_custom_reducer(self): + queue = self.Queue() + parent_can_continue = self.Event() + p = self.Process(target=self._put_and_get_in_queue, + args=(queue, parent_can_continue)) + p.daemon = True + p.start() + parent_can_continue.wait() + element = queue.get(timeout=TIMEOUT3) + self.assertEqual(element, "Something") + queue.put("Other_Something") + p.join(timeout=TIMEOUT3) + self.assertEqual(p.exitcode, 0) + self.assertEqual(element, "Something") + self.assertEqual(self.spy.ForkingPickler.dumps.call_count, 1) + self.assertEqual(self.spy.ForkingPickler.loads.call_count, 1) + close_queue(queue) + # # Mixins # From 1fcf62e899d671967a7dd7d5022ef438da67d0b1 Mon Sep 17 00:00:00 2001 From: Pablo Galindo Date: Sun, 9 Dec 2018 19:08:45 +0000 Subject: [PATCH 02/32] Simplify the interface and use convenience functions --- Doc/library/multiprocessing.rst | 32 +++------ Lib/multiprocessing/connection.py | 4 +- Lib/multiprocessing/context.py | 14 ++++ Lib/multiprocessing/queues.py | 8 +-- Lib/multiprocessing/reduction.py | 68 ++++++++++++++----- Lib/test/_test_multiprocessing.py | 42 ++++++++++-- .../2018-12-09-19-20-12.bpo-28053.JiOeSY.rst | 1 + 7 files changed, 116 insertions(+), 53 deletions(-) create mode 100644 Misc/NEWS.d/next/Library/2018-12-09-19-20-12.bpo-28053.JiOeSY.rst diff --git a/Doc/library/multiprocessing.rst b/Doc/library/multiprocessing.rst index e797b65162c9c9..6664688d23abfa 100644 --- a/Doc/library/multiprocessing.rst +++ b/Doc/library/multiprocessing.rst @@ -1215,17 +1215,10 @@ the reduction mechanism: Abstract base class for use in implementing a Reduction class suitable for use in replacing the standard reduction mechanism used in multiprocessing. - .. class:: ForkingPickler + .. function:: get_pickler_class(): - A subclass of :class:`pickle.Pickler` to be used by multiprocessing. - - .. function:: dump(obj, file, protocol=None) - - Replacement for pickle.dump() to be used by multiprocessing. - - .. function:: register(cls, type, reduce) - - Register a reduce function for a type to be used by multiprocessing. + This method must return a subclass of :class:`pickle.Pickler` to be used by + the multiprocessing reducer mechanism. .. currentmodule:: multiprocessing @@ -1244,26 +1237,17 @@ version 2 to be able to communicate with a Python 2.x programs.:: import multiprocessing from multiprocessing.reduction import AbstractReducer, ForkingPickler - import pickle - class ForkingPickler2(ForkingPickler): + class ForkingPicklerProtocol2(ForkingPickler): @classmethod def dumps(cls, obj, pickle_protocol=2): return super().dumps(obj, protocol=pickle_protocol) - loads = pickle.loads - - - def dump(obj, file, protocol=2): - return ForkingPickler2(file, protocol).dump(obj) - - - class Pickle2Reducer(AbstractReducer): - ForkingPickler = ForkingPickler2 - register = ForkingPickler2.register - dump = dump + class PickleProtocol2Reducer(AbstractReducer): + def get_pickler_class(self): + return ForkingPicklerProtocol2 - multiprocessing.set_reducer(Pickle2Reducer) + multiprocessing.set_reducer(PickleProtocol2Reducer) Synchronization primitives ~~~~~~~~~~~~~~~~~~~~~~~~~~ diff --git a/Lib/multiprocessing/connection.py b/Lib/multiprocessing/connection.py index 2aefb87617103f..e8fc0cf8e550a1 100644 --- a/Lib/multiprocessing/connection.py +++ b/Lib/multiprocessing/connection.py @@ -202,7 +202,7 @@ def send(self, obj): """Send a (picklable) object""" self._check_closed() self._check_writable() - self._send_bytes(context.reduction.ForkingPickler.dumps(obj)) + self._send_bytes(context.reduction.dumps(obj)) def recv_bytes(self, maxlength=None): """ @@ -247,7 +247,7 @@ def recv(self): self._check_closed() self._check_readable() buf = self._recv_bytes() - return context.reduction.ForkingPickler.loads(buf.getbuffer()) + return context.reduction.loads(buf.getbuffer()) def poll(self, timeout=0.0): """Whether there is any input available to be read""" diff --git a/Lib/multiprocessing/context.py b/Lib/multiprocessing/context.py index 4e66b056a5d2ff..c3986f1e50161f 100644 --- a/Lib/multiprocessing/context.py +++ b/Lib/multiprocessing/context.py @@ -1,10 +1,13 @@ import os +import pickle import sys import threading from . import process from . import reduction +original_reducer = reduction + __all__ = () # @@ -206,6 +209,17 @@ def get_reducer(self): return globals().get('reduction') def set_reducer(self, reduction): + if reduction is original_reducer: + globals()['reduction'] = reduction + return + + if not isinstance(reduction, original_reducer.AbstractReducer): + raise TypeError("Custom reducers must subclass " + "multiprocessing.reduction.AbstractReducer") + if not isinstance(reduction.get_pickler_class(), pickle.Pickler): + raise TypeError("Custom reducers must return a subclass of " + "pickler.Pickler in the get_pickler_class() method") + globals()['reduction'] = reduction def _check_available(self): diff --git a/Lib/multiprocessing/queues.py b/Lib/multiprocessing/queues.py index 46b41a99bcd896..51ebc6772395b0 100644 --- a/Lib/multiprocessing/queues.py +++ b/Lib/multiprocessing/queues.py @@ -112,7 +112,7 @@ def get(self, block=True, timeout=None): finally: self._rlock.release() # unserialize the data after having released the lock - return context.reduction.ForkingPickler.loads(res) + return context.reduction.loads(res) def qsize(self): # Raises NotImplementedError on Mac OSX because of broken sem_getvalue() @@ -235,7 +235,7 @@ def _feed(buffer, notempty, send_bytes, writelock, close, ignore_epipe, return # serialize the data before acquiring the lock - obj = context.reduction.ForkingPickler.dumps(obj) + obj = context.reduction.dumps(obj) if wacquire is None: send_bytes(obj) else: @@ -354,11 +354,11 @@ def get(self): with self._rlock: res = self._reader.recv_bytes() # unserialize the data after having released the lock - return context.reduction.ForkingPickler.loads(res) + return context.reduction.loads(res) def put(self, obj): # serialize the data before acquiring the lock - obj = context.reduction.ForkingPickler.dumps(obj) + obj = context.reduction.dumps(obj) if self._wlock is None: # writes to a message oriented win32 pipe are atomic self._writer.send_bytes(obj) diff --git a/Lib/multiprocessing/reduction.py b/Lib/multiprocessing/reduction.py index 5593f0682f7fce..91512706f5b731 100644 --- a/Lib/multiprocessing/reduction.py +++ b/Lib/multiprocessing/reduction.py @@ -7,7 +7,7 @@ # Licensed to PSF under a Contributor Agreement. # -from abc import ABCMeta +import abc import copyreg import functools import io @@ -53,12 +53,26 @@ def dumps(cls, obj, protocol=None): loads = pickle.loads -register = ForkingPickler.register + + +def register(type_, reduce_): + return ForkingPickler.register(type_, reduce_) + def dump(obj, file, protocol=None): '''Replacement for pickle.dump() using ForkingPickler.''' ForkingPickler(file, protocol).dump(obj) + +def dumps(obj, protocol=None): + return ForkingPickler.dumps(obj, protocol=protocol) + + +def loads(s, *, fix_imports=True, encoding="ASCII", errors="strict"): + return ForkingPickler.loads(s, fix_imports=fix_imports, + encoding=encoding, errors=errors) + + # # Platform specific definitions # @@ -248,30 +262,27 @@ def _rebuild_socket(df, family, type, proto): register(socket.socket, _reduce_socket) -class AbstractReducer(metaclass=ABCMeta): +class AbstractReducer(metaclass=abc.ABCMeta): '''Abstract base class for use in implementing a Reduction class suitable for use in replacing the standard reduction mechanism used in multiprocessing.''' - ForkingPickler = ForkingPickler - register = register - dump = dump send_handle = send_handle recv_handle = recv_handle if sys.platform == 'win32': - steal_handle = steal_handle - duplicate = duplicate + steal_handle = staticmethod(steal_handle) + duplicate = staticmethod(duplicate) DupHandle = DupHandle else: - sendfds = sendfds - recvfds = recvfds - DupFd = DupFd + sendfds = staticmethod(sendfds) + recvfds = staticmethod(recvfds) + DupFd = staticmethod(DupFd) - _reduce_method = _reduce_method - _reduce_method_descriptor = _reduce_method_descriptor - _rebuild_partial = _rebuild_partial - _reduce_socket = _reduce_socket - _rebuild_socket = _rebuild_socket + _reduce_method = staticmethod(_reduce_method) + _reduce_method_descriptor = staticmethod(_reduce_method_descriptor) + _rebuild_partial = staticmethod(_rebuild_partial) + _reduce_socket = staticmethod(_reduce_socket) + _rebuild_socket = staticmethod(_rebuild_socket) def __init__(self, *args): register(type(_C().f), _reduce_method) @@ -279,3 +290,28 @@ def __init__(self, *args): register(type(int.__add__), _reduce_method_descriptor) register(functools.partial, _reduce_partial) register(socket.socket, _reduce_socket) + + @property + def HAVE_SEND_HANDLE(self): + import sys + return (sys.platform == 'win32' or + (hasattr(socket, 'CMSG_LEN') and + hasattr(socket, 'SCM_RIGHTS') and + hasattr(socket.socket, 'sendmsg'))) + + @abc.abstractmethod + def get_pickler_class(self): + pass + + def register(self, type_, reduce_): + return self.get_pickler_class().register(type_, reduce_) + + def dump(self, obj, file, protocol=None): + self.get_pickler_class()(file, protocol).dump(obj) + + def dumps(self, obj, protocol=None): + return self.get_pickler_class().dumps(obj, protocol=protocol) + + def loads(self, s, *, fix_imports=True, encoding="ASCII", errors="strict"): + return self.get_pickler_class().loads(s, fix_imports=fix_imports, + encoding=encoding, errors=errors) diff --git a/Lib/test/_test_multiprocessing.py b/Lib/test/_test_multiprocessing.py index 55dc7144bf9be0..00d75e1730a568 100644 --- a/Lib/test/_test_multiprocessing.py +++ b/Lib/test/_test_multiprocessing.py @@ -5482,10 +5482,15 @@ class _TestCustomReducer(BaseTestCase): def setUp(self): self.original_reducer = multiprocessing.get_reducer() - self.spy = unittest.mock.MagicMock(spec_set=self.original_reducer, - wraps=self.original_reducer) - multiprocessing.set_reducer(self.spy) + spy = unittest.mock.MagicMock(spec_set=self.original_reducer.ForkingPickler, + wraps=self.original_reducer.ForkingPickler) + self.spy = spy + class SpyReducer(reduction.AbstractReducer): + def get_pickler_class(self): + return spy + + multiprocessing.set_reducer(SpyReducer()) def tearDown(self): multiprocessing.set_reducer(self.original_reducer) @@ -5501,8 +5506,8 @@ def test_connection_custom_reducer(self): if self.TYPE == 'processes': self.assertRaises(OSError, l.accept) - self.assertEqual(self.spy.ForkingPickler.dumps.call_count, 1) - self.assertEqual(self.spy.ForkingPickler.loads.call_count, 1) + self.assertEqual(self.spy.dumps.call_count, 1) + self.assertEqual(self.spy.loads.call_count, 1) @classmethod def _put_and_get_in_queue(cls, queue, parent_can_continue): @@ -5525,10 +5530,33 @@ def test_queue_custom_reducer(self): p.join(timeout=TIMEOUT3) self.assertEqual(p.exitcode, 0) self.assertEqual(element, "Something") - self.assertEqual(self.spy.ForkingPickler.dumps.call_count, 1) - self.assertEqual(self.spy.ForkingPickler.loads.call_count, 1) + self.assertEqual(self.spy.dumps.call_count, 1) + self.assertEqual(self.spy.loads.call_count, 1) close_queue(queue) + +class _TestCustomReducerInvalidParameters(BaseTestCase): + def test_setting_invalid_reducer(self): + class BadReducer(): + def get_pickler_class(self): + return + + with self.assertRaises(TypeError): + multiprocessing.set_reducer(BadReducer()) + + + def test_setting_valid_reducer_that_returns_ivalid_pickler(self): + + class PicklerClass: + pass + + class BadReducer(reduction.AbstractReducer): + def get_pickler_class(self): + return PicklerClass + + with self.assertRaises(TypeError): + multiprocessing.set_reducer(BadReducer()) + # # Mixins # diff --git a/Misc/NEWS.d/next/Library/2018-12-09-19-20-12.bpo-28053.JiOeSY.rst b/Misc/NEWS.d/next/Library/2018-12-09-19-20-12.bpo-28053.JiOeSY.rst new file mode 100644 index 00000000000000..c672a205d135cf --- /dev/null +++ b/Misc/NEWS.d/next/Library/2018-12-09-19-20-12.bpo-28053.JiOeSY.rst @@ -0,0 +1 @@ +Fix the implementation of custom reducers for in `multiprocessing`. From 4dd92e0351f85ba6972f0a7709e3b7973b819e69 Mon Sep 17 00:00:00 2001 From: Pablo Galindo Date: Sat, 16 Mar 2019 20:18:28 +0000 Subject: [PATCH 03/32] Fix some typos and add new test for custom reducers --- Doc/library/multiprocessing.rst | 27 ++++++++-------- Lib/multiprocessing/context.py | 8 ++--- Lib/multiprocessing/reduction.py | 10 +++--- Lib/test/_test_multiprocessing.py | 52 ++++++++++++++++++++++++++++++- 4 files changed, 74 insertions(+), 23 deletions(-) diff --git a/Doc/library/multiprocessing.rst b/Doc/library/multiprocessing.rst index 6664688d23abfa..474078f11192ce 100644 --- a/Doc/library/multiprocessing.rst +++ b/Doc/library/multiprocessing.rst @@ -1201,34 +1201,35 @@ Custom Reduction Several primitives of the :mod:`multiprocessing` module such as :class:`multiprocessing.Queue`, :class:`multiprocessing.connection.Listener` or :class:`multiprocessing.connection.Server` need to serialize and deserialize Python -objects to communicate between processes. Sometimes is useful to control what -serialization is to be used for the transport of data for supporting communication with -different versions of Python, use more performant 3rd party tools or custom strategies. +objects to communicate between processes. Sometimes it is useful to control what +serialization is to be used for the transport of data in order to support communication +with different versions of Python, use more performant third party tools or custom +strategies. -For this pourpose a set of hooks is available to provide alternate implementations of +For this purpose a set of hooks is available to provide alternate implementations of the reduction mechanism: .. currentmodule:: multiprocessing.reduction .. class:: AbstractReducer() - Abstract base class for use in implementing a Reduction class suitable for use - in replacing the standard reduction mechanism used in multiprocessing. + Abstract base class that can be implemented in order to replace the standard + reduction mechanism used in multiprocessing - .. function:: get_pickler_class(): + .. function:: get_pickler(): - This method must return a subclass of :class:`pickle.Pickler` to be used by - the multiprocessing reducer mechanism. + This method must return a instance of a subclass of :class:`pickle.Pickler` + to be used by the multiprocessing reducer mechanism. .. currentmodule:: multiprocessing -.. function:: set_reducer(reduction) +.. method:: set_reducer(reduction) Sets a reduction class to be used for serialization and deserialization by the module - primitive internals. **reduction** must be a subclass of + primitive internals. **reduction** must be an instance of a subclass of :class:`multiprocessing.reduction.AbstractReducer`. -.. function:: get_reducer() +.. method:: get_reducer() Gets the current reduction class in use by the module's primitive internals. @@ -1244,7 +1245,7 @@ version 2 to be able to communicate with a Python 2.x programs.:: return super().dumps(obj, protocol=pickle_protocol) class PickleProtocol2Reducer(AbstractReducer): - def get_pickler_class(self): + def get_pickler(self): return ForkingPicklerProtocol2 multiprocessing.set_reducer(PickleProtocol2Reducer) diff --git a/Lib/multiprocessing/context.py b/Lib/multiprocessing/context.py index c3986f1e50161f..45e6026736313d 100644 --- a/Lib/multiprocessing/context.py +++ b/Lib/multiprocessing/context.py @@ -214,11 +214,11 @@ def set_reducer(self, reduction): return if not isinstance(reduction, original_reducer.AbstractReducer): - raise TypeError("Custom reducers must subclass " - "multiprocessing.reduction.AbstractReducer") - if not isinstance(reduction.get_pickler_class(), pickle.Pickler): + raise TypeError("Custom reducers must be instances of a " + "subclass of multiprocessing.reduction.AbstractReducer") + if not isinstance(reduction.get_pickler(), pickle.Pickler): raise TypeError("Custom reducers must return a subclass of " - "pickler.Pickler in the get_pickler_class() method") + "pickler.Pickler in the get_pickler() method") globals()['reduction'] = reduction diff --git a/Lib/multiprocessing/reduction.py b/Lib/multiprocessing/reduction.py index 91512706f5b731..c73b0c41691e32 100644 --- a/Lib/multiprocessing/reduction.py +++ b/Lib/multiprocessing/reduction.py @@ -300,18 +300,18 @@ def HAVE_SEND_HANDLE(self): hasattr(socket.socket, 'sendmsg'))) @abc.abstractmethod - def get_pickler_class(self): + def get_pickler(self): pass def register(self, type_, reduce_): - return self.get_pickler_class().register(type_, reduce_) + return self.get_pickler().register(type_, reduce_) def dump(self, obj, file, protocol=None): - self.get_pickler_class()(file, protocol).dump(obj) + self.get_pickler()(file, protocol).dump(obj) def dumps(self, obj, protocol=None): - return self.get_pickler_class().dumps(obj, protocol=protocol) + return self.get_pickler().dumps(obj, protocol=protocol) def loads(self, s, *, fix_imports=True, encoding="ASCII", errors="strict"): - return self.get_pickler_class().loads(s, fix_imports=fix_imports, + return self.get_pickler().loads(s, fix_imports=fix_imports, encoding=encoding, errors=errors) diff --git a/Lib/test/_test_multiprocessing.py b/Lib/test/_test_multiprocessing.py index 00d75e1730a568..d140f438132de1 100644 --- a/Lib/test/_test_multiprocessing.py +++ b/Lib/test/_test_multiprocessing.py @@ -5487,7 +5487,7 @@ def setUp(self): self.spy = spy class SpyReducer(reduction.AbstractReducer): - def get_pickler_class(self): + def get_pickler(self): return spy multiprocessing.set_reducer(SpyReducer()) @@ -5535,6 +5535,56 @@ def test_queue_custom_reducer(self): close_queue(queue) +class _TestCustomReducerWithContext(BaseTestCase): + + ALLOWED_TYPES = ('processes',) + + def setUp(self): + class CustomContext(multiprocessing.context.BaseContext): + _name = "custom" + Process = multiprocessing.get_context().Process + + self.ctx = CustomContext() + self.original_reducer = self.ctx.get_reducer() + spy = unittest.mock.MagicMock(spec_set=self.original_reducer.ForkingPickler, + wraps=self.original_reducer.ForkingPickler) + self.spy = spy + + class SpyReducer(reduction.AbstractReducer): + def get_pickler(self): + return spy + + self.ctx.set_reducer(SpyReducer()) + + def tearDown(self): + self.ctx.set_reducer(self.original_reducer) + + @classmethod + def _put_and_get_in_queue(cls, queue, parent_can_continue): + parent_can_continue.set() + queue.put("Something") + queue.get() + + @unittest.skipUnless(HAS_REDUCTION, "test needs multiprocessing.reduction") + def test_queue_custom_reducer_with_context(self): + queue = self.ctx.Queue() + parent_can_continue = self.ctx.Event() + p = self.ctx.Process(target=self._put_and_get_in_queue, + args=(queue, parent_can_continue)) + p.daemon = True + p.start() + parent_can_continue.wait() + element = queue.get(timeout=TIMEOUT3) + self.assertEqual(element, "Something") + queue.put("Other_Something") + p.join(timeout=TIMEOUT3) + self.assertEqual(p.exitcode, 0) + self.assertEqual(element, "Something") + self.assertEqual(self.spy.dumps.call_count, 1) + self.assertEqual(self.spy.loads.call_count, 1) + close_queue(queue) + + class _TestCustomReducerInvalidParameters(BaseTestCase): def test_setting_invalid_reducer(self): class BadReducer(): From 531239078b64bb3307a0cdbfb530b98b779f6c40 Mon Sep 17 00:00:00 2001 From: Pablo Galindo Date: Sun, 17 Mar 2019 02:14:57 +0000 Subject: [PATCH 04/32] Implement per-context reducers --- Doc/library/multiprocessing.rst | 8 +++- Lib/multiprocessing/connection.py | 8 ++-- Lib/multiprocessing/context.py | 10 +++-- Lib/multiprocessing/queues.py | 24 ++++++----- Lib/multiprocessing/reduction.py | 1 - Lib/test/_test_multiprocessing.py | 71 ++++++++++++++++++------------- 6 files changed, 73 insertions(+), 49 deletions(-) diff --git a/Doc/library/multiprocessing.rst b/Doc/library/multiprocessing.rst index 474078f11192ce..725ee2df6bb018 100644 --- a/Doc/library/multiprocessing.rst +++ b/Doc/library/multiprocessing.rst @@ -1218,7 +1218,7 @@ the reduction mechanism: .. function:: get_pickler(): - This method must return a instance of a subclass of :class:`pickle.Pickler` + This method must return an instance of a subclass of :class:`pickler.Pickler` to be used by the multiprocessing reducer mechanism. .. currentmodule:: multiprocessing @@ -1245,11 +1245,15 @@ version 2 to be able to communicate with a Python 2.x programs.:: return super().dumps(obj, protocol=pickle_protocol) class PickleProtocol2Reducer(AbstractReducer): - def get_pickler(self): + def get_pickler_class(self): return ForkingPicklerProtocol2 multiprocessing.set_reducer(PickleProtocol2Reducer) +Notice that using :meth:`multiprocessing.set_reducer` changes the reducer globally. If +changing this setting globally is undesirable you could call :meth:`context.set_reducer`, +where *context* is a context object obtained by calling :func:`get_context`. + Synchronization primitives ~~~~~~~~~~~~~~~~~~~~~~~~~~ diff --git a/Lib/multiprocessing/connection.py b/Lib/multiprocessing/connection.py index e8fc0cf8e550a1..3b59f18157d239 100644 --- a/Lib/multiprocessing/connection.py +++ b/Lib/multiprocessing/connection.py @@ -24,6 +24,7 @@ from . import AuthenticationError, BufferTooShort from . import context +from . import get_context try: import _winapi @@ -113,7 +114,8 @@ def address_type(address): class _ConnectionBase: _handle = None - def __init__(self, handle, readable=True, writable=True): + def __init__(self, handle, readable=True, writable=True, ctx=None): + self._ctx = ctx or get_context() handle = handle.__index__() if handle < 0: raise ValueError("invalid handle") @@ -202,7 +204,7 @@ def send(self, obj): """Send a (picklable) object""" self._check_closed() self._check_writable() - self._send_bytes(context.reduction.dumps(obj)) + self._send_bytes(self._ctx.get_reducer().dumps(obj)) def recv_bytes(self, maxlength=None): """ @@ -247,7 +249,7 @@ def recv(self): self._check_closed() self._check_readable() buf = self._recv_bytes() - return context.reduction.loads(buf.getbuffer()) + return self._ctx.get_reducer().loads(buf.getbuffer()) def poll(self, timeout=0.0): """Whether there is any input available to be read""" diff --git a/Lib/multiprocessing/context.py b/Lib/multiprocessing/context.py index 45e6026736313d..7a6c73f47503f1 100644 --- a/Lib/multiprocessing/context.py +++ b/Lib/multiprocessing/context.py @@ -206,21 +206,25 @@ def set_start_method(self, method, force=False): def get_reducer(self): '''Controls how objects will be reduced to a form that can be shared with other processes.''' - return globals().get('reduction') + reducer = getattr(self, "_reducer", None) + if reducer is None: + return reduction + return reducer def set_reducer(self, reduction): if reduction is original_reducer: - globals()['reduction'] = reduction + self._reducer = None return if not isinstance(reduction, original_reducer.AbstractReducer): raise TypeError("Custom reducers must be instances of a " "subclass of multiprocessing.reduction.AbstractReducer") if not isinstance(reduction.get_pickler(), pickle.Pickler): + breakpoint() raise TypeError("Custom reducers must return a subclass of " "pickler.Pickler in the get_pickler() method") - globals()['reduction'] = reduction + self._reducer = reduction def _check_available(self): pass diff --git a/Lib/multiprocessing/queues.py b/Lib/multiprocessing/queues.py index 51ebc6772395b0..91a213ec0faf97 100644 --- a/Lib/multiprocessing/queues.py +++ b/Lib/multiprocessing/queues.py @@ -48,6 +48,7 @@ def __init__(self, maxsize=0, *, ctx): # For use by concurrent.futures self._ignore_epipe = False + self._ctx = ctx self._after_fork() if sys.platform != 'win32': @@ -56,11 +57,11 @@ def __init__(self, maxsize=0, *, ctx): def __getstate__(self): context.assert_spawning(self) return (self._ignore_epipe, self._maxsize, self._reader, self._writer, - self._rlock, self._wlock, self._sem, self._opid) + self._rlock, self._wlock, self._sem, self._opid, self._ctx) def __setstate__(self, state): (self._ignore_epipe, self._maxsize, self._reader, self._writer, - self._rlock, self._wlock, self._sem, self._opid) = state + self._rlock, self._wlock, self._sem, self._opid, self._ctx) = state self._after_fork() def _after_fork(self): @@ -112,7 +113,7 @@ def get(self, block=True, timeout=None): finally: self._rlock.release() # unserialize the data after having released the lock - return context.reduction.loads(res) + return self._ctx.get_reducer().loads(res) def qsize(self): # Raises NotImplementedError on Mac OSX because of broken sem_getvalue() @@ -163,7 +164,7 @@ def _start_thread(self): target=Queue._feed, args=(self._buffer, self._notempty, self._send_bytes, self._wlock, self._writer.close, self._ignore_epipe, - self._on_queue_feeder_error, self._sem), + self._on_queue_feeder_error, self._sem, self._ctx), name='QueueFeederThread' ) self._thread.daemon = True @@ -205,7 +206,7 @@ def _finalize_close(buffer, notempty): @staticmethod def _feed(buffer, notempty, send_bytes, writelock, close, ignore_epipe, - onerror, queue_sem): + onerror, queue_sem, ctx): debug('starting thread to feed data to pipe') nacquire = notempty.acquire nrelease = notempty.release @@ -235,7 +236,7 @@ def _feed(buffer, notempty, send_bytes, writelock, close, ignore_epipe, return # serialize the data before acquiring the lock - obj = context.reduction.dumps(obj) + obj = ctx.get_reducer().dumps(obj) if wacquire is None: send_bytes(obj) else: @@ -331,6 +332,7 @@ def join(self): class SimpleQueue(object): def __init__(self, *, ctx): + self._ctx = ctx self._reader, self._writer = connection.Pipe(duplex=False) self._rlock = ctx.Lock() self._poll = self._reader.poll @@ -344,21 +346,23 @@ def empty(self): def __getstate__(self): context.assert_spawning(self) - return (self._reader, self._writer, self._rlock, self._wlock) + return (self._reader, self._writer, self._rlock, self._wlock, + self._ctx) def __setstate__(self, state): - (self._reader, self._writer, self._rlock, self._wlock) = state + (self._reader, self._writer, self._rlock, self._wlock, + self._ctx) = state self._poll = self._reader.poll def get(self): with self._rlock: res = self._reader.recv_bytes() # unserialize the data after having released the lock - return context.reduction.loads(res) + return self._ctx.get_reducer().loads(res) def put(self, obj): # serialize the data before acquiring the lock - obj = context.reduction.dumps(obj) + obj = self._ctx.get_reducer().dumps(obj) if self._wlock is None: # writes to a message oriented win32 pipe are atomic self._writer.send_bytes(obj) diff --git a/Lib/multiprocessing/reduction.py b/Lib/multiprocessing/reduction.py index c73b0c41691e32..948e524c77bfa6 100644 --- a/Lib/multiprocessing/reduction.py +++ b/Lib/multiprocessing/reduction.py @@ -54,7 +54,6 @@ def dumps(cls, obj, protocol=None): loads = pickle.loads - def register(type_, reduce_): return ForkingPickler.register(type_, reduce_) diff --git a/Lib/test/_test_multiprocessing.py b/Lib/test/_test_multiprocessing.py index d140f438132de1..23080993baca53 100644 --- a/Lib/test/_test_multiprocessing.py +++ b/Lib/test/_test_multiprocessing.py @@ -5476,24 +5476,40 @@ def test__all__(self): blacklist=['SUBDEBUG', 'SUBWARNING']) +class SpyReducer(reduction.AbstractReducer): + """Custom reducer that records the call to the Pickler + class it returns. Mocks are not used because they do not + play good with pickling""" + def __init__(self, spy): + self.spy = spy + def get_pickler(self): + spy = self.spy + class Pickler(pickle.Pickler): + def __init__(self): + pass + @classmethod + def dumps(self, obj, protocol=None): + spy["dumps"] += 1 + return pickle.dumps(obj) + @classmethod + def loads(self, *args, **kwargs): + spy["loads"] += 1 + return pickle.loads(*args, **kwargs) + return Pickler() + class _TestCustomReducer(BaseTestCase): ALLOWED_TYPES = ('processes',) def setUp(self): + self.context = multiprocessing.get_context() self.original_reducer = multiprocessing.get_reducer() - spy = unittest.mock.MagicMock(spec_set=self.original_reducer.ForkingPickler, - wraps=self.original_reducer.ForkingPickler) - self.spy = spy - - class SpyReducer(reduction.AbstractReducer): - def get_pickler(self): - return spy - - multiprocessing.set_reducer(SpyReducer()) + self.spy = {'loads': 0, 'dumps': 0} + self.context.set_reducer(SpyReducer(self.spy)) def tearDown(self): - multiprocessing.set_reducer(self.original_reducer) + self.context.set_reducer(self.original_reducer) + del self.spy @unittest.skipUnless(HAS_REDUCTION, "test needs multiprocessing.reduction") def test_connection_custom_reducer(self): @@ -5506,8 +5522,8 @@ def test_connection_custom_reducer(self): if self.TYPE == 'processes': self.assertRaises(OSError, l.accept) - self.assertEqual(self.spy.dumps.call_count, 1) - self.assertEqual(self.spy.loads.call_count, 1) + self.assertEqual(self.spy['loads'], 1) + self.assertEqual(self.spy['dumps'], 1) @classmethod def _put_and_get_in_queue(cls, queue, parent_can_continue): @@ -5530,31 +5546,26 @@ def test_queue_custom_reducer(self): p.join(timeout=TIMEOUT3) self.assertEqual(p.exitcode, 0) self.assertEqual(element, "Something") - self.assertEqual(self.spy.dumps.call_count, 1) - self.assertEqual(self.spy.loads.call_count, 1) + self.assertEqual(self.spy['loads'], 1) + self.assertEqual(self.spy['dumps'], 1) close_queue(queue) +class CustomContext(multiprocessing.context.BaseContext): + _name = "custom" + Process = multiprocessing.get_context().Process + + class _TestCustomReducerWithContext(BaseTestCase): ALLOWED_TYPES = ('processes',) def setUp(self): - class CustomContext(multiprocessing.context.BaseContext): - _name = "custom" - Process = multiprocessing.get_context().Process - self.ctx = CustomContext() - self.original_reducer = self.ctx.get_reducer() - spy = unittest.mock.MagicMock(spec_set=self.original_reducer.ForkingPickler, - wraps=self.original_reducer.ForkingPickler) - self.spy = spy - - class SpyReducer(reduction.AbstractReducer): - def get_pickler(self): - return spy - self.ctx.set_reducer(SpyReducer()) + self.original_reducer = self.ctx.get_reducer() + self.spy = {'loads': 0, 'dumps': 0} + self.ctx.set_reducer(SpyReducer(self.spy)) def tearDown(self): self.ctx.set_reducer(self.original_reducer) @@ -5580,8 +5591,8 @@ def test_queue_custom_reducer_with_context(self): p.join(timeout=TIMEOUT3) self.assertEqual(p.exitcode, 0) self.assertEqual(element, "Something") - self.assertEqual(self.spy.dumps.call_count, 1) - self.assertEqual(self.spy.loads.call_count, 1) + self.assertEqual(self.spy['loads'], 1) + self.assertEqual(self.spy['dumps'], 1) close_queue(queue) @@ -5595,7 +5606,7 @@ def get_pickler_class(self): multiprocessing.set_reducer(BadReducer()) - def test_setting_valid_reducer_that_returns_ivalid_pickler(self): + def test_setting_valid_reducer_that_returns_invalid_pickler(self): class PicklerClass: pass From ac9a8d3c598f43a0ed37609badae950ea2402c1f Mon Sep 17 00:00:00 2001 From: Pablo Galindo Date: Sat, 23 Mar 2019 01:55:48 +0000 Subject: [PATCH 05/32] Implement tests for custom contexts --- Lib/test/_test_multiprocessing.py | 57 ++++++++++++++++++++++++------- 1 file changed, 45 insertions(+), 12 deletions(-) diff --git a/Lib/test/_test_multiprocessing.py b/Lib/test/_test_multiprocessing.py index 23080993baca53..12e6a887fa0f68 100644 --- a/Lib/test/_test_multiprocessing.py +++ b/Lib/test/_test_multiprocessing.py @@ -5561,14 +5561,16 @@ class _TestCustomReducerWithContext(BaseTestCase): ALLOWED_TYPES = ('processes',) def setUp(self): - self.ctx = CustomContext() - - self.original_reducer = self.ctx.get_reducer() - self.spy = {'loads': 0, 'dumps': 0} - self.ctx.set_reducer(SpyReducer(self.spy)) + self.custom_ctx = CustomContext() + self.default_ctx = multiprocessing.get_context() + self.original_custom_reducer = self.custom_ctx.get_reducer() + self.original_default_reducer = self.default_ctx.get_reducer() + self.default_spy = {'loads': 0, 'dumps': 0} + self.custom_spy = {'loads': 0, 'dumps': 0} def tearDown(self): - self.ctx.set_reducer(self.original_reducer) + self.custom_ctx.set_reducer(self.original_custom_reducer) + self.default_ctx.set_reducer(self.original_default_reducer) @classmethod def _put_and_get_in_queue(cls, queue, parent_can_continue): @@ -5577,10 +5579,13 @@ def _put_and_get_in_queue(cls, queue, parent_can_continue): queue.get() @unittest.skipUnless(HAS_REDUCTION, "test needs multiprocessing.reduction") - def test_queue_custom_reducer_with_context(self): - queue = self.ctx.Queue() - parent_can_continue = self.ctx.Event() - p = self.ctx.Process(target=self._put_and_get_in_queue, + def test_queue_custom_reducer_over_custom_context(self): + self.default_ctx.set_reducer(SpyReducer(self.default_spy)) + self.custom_ctx.set_reducer(SpyReducer(self.custom_spy)) + + queue = self.custom_ctx.Queue() + parent_can_continue = self.custom_ctx.Event() + p = self.custom_ctx.Process(target=self._put_and_get_in_queue, args=(queue, parent_can_continue)) p.daemon = True p.start() @@ -5591,10 +5596,38 @@ def test_queue_custom_reducer_with_context(self): p.join(timeout=TIMEOUT3) self.assertEqual(p.exitcode, 0) self.assertEqual(element, "Something") - self.assertEqual(self.spy['loads'], 1) - self.assertEqual(self.spy['dumps'], 1) + self.assertEqual(self.custom_spy['loads'], 1) + self.assertEqual(self.custom_spy['dumps'], 1) + self.assertEqual(self.default_spy['loads'], 0) + self.assertEqual(self.default_spy['dumps'], 0) close_queue(queue) + @unittest.skipUnless(HAS_REDUCTION, "test needs multiprocessing.reduction") + def test_queue_custom_reducer_over_default_context(self): + self.default_ctx.set_reducer(SpyReducer(self.default_spy)) + self.custom_ctx.set_reducer(SpyReducer(self.custom_spy)) + + queue = self.default_ctx.Queue() + parent_can_continue = self.default_ctx.Event() + p = self.default_ctx.Process(target=self._put_and_get_in_queue, + args=(queue, parent_can_continue)) + p.daemon = True + p.start() + parent_can_continue.wait() + element = queue.get(timeout=TIMEOUT3) + self.assertEqual(element, "Something") + queue.put("Other_Something") + p.join(timeout=TIMEOUT3) + self.assertEqual(p.exitcode, 0) + self.assertEqual(element, "Something") + self.assertEqual(self.custom_spy['loads'], 0) + self.assertEqual(self.custom_spy['dumps'], 0) + self.assertEqual(self.default_spy['loads'], 1) + self.assertEqual(self.default_spy['dumps'], 1) + close_queue(queue) + + + class _TestCustomReducerInvalidParameters(BaseTestCase): def test_setting_invalid_reducer(self): From 95046c20441917a59203c8969ecf898e381e0500 Mon Sep 17 00:00:00 2001 From: Pablo Galindo Date: Sat, 23 Mar 2019 01:56:29 +0000 Subject: [PATCH 06/32] Make reducer default to None at class level --- Lib/multiprocessing/context.py | 6 +++--- 1 file changed, 3 insertions(+), 3 deletions(-) diff --git a/Lib/multiprocessing/context.py b/Lib/multiprocessing/context.py index 7a6c73f47503f1..527d583f8ba76f 100644 --- a/Lib/multiprocessing/context.py +++ b/Lib/multiprocessing/context.py @@ -40,6 +40,7 @@ class BaseContext(object): current_process = staticmethod(process.current_process) parent_process = staticmethod(process.parent_process) active_children = staticmethod(process.active_children) + _reducer = None def cpu_count(self): '''Returns the number of CPUs in the system''' @@ -206,7 +207,7 @@ def set_start_method(self, method, force=False): def get_reducer(self): '''Controls how objects will be reduced to a form that can be shared with other processes.''' - reducer = getattr(self, "_reducer", None) + reducer = self._reducer if reducer is None: return reduction return reducer @@ -220,8 +221,7 @@ def set_reducer(self, reduction): raise TypeError("Custom reducers must be instances of a " "subclass of multiprocessing.reduction.AbstractReducer") if not isinstance(reduction.get_pickler(), pickle.Pickler): - breakpoint() - raise TypeError("Custom reducers must return a subclass of " + raise TypeError("Custom reducers must return an instance of " "pickler.Pickler in the get_pickler() method") self._reducer = reduction From c9bf4f57c0b734b476d3c5061113f296656c5228 Mon Sep 17 00:00:00 2001 From: Pablo Galindo Date: Sat, 23 Mar 2019 01:59:41 +0000 Subject: [PATCH 07/32] Fix doc reference --- Doc/library/multiprocessing.rst | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/Doc/library/multiprocessing.rst b/Doc/library/multiprocessing.rst index 725ee2df6bb018..c2675994e49c12 100644 --- a/Doc/library/multiprocessing.rst +++ b/Doc/library/multiprocessing.rst @@ -1245,7 +1245,7 @@ version 2 to be able to communicate with a Python 2.x programs.:: return super().dumps(obj, protocol=pickle_protocol) class PickleProtocol2Reducer(AbstractReducer): - def get_pickler_class(self): + def get_pickler(self): return ForkingPicklerProtocol2 multiprocessing.set_reducer(PickleProtocol2Reducer) From 94c6a66eb3b23e21bc622aa8ff638287cb063b0b Mon Sep 17 00:00:00 2001 From: Pablo Galindo Date: Sat, 23 Mar 2019 02:04:43 +0000 Subject: [PATCH 08/32] Add docstrings to the test classes --- Lib/test/_test_multiprocessing.py | 2 ++ 1 file changed, 2 insertions(+) diff --git a/Lib/test/_test_multiprocessing.py b/Lib/test/_test_multiprocessing.py index 12e6a887fa0f68..79b98f47123bc1 100644 --- a/Lib/test/_test_multiprocessing.py +++ b/Lib/test/_test_multiprocessing.py @@ -5498,6 +5498,7 @@ def loads(self, *args, **kwargs): return Pickler() class _TestCustomReducer(BaseTestCase): + """Test case for the global reducer""" ALLOWED_TYPES = ('processes',) @@ -5557,6 +5558,7 @@ class CustomContext(multiprocessing.context.BaseContext): class _TestCustomReducerWithContext(BaseTestCase): + """Test case for per-context reducers""" ALLOWED_TYPES = ('processes',) From 9d39a347be385f90222333840c3f37f733172dab Mon Sep 17 00:00:00 2001 From: Pablo Galindo Date: Sat, 23 Mar 2019 02:39:43 +0000 Subject: [PATCH 09/32] Add context to Process classes && revert back to use get_pickler_class --- Doc/library/multiprocessing.rst | 4 ++-- Lib/multiprocessing/context.py | 28 +++++++++++++----------- Lib/multiprocessing/popen_fork.py | 4 +++- Lib/multiprocessing/popen_forkserver.py | 8 +++---- Lib/multiprocessing/popen_spawn_posix.py | 11 +++++----- Lib/multiprocessing/popen_spawn_win32.py | 8 ++++--- Lib/multiprocessing/reduction.py | 10 ++++----- Lib/test/_test_multiprocessing.py | 8 +++---- 8 files changed, 43 insertions(+), 38 deletions(-) diff --git a/Doc/library/multiprocessing.rst b/Doc/library/multiprocessing.rst index c2675994e49c12..94c82a6462f9da 100644 --- a/Doc/library/multiprocessing.rst +++ b/Doc/library/multiprocessing.rst @@ -1216,7 +1216,7 @@ the reduction mechanism: Abstract base class that can be implemented in order to replace the standard reduction mechanism used in multiprocessing - .. function:: get_pickler(): + .. function:: get_pickler_class(): This method must return an instance of a subclass of :class:`pickler.Pickler` to be used by the multiprocessing reducer mechanism. @@ -1245,7 +1245,7 @@ version 2 to be able to communicate with a Python 2.x programs.:: return super().dumps(obj, protocol=pickle_protocol) class PickleProtocol2Reducer(AbstractReducer): - def get_pickler(self): + def get_pickler_class(self): return ForkingPicklerProtocol2 multiprocessing.set_reducer(PickleProtocol2Reducer) diff --git a/Lib/multiprocessing/context.py b/Lib/multiprocessing/context.py index 527d583f8ba76f..fd0bfeef70b16d 100644 --- a/Lib/multiprocessing/context.py +++ b/Lib/multiprocessing/context.py @@ -220,9 +220,9 @@ def set_reducer(self, reduction): if not isinstance(reduction, original_reducer.AbstractReducer): raise TypeError("Custom reducers must be instances of a " "subclass of multiprocessing.reduction.AbstractReducer") - if not isinstance(reduction.get_pickler(), pickle.Pickler): - raise TypeError("Custom reducers must return an instance of " - "pickler.Pickler in the get_pickler() method") + if not issubclass(reduction.get_pickler_class(), pickle.Pickler): + raise TypeError("Custom reducers must return an subclass of " + "pickler.Pickler in the get_pickler_class() method") self._reducer = reduction @@ -236,8 +236,10 @@ def _check_available(self): class Process(process.BaseProcess): _start_method = None @staticmethod - def _Popen(process_obj): - return _default_context.get_context().Process._Popen(process_obj) + def _Popen(process_obj, ctx=None): + if ctx is None: + ctx = _default_context.get_context() + return ctx.Process._Popen(process_obj, ctx) class DefaultContext(BaseContext): Process = Process @@ -287,23 +289,23 @@ def get_all_start_methods(self): class ForkProcess(process.BaseProcess): _start_method = 'fork' @staticmethod - def _Popen(process_obj): + def _Popen(process_obj, ctx=None): from .popen_fork import Popen - return Popen(process_obj) + return Popen(process_obj, ctx=ctx) class SpawnProcess(process.BaseProcess): _start_method = 'spawn' @staticmethod - def _Popen(process_obj): + def _Popen(process_obj, ctx=None): from .popen_spawn_posix import Popen - return Popen(process_obj) + return Popen(process_obj, ctx=ctx) class ForkServerProcess(process.BaseProcess): _start_method = 'forkserver' @staticmethod - def _Popen(process_obj): + def _Popen(process_obj, ctx=None): from .popen_forkserver import Popen - return Popen(process_obj) + return Popen(process_obj, ctx=ctx) class ForkContext(BaseContext): _name = 'fork' @@ -337,9 +339,9 @@ def _check_available(self): class SpawnProcess(process.BaseProcess): _start_method = 'spawn' @staticmethod - def _Popen(process_obj): + def _Popen(process_obj, ctx=None): from .popen_spawn_win32 import Popen - return Popen(process_obj) + return Popen(process_obj, ctx) class SpawnContext(BaseContext): _name = 'spawn' diff --git a/Lib/multiprocessing/popen_fork.py b/Lib/multiprocessing/popen_fork.py index a65b06f1b2c75e..69ae4574c47919 100644 --- a/Lib/multiprocessing/popen_fork.py +++ b/Lib/multiprocessing/popen_fork.py @@ -1,6 +1,7 @@ import os import signal +import multiprocessing from . import util __all__ = ['Popen'] @@ -12,10 +13,11 @@ class Popen(object): method = 'fork' - def __init__(self, process_obj): + def __init__(self, process_obj, ctx=None): util._flush_std_streams() self.returncode = None self.finalizer = None + self.ctx = ctx or multiprocessing.get_context() self._launch(process_obj) def duplicate_for_child(self, fd): diff --git a/Lib/multiprocessing/popen_forkserver.py b/Lib/multiprocessing/popen_forkserver.py index 93c9a17cbc1db0..0902a427299c9a 100644 --- a/Lib/multiprocessing/popen_forkserver.py +++ b/Lib/multiprocessing/popen_forkserver.py @@ -31,9 +31,9 @@ class Popen(popen_fork.Popen): method = 'forkserver' DupFd = _DupFd - def __init__(self, process_obj): + def __init__(self, process_obj, ctx=None): self._fds = [] - super().__init__(process_obj) + super().__init__(process_obj, ctx) def duplicate_for_child(self, fd): self._fds.append(fd) @@ -44,8 +44,8 @@ def _launch(self, process_obj): buf = io.BytesIO() set_spawning_popen(self) try: - context.reduction.dump(prep_data, buf) - context.reduction.dump(process_obj, buf) + self.ctx.get_reducer().dump(prep_data, buf) + self.ctx.get_reducer().dump(process_obj, buf) finally: set_spawning_popen(None) diff --git a/Lib/multiprocessing/popen_spawn_posix.py b/Lib/multiprocessing/popen_spawn_posix.py index a3572c68c1d501..5b588f74c10f91 100644 --- a/Lib/multiprocessing/popen_spawn_posix.py +++ b/Lib/multiprocessing/popen_spawn_posix.py @@ -1,7 +1,7 @@ import io import os -from . import context +import multiprocessing from .context import set_spawning_popen from . import popen_fork from . import spawn @@ -28,9 +28,10 @@ class Popen(popen_fork.Popen): method = 'spawn' DupFd = _DupFd - def __init__(self, process_obj): + def __init__(self, process_obj, ctx=None): self._fds = [] - super().__init__(process_obj) + ctx = ctx or multiprocessing.get_context() + super().__init__(process_obj, ctx) def duplicate_for_child(self, fd): self._fds.append(fd) @@ -44,8 +45,8 @@ def _launch(self, process_obj): fp = io.BytesIO() set_spawning_popen(self) try: - context.reduction.dump(prep_data, fp) - context.reduction.dump(process_obj, fp) + self.ctx.get_reducer().dump(prep_data, fp) + self.ctx.get_reducer().dump(process_obj, fp) finally: set_spawning_popen(None) diff --git a/Lib/multiprocessing/popen_spawn_win32.py b/Lib/multiprocessing/popen_spawn_win32.py index 4086f294870115..83112027a26e91 100644 --- a/Lib/multiprocessing/popen_spawn_win32.py +++ b/Lib/multiprocessing/popen_spawn_win32.py @@ -4,6 +4,7 @@ import sys import _winapi +import multiprocessing from . import context from .context import get_spawning_popen, set_spawning_popen from . import spawn @@ -42,7 +43,7 @@ class Popen(object): ''' method = 'spawn' - def __init__(self, process_obj): + def __init__(self, process_obj, ctx=None): prep_data = spawn.get_preparation_data(process_obj._name) # read end of pipe will be duplicated by the child process @@ -51,6 +52,7 @@ def __init__(self, process_obj): # bpo-33929: Previously, the read end of pipe was "stolen" by the child # process, but it leaked a handle if the child process had been # terminated before it could steal the handle from the parent process. + self.ctx = ctx or multiprocessing.get_context() rhandle, whandle = _winapi.CreatePipe(None, 0) wfd = msvcrt.open_osfhandle(whandle, 0) cmd = spawn.get_command_line(parent_pid=os.getpid(), @@ -90,8 +92,8 @@ def __init__(self, process_obj): # send information to child set_spawning_popen(self) try: - context.reduction.dump(prep_data, to_child) - context.reduction.dump(process_obj, to_child) + self.ctx.get_reducer().dump(prep_data, to_child) + self.ctx.get_reducer().dump(process_obj, to_child) finally: set_spawning_popen(None) diff --git a/Lib/multiprocessing/reduction.py b/Lib/multiprocessing/reduction.py index 948e524c77bfa6..87c3a111771be6 100644 --- a/Lib/multiprocessing/reduction.py +++ b/Lib/multiprocessing/reduction.py @@ -299,18 +299,18 @@ def HAVE_SEND_HANDLE(self): hasattr(socket.socket, 'sendmsg'))) @abc.abstractmethod - def get_pickler(self): + def get_pickler_class(self): pass def register(self, type_, reduce_): - return self.get_pickler().register(type_, reduce_) + return self.get_pickler_class().register(type_, reduce_) def dump(self, obj, file, protocol=None): - self.get_pickler()(file, protocol).dump(obj) + self.get_pickler_class()(file, protocol).dump(obj) def dumps(self, obj, protocol=None): - return self.get_pickler().dumps(obj, protocol=protocol) + return self.get_pickler_class().dumps(obj, protocol=protocol) def loads(self, s, *, fix_imports=True, encoding="ASCII", errors="strict"): - return self.get_pickler().loads(s, fix_imports=fix_imports, + return self.get_pickler_class().loads(s, fix_imports=fix_imports, encoding=encoding, errors=errors) diff --git a/Lib/test/_test_multiprocessing.py b/Lib/test/_test_multiprocessing.py index 79b98f47123bc1..83003323f94bf0 100644 --- a/Lib/test/_test_multiprocessing.py +++ b/Lib/test/_test_multiprocessing.py @@ -5482,11 +5482,9 @@ class it returns. Mocks are not used because they do not play good with pickling""" def __init__(self, spy): self.spy = spy - def get_pickler(self): + def get_pickler_class(self): spy = self.spy - class Pickler(pickle.Pickler): - def __init__(self): - pass + class Pickler(multiprocessing.reduction.ForkingPickler): @classmethod def dumps(self, obj, protocol=None): spy["dumps"] += 1 @@ -5495,7 +5493,7 @@ def dumps(self, obj, protocol=None): def loads(self, *args, **kwargs): spy["loads"] += 1 return pickle.loads(*args, **kwargs) - return Pickler() + return Pickler class _TestCustomReducer(BaseTestCase): """Test case for the global reducer""" From 9ccc0672b26e7afe9701d8d1819ffdc118199993 Mon Sep 17 00:00:00 2001 From: Pablo Galindo Date: Mon, 27 May 2019 15:32:49 +0100 Subject: [PATCH 10/32] Adress feedback and add custom context for Listener/Clients --- Doc/library/multiprocessing.rst | 18 +++++----- Lib/multiprocessing/connection.py | 30 ++++++++-------- Lib/multiprocessing/context.py | 18 ++++++++-- Lib/multiprocessing/spawn.py | 8 ++--- Lib/test/_test_multiprocessing.py | 59 +++++++++++++++++++++++++------ 5 files changed, 90 insertions(+), 43 deletions(-) diff --git a/Doc/library/multiprocessing.rst b/Doc/library/multiprocessing.rst index 94c82a6462f9da..4b4c9729fd1d0c 100644 --- a/Doc/library/multiprocessing.rst +++ b/Doc/library/multiprocessing.rst @@ -1216,18 +1216,18 @@ the reduction mechanism: Abstract base class that can be implemented in order to replace the standard reduction mechanism used in multiprocessing - .. function:: get_pickler_class(): + .. method:: get_pickler_class(): - This method must return an instance of a subclass of :class:`pickler.Pickler` - to be used by the multiprocessing reducer mechanism. + This method must return an subclass of :class:`pickler.Pickler` to be used by + the multiprocessing reducer mechanism. .. currentmodule:: multiprocessing .. method:: set_reducer(reduction) - Sets a reduction class to be used for serialization and deserialization by the module - primitive internals. **reduction** must be an instance of a subclass of - :class:`multiprocessing.reduction.AbstractReducer`. + Sets a reduction instance to be used for serialization and deserialization + by the module primitive internals. **reduction** must be an instance of a + subclass of :class:`multiprocessing.reduction.AbstractReducer`. .. method:: get_reducer() @@ -1241,14 +1241,14 @@ version 2 to be able to communicate with a Python 2.x programs.:: class ForkingPicklerProtocol2(ForkingPickler): @classmethod - def dumps(cls, obj, pickle_protocol=2): - return super().dumps(obj, protocol=pickle_protocol) + def dumps(cls, obj, protocol=2): + return super().dumps(obj, protocol=protocol) class PickleProtocol2Reducer(AbstractReducer): def get_pickler_class(self): return ForkingPicklerProtocol2 - multiprocessing.set_reducer(PickleProtocol2Reducer) + multiprocessing.set_reducer(PickleProtocol2Reducer()) Notice that using :meth:`multiprocessing.set_reducer` changes the reducer globally. If changing this setting globally is undesirable you could call :meth:`context.set_reducer`, diff --git a/Lib/multiprocessing/connection.py b/Lib/multiprocessing/connection.py index 3b59f18157d239..b79948b84b9f19 100644 --- a/Lib/multiprocessing/connection.py +++ b/Lib/multiprocessing/connection.py @@ -437,16 +437,16 @@ class Listener(object): This is a wrapper for a bound socket which is 'listening' for connections, or for a Windows named pipe. ''' - def __init__(self, address=None, family=None, backlog=1, authkey=None): + def __init__(self, address=None, family=None, backlog=1, authkey=None, *, ctx=None): family = family or (address and address_type(address)) \ or default_family address = address or arbitrary_address(family) _validate_family(family) if family == 'AF_PIPE': - self._listener = PipeListener(address, backlog) + self._listener = PipeListener(address, backlog, ctx=ctx) else: - self._listener = SocketListener(address, family, backlog) + self._listener = SocketListener(address, family, backlog, ctx=ctx) if authkey is not None and not isinstance(authkey, bytes): raise TypeError('authkey should be a byte string') @@ -491,16 +491,16 @@ def __exit__(self, exc_type, exc_value, exc_tb): self.close() -def Client(address, family=None, authkey=None): +def Client(address, family=None, authkey=None, *, ctx=None): ''' Returns a connection to the address of a `Listener` ''' family = family or address_type(address) _validate_family(family) if family == 'AF_PIPE': - c = PipeClient(address) + c = PipeClient(address, ctx=ctx) else: - c = SocketClient(address) + c = SocketClient(address, ctx=ctx) if authkey is not None and not isinstance(authkey, bytes): raise TypeError('authkey should be a byte string') @@ -581,7 +581,7 @@ class SocketListener(object): ''' Representation of a socket which is bound to an address and listening ''' - def __init__(self, address, family, backlog=1): + def __init__(self, address, family, backlog=1, *, ctx=None): self._socket = socket.socket(getattr(socket, family)) try: # SO_REUSEADDR has different semantics on Windows (issue #2550). @@ -604,11 +604,12 @@ def __init__(self, address, family, backlog=1): ) else: self._unlink = None + self._ctx = ctx def accept(self): s, self._last_accepted = self._socket.accept() s.setblocking(True) - return Connection(s.detach()) + return Connection(s.detach(), ctx=self._ctx) def close(self): try: @@ -620,7 +621,7 @@ def close(self): unlink() -def SocketClient(address): +def SocketClient(address, *, ctx=None): ''' Return a connection object connected to the socket given by `address` ''' @@ -628,7 +629,7 @@ def SocketClient(address): with socket.socket( getattr(socket, family) ) as s: s.setblocking(True) s.connect(address) - return Connection(s.detach()) + return Connection(s.detach(), ctx=ctx) # # Definitions for connections based on named pipes @@ -640,7 +641,7 @@ class PipeListener(object): ''' Representation of a named pipe ''' - def __init__(self, address, backlog=None): + def __init__(self, address, backlog=None, *, ctx=None): self._address = address self._handle_queue = [self._new_handle(first=True)] @@ -650,6 +651,7 @@ def __init__(self, address, backlog=None): self, PipeListener._finalize_pipe_listener, args=(self._handle_queue, self._address), exitpriority=0 ) + self._ctx = ctx def _new_handle(self, first=False): flags = _winapi.PIPE_ACCESS_DUPLEX | _winapi.FILE_FLAG_OVERLAPPED @@ -684,7 +686,7 @@ def accept(self): finally: _, err = ov.GetOverlappedResult(True) assert err == 0 - return PipeConnection(handle) + return PipeConnection(handle, ctx=self._ctx) @staticmethod def _finalize_pipe_listener(queue, address): @@ -692,7 +694,7 @@ def _finalize_pipe_listener(queue, address): for handle in queue: _winapi.CloseHandle(handle) - def PipeClient(address): + def PipeClient(address, *, ctx=None): ''' Return a connection object connected to the pipe given by `address` ''' @@ -717,7 +719,7 @@ def PipeClient(address): _winapi.SetNamedPipeHandleState( h, _winapi.PIPE_READMODE_MESSAGE, None, None ) - return PipeConnection(h) + return PipeConnection(h, ctx=ctx) # # Authentication stuff diff --git a/Lib/multiprocessing/context.py b/Lib/multiprocessing/context.py index fd0bfeef70b16d..daa1965ebdd179 100644 --- a/Lib/multiprocessing/context.py +++ b/Lib/multiprocessing/context.py @@ -66,6 +66,16 @@ def Pipe(self, duplex=True): from .connection import Pipe return Pipe(duplex) + def Listener(self, *args, **kwargs): + '''Returns a connection Listener object''' + from .connection import Listener + return Listener(*args, ctx=self.get_context(), **kwargs) + + def Client(self, *args, **kwargs): + '''Returns a connection Client object''' + from .connection import Client + return Client(*args, ctx=self.get_context(), **kwargs) + def Lock(self): '''Returns a non-recursive lock object''' from .synchronize import Lock @@ -207,14 +217,16 @@ def set_start_method(self, method, force=False): def get_reducer(self): '''Controls how objects will be reduced to a form that can be shared with other processes.''' - reducer = self._reducer + ctx = self.get_context() + reducer = ctx._reducer if reducer is None: return reduction return reducer def set_reducer(self, reduction): + ctx = self.get_context() if reduction is original_reducer: - self._reducer = None + ctx._reducer = None return if not isinstance(reduction, original_reducer.AbstractReducer): @@ -224,7 +236,7 @@ def set_reducer(self, reduction): raise TypeError("Custom reducers must return an subclass of " "pickler.Pickler in the get_pickler_class() method") - self._reducer = reduction + ctx._reducer = reduction def _check_available(self): pass diff --git a/Lib/multiprocessing/spawn.py b/Lib/multiprocessing/spawn.py index 81ab3f49e61100..3e114d7e446c14 100644 --- a/Lib/multiprocessing/spawn.py +++ b/Lib/multiprocessing/spawn.py @@ -104,12 +104,8 @@ def spawn_main(pipe_handle, parent_pid=None, tracker_fd=None): False, parent_pid) else: source_process = None - try: - new_handle = context.reduction.duplicate(pipe_handle, - source_process=source_process) - finally: - if source_process is not None: - _winapi.CloseHandle(source_process) + new_handle = context.reduction.duplicate(pipe_handle, + source_process=source_process) fd = msvcrt.open_osfhandle(new_handle, os.O_RDONLY) parent_sentinel = source_process else: diff --git a/Lib/test/_test_multiprocessing.py b/Lib/test/_test_multiprocessing.py index 83003323f94bf0..c2251ba88c7fd4 100644 --- a/Lib/test/_test_multiprocessing.py +++ b/Lib/test/_test_multiprocessing.py @@ -5528,27 +5528,43 @@ def test_connection_custom_reducer(self): def _put_and_get_in_queue(cls, queue, parent_can_continue): parent_can_continue.set() queue.put("Something") - queue.get() + queue.get(timeout=TIMEOUT2) + close_queue(queue) + parent_can_continue.set() @unittest.skipUnless(HAS_REDUCTION, "test needs multiprocessing.reduction") def test_queue_custom_reducer(self): queue = self.Queue() parent_can_continue = self.Event() p = self.Process(target=self._put_and_get_in_queue, - args=(queue, parent_can_continue)) - p.daemon = True + args=(queue, parent_can_continue)) p.start() parent_can_continue.wait() element = queue.get(timeout=TIMEOUT3) self.assertEqual(element, "Something") queue.put("Other_Something") + parent_can_continue.wait() + close_queue(queue) p.join(timeout=TIMEOUT3) - self.assertEqual(p.exitcode, 0) + p.terminate() self.assertEqual(element, "Something") self.assertEqual(self.spy['loads'], 1) self.assertEqual(self.spy['dumps'], 1) - close_queue(queue) + self.assertEqual(p.exitcode, 0) + @unittest.skipUnless(HAS_REDUCTION, "test needs multiprocessing.reduction") + def test_listener_custom_reduction(self): + with self.connection.Listener() as l: + with self.connection.Client(l.address) as c: + with l.accept() as d: + c.send(1729) + self.assertEqual(d.recv(), 1729) + + if self.TYPE == 'processes': + self.assertRaises(OSError, l.accept) + + self.assertEqual(self.spy['loads'], 1) + self.assertEqual(self.spy['dumps'], 1) class CustomContext(multiprocessing.context.BaseContext): _name = "custom" @@ -5576,7 +5592,9 @@ def tearDown(self): def _put_and_get_in_queue(cls, queue, parent_can_continue): parent_can_continue.set() queue.put("Something") - queue.get() + queue.get(timeout=TIMEOUT2) + close_queue(queue) + parent_can_continue.set() @unittest.skipUnless(HAS_REDUCTION, "test needs multiprocessing.reduction") def test_queue_custom_reducer_over_custom_context(self): @@ -5587,20 +5605,41 @@ def test_queue_custom_reducer_over_custom_context(self): parent_can_continue = self.custom_ctx.Event() p = self.custom_ctx.Process(target=self._put_and_get_in_queue, args=(queue, parent_can_continue)) - p.daemon = True p.start() parent_can_continue.wait() element = queue.get(timeout=TIMEOUT3) self.assertEqual(element, "Something") queue.put("Other_Something") + parent_can_continue.wait() + close_queue(queue) p.join(timeout=TIMEOUT3) - self.assertEqual(p.exitcode, 0) + p.terminate() self.assertEqual(element, "Something") self.assertEqual(self.custom_spy['loads'], 1) self.assertEqual(self.custom_spy['dumps'], 1) self.assertEqual(self.default_spy['loads'], 0) self.assertEqual(self.default_spy['dumps'], 0) - close_queue(queue) + self.assertEqual(p.exitcode, 0) + + @unittest.skipUnless(HAS_REDUCTION, "test needs multiprocessing.reduction") + def test_listener_custom_reduction_custom_context(self): + self.default_ctx.set_reducer(SpyReducer(self.default_spy)) + self.custom_ctx.set_reducer(SpyReducer(self.custom_spy)) + + with self.custom_ctx.Listener() as l: + with self.custom_ctx.Client(l.address) as c: + with l.accept() as d: + c.send(1729) + self.assertEqual(d.recv(), 1729) + + if self.TYPE == 'processes': + self.assertRaises(OSError, l.accept) + + self.assertEqual(self.custom_spy['loads'], 1) + self.assertEqual(self.custom_spy['dumps'], 1) + self.assertEqual(self.default_spy['loads'], 0) + self.assertEqual(self.default_spy['dumps'], 0) + @unittest.skipUnless(HAS_REDUCTION, "test needs multiprocessing.reduction") def test_queue_custom_reducer_over_default_context(self): @@ -5627,8 +5666,6 @@ def test_queue_custom_reducer_over_default_context(self): close_queue(queue) - - class _TestCustomReducerInvalidParameters(BaseTestCase): def test_setting_invalid_reducer(self): class BadReducer(): From 8704e8eabd2c40375261c679860647a8fd0fc8f0 Mon Sep 17 00:00:00 2001 From: Pablo Galindo Date: Mon, 27 May 2019 20:46:40 +0100 Subject: [PATCH 11/32] Provide and document a base class (AbstractPickler) for custom picklers --- Doc/library/multiprocessing.rst | 21 ++++++++++++++++--- Lib/multiprocessing/context.py | 5 +++-- Lib/multiprocessing/reduction.py | 10 ++++++++- Lib/test/_test_multiprocessing.py | 35 +++++++++++++++---------------- 4 files changed, 47 insertions(+), 24 deletions(-) diff --git a/Doc/library/multiprocessing.rst b/Doc/library/multiprocessing.rst index 4b4c9729fd1d0c..842a1da38ac7a3 100644 --- a/Doc/library/multiprocessing.rst +++ b/Doc/library/multiprocessing.rst @@ -1211,6 +1211,21 @@ the reduction mechanism: .. currentmodule:: multiprocessing.reduction +.. class:: AbstractPickler() + Abstract base class that can be implemented in order to override specidifc + methods of the reduction machinery used by multiprocessing. + + .. method:: dump(obj) + Write a pickled representation of obj to the open file. + + Defaults to :meth:`pickle.Pickle.dump` + + .. classmethod:: loads(bytes_object, *, fix_imports=True, encoding="ASCII", errors="strict") + Read a pickled object hierarchy from a bytes object and return the reconstituted + object hierarchy specified therein. + + Defaults to :func:`pickle.loads` + .. class:: AbstractReducer() Abstract base class that can be implemented in order to replace the standard @@ -1237,16 +1252,16 @@ For example, substituting the reducer class to use the :mod:`pickle` protocol version 2 to be able to communicate with a Python 2.x programs.:: import multiprocessing - from multiprocessing.reduction import AbstractReducer, ForkingPickler + from multiprocessing.reduction import AbstractReducer, AbstractPickler - class ForkingPicklerProtocol2(ForkingPickler): + class PicklerProtocol2(AbstractPickler): @classmethod def dumps(cls, obj, protocol=2): return super().dumps(obj, protocol=protocol) class PickleProtocol2Reducer(AbstractReducer): def get_pickler_class(self): - return ForkingPicklerProtocol2 + return PicklerProtocol2 multiprocessing.set_reducer(PickleProtocol2Reducer()) diff --git a/Lib/multiprocessing/context.py b/Lib/multiprocessing/context.py index daa1965ebdd179..b152c6ba5bbfba 100644 --- a/Lib/multiprocessing/context.py +++ b/Lib/multiprocessing/context.py @@ -232,9 +232,10 @@ def set_reducer(self, reduction): if not isinstance(reduction, original_reducer.AbstractReducer): raise TypeError("Custom reducers must be instances of a " "subclass of multiprocessing.reduction.AbstractReducer") - if not issubclass(reduction.get_pickler_class(), pickle.Pickler): + if not issubclass(reduction.get_pickler_class(), original_reducer.AbstractPickler): raise TypeError("Custom reducers must return an subclass of " - "pickler.Pickler in the get_pickler_class() method") + "multiprocessing.reduction.AbstractPickler " + "in the get_pickler_class() method") ctx._reducer = reduction diff --git a/Lib/multiprocessing/reduction.py b/Lib/multiprocessing/reduction.py index 87c3a111771be6..72d80fbb3729e8 100644 --- a/Lib/multiprocessing/reduction.py +++ b/Lib/multiprocessing/reduction.py @@ -51,7 +51,15 @@ def dumps(cls, obj, protocol=None): cls(buf, protocol).dump(obj) return buf.getbuffer() - loads = pickle.loads + @classmethod + def loads(cls, bytes_object, *, fix_imports=True, + encoding="ASCII", errors="strict"): + return pickle.loads(bytes_object, fix_imports=fix_imports, + encoding=encoding, errors=errors) + + +class AbstractPickler(ForkingPickler): + pass def register(type_, reduce_): diff --git a/Lib/test/_test_multiprocessing.py b/Lib/test/_test_multiprocessing.py index c2251ba88c7fd4..ea52ee4122444f 100644 --- a/Lib/test/_test_multiprocessing.py +++ b/Lib/test/_test_multiprocessing.py @@ -5484,13 +5484,12 @@ def __init__(self, spy): self.spy = spy def get_pickler_class(self): spy = self.spy - class Pickler(multiprocessing.reduction.ForkingPickler): + class Pickler(multiprocessing.reduction.AbstractPickler): + def dump(self, obj, protocol=None): + spy["dump"] += 1 + super().dump(obj) @classmethod - def dumps(self, obj, protocol=None): - spy["dumps"] += 1 - return pickle.dumps(obj) - @classmethod - def loads(self, *args, **kwargs): + def loads(cls, *args, **kwargs): spy["loads"] += 1 return pickle.loads(*args, **kwargs) return Pickler @@ -5503,7 +5502,7 @@ class _TestCustomReducer(BaseTestCase): def setUp(self): self.context = multiprocessing.get_context() self.original_reducer = multiprocessing.get_reducer() - self.spy = {'loads': 0, 'dumps': 0} + self.spy = {'loads': 0, 'dump': 0} self.context.set_reducer(SpyReducer(self.spy)) def tearDown(self): @@ -5522,7 +5521,7 @@ def test_connection_custom_reducer(self): self.assertRaises(OSError, l.accept) self.assertEqual(self.spy['loads'], 1) - self.assertEqual(self.spy['dumps'], 1) + self.assertEqual(self.spy['dump'], 1) @classmethod def _put_and_get_in_queue(cls, queue, parent_can_continue): @@ -5549,7 +5548,7 @@ def test_queue_custom_reducer(self): p.terminate() self.assertEqual(element, "Something") self.assertEqual(self.spy['loads'], 1) - self.assertEqual(self.spy['dumps'], 1) + self.assertEqual(self.spy['dump'], 3) self.assertEqual(p.exitcode, 0) @unittest.skipUnless(HAS_REDUCTION, "test needs multiprocessing.reduction") @@ -5564,7 +5563,7 @@ def test_listener_custom_reduction(self): self.assertRaises(OSError, l.accept) self.assertEqual(self.spy['loads'], 1) - self.assertEqual(self.spy['dumps'], 1) + self.assertEqual(self.spy['dump'], 1) class CustomContext(multiprocessing.context.BaseContext): _name = "custom" @@ -5581,8 +5580,8 @@ def setUp(self): self.default_ctx = multiprocessing.get_context() self.original_custom_reducer = self.custom_ctx.get_reducer() self.original_default_reducer = self.default_ctx.get_reducer() - self.default_spy = {'loads': 0, 'dumps': 0} - self.custom_spy = {'loads': 0, 'dumps': 0} + self.default_spy = {'loads': 0, 'dump': 0} + self.custom_spy = {'loads': 0, 'dump': 0} def tearDown(self): self.custom_ctx.set_reducer(self.original_custom_reducer) @@ -5616,9 +5615,9 @@ def test_queue_custom_reducer_over_custom_context(self): p.terminate() self.assertEqual(element, "Something") self.assertEqual(self.custom_spy['loads'], 1) - self.assertEqual(self.custom_spy['dumps'], 1) + self.assertEqual(self.custom_spy['dump'], 1) self.assertEqual(self.default_spy['loads'], 0) - self.assertEqual(self.default_spy['dumps'], 0) + self.assertEqual(self.default_spy['dump'], 0) self.assertEqual(p.exitcode, 0) @unittest.skipUnless(HAS_REDUCTION, "test needs multiprocessing.reduction") @@ -5636,9 +5635,9 @@ def test_listener_custom_reduction_custom_context(self): self.assertRaises(OSError, l.accept) self.assertEqual(self.custom_spy['loads'], 1) - self.assertEqual(self.custom_spy['dumps'], 1) + self.assertEqual(self.custom_spy['dump'], 1) self.assertEqual(self.default_spy['loads'], 0) - self.assertEqual(self.default_spy['dumps'], 0) + self.assertEqual(self.default_spy['dump'], 0) @unittest.skipUnless(HAS_REDUCTION, "test needs multiprocessing.reduction") @@ -5660,9 +5659,9 @@ def test_queue_custom_reducer_over_default_context(self): self.assertEqual(p.exitcode, 0) self.assertEqual(element, "Something") self.assertEqual(self.custom_spy['loads'], 0) - self.assertEqual(self.custom_spy['dumps'], 0) + self.assertEqual(self.custom_spy['dump'], 0) self.assertEqual(self.default_spy['loads'], 1) - self.assertEqual(self.default_spy['dumps'], 1) + self.assertEqual(self.default_spy['dump'], 3) close_queue(queue) From 722f2b952c60224843f17515d63b43b64c29e9bc Mon Sep 17 00:00:00 2001 From: Pablo Galindo Date: Tue, 28 May 2019 00:23:12 +0100 Subject: [PATCH 12/32] Try to simplify tests --- Lib/test/_test_multiprocessing.py | 49 ++++++++++++------------------- 1 file changed, 18 insertions(+), 31 deletions(-) diff --git a/Lib/test/_test_multiprocessing.py b/Lib/test/_test_multiprocessing.py index ea52ee4122444f..27fded583eda2b 100644 --- a/Lib/test/_test_multiprocessing.py +++ b/Lib/test/_test_multiprocessing.py @@ -5524,31 +5524,26 @@ def test_connection_custom_reducer(self): self.assertEqual(self.spy['dump'], 1) @classmethod - def _put_and_get_in_queue(cls, queue, parent_can_continue): - parent_can_continue.set() + def _put_and_get_in_queue(cls, queue,): queue.put("Something") - queue.get(timeout=TIMEOUT2) - close_queue(queue) - parent_can_continue.set() + queue.get(timeout=TIMEOUT) @unittest.skipUnless(HAS_REDUCTION, "test needs multiprocessing.reduction") def test_queue_custom_reducer(self): queue = self.Queue() - parent_can_continue = self.Event() p = self.Process(target=self._put_and_get_in_queue, - args=(queue, parent_can_continue)) + args=(queue,)) p.start() - parent_can_continue.wait() - element = queue.get(timeout=TIMEOUT3) + element = queue.get(timeout=TIMEOUT) self.assertEqual(element, "Something") queue.put("Other_Something") - parent_can_continue.wait() close_queue(queue) - p.join(timeout=TIMEOUT3) + p.join(timeout=TIMEOUT) p.terminate() self.assertEqual(element, "Something") self.assertEqual(self.spy['loads'], 1) - self.assertEqual(self.spy['dump'], 3) + # Fork and spawn call this different ammount of times + self.assertGreaterEqual(self.spy['dump'], 1) self.assertEqual(p.exitcode, 0) @unittest.skipUnless(HAS_REDUCTION, "test needs multiprocessing.reduction") @@ -5588,12 +5583,9 @@ def tearDown(self): self.default_ctx.set_reducer(self.original_default_reducer) @classmethod - def _put_and_get_in_queue(cls, queue, parent_can_continue): - parent_can_continue.set() + def _put_and_get_in_queue(cls, queue): queue.put("Something") - queue.get(timeout=TIMEOUT2) - close_queue(queue) - parent_can_continue.set() + queue.get(timeout=TIMEOUT) @unittest.skipUnless(HAS_REDUCTION, "test needs multiprocessing.reduction") def test_queue_custom_reducer_over_custom_context(self): @@ -5601,24 +5593,21 @@ def test_queue_custom_reducer_over_custom_context(self): self.custom_ctx.set_reducer(SpyReducer(self.custom_spy)) queue = self.custom_ctx.Queue() - parent_can_continue = self.custom_ctx.Event() p = self.custom_ctx.Process(target=self._put_and_get_in_queue, - args=(queue, parent_can_continue)) + args=(queue,)) p.start() - parent_can_continue.wait() - element = queue.get(timeout=TIMEOUT3) + element = queue.get(timeout=TIMEOUT) self.assertEqual(element, "Something") queue.put("Other_Something") - parent_can_continue.wait() close_queue(queue) - p.join(timeout=TIMEOUT3) + p.join(timeout=TIMEOUT) p.terminate() + self.assertEqual(p.exitcode, 0) self.assertEqual(element, "Something") self.assertEqual(self.custom_spy['loads'], 1) self.assertEqual(self.custom_spy['dump'], 1) self.assertEqual(self.default_spy['loads'], 0) self.assertEqual(self.default_spy['dump'], 0) - self.assertEqual(p.exitcode, 0) @unittest.skipUnless(HAS_REDUCTION, "test needs multiprocessing.reduction") def test_listener_custom_reduction_custom_context(self): @@ -5646,22 +5635,20 @@ def test_queue_custom_reducer_over_default_context(self): self.custom_ctx.set_reducer(SpyReducer(self.custom_spy)) queue = self.default_ctx.Queue() - parent_can_continue = self.default_ctx.Event() p = self.default_ctx.Process(target=self._put_and_get_in_queue, - args=(queue, parent_can_continue)) - p.daemon = True + args=(queue, )) p.start() - parent_can_continue.wait() - element = queue.get(timeout=TIMEOUT3) + element = queue.get(timeout=TIMEOUT) self.assertEqual(element, "Something") queue.put("Other_Something") - p.join(timeout=TIMEOUT3) + p.join(timeout=TIMEOUT) self.assertEqual(p.exitcode, 0) self.assertEqual(element, "Something") self.assertEqual(self.custom_spy['loads'], 0) self.assertEqual(self.custom_spy['dump'], 0) self.assertEqual(self.default_spy['loads'], 1) - self.assertEqual(self.default_spy['dump'], 3) + # Fork and spawn call this different ammount of times + self.assertGreaterEqual(self.default_spy['dump'], 1) close_queue(queue) From 5e03ffa16d2a2c096ef04c0c0993d5022dfa61a9 Mon Sep 17 00:00:00 2001 From: Pablo Galindo Date: Tue, 28 May 2019 01:59:25 +0100 Subject: [PATCH 13/32] =?UTF-8?q?Try=20to=20pass=20down=20the=20context=20?= =?UTF-8?q?manually=20to=20Process=C2=B7?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- Lib/multiprocessing/process.py | 5 +++-- Lib/test/_test_multiprocessing.py | 2 +- 2 files changed, 4 insertions(+), 3 deletions(-) diff --git a/Lib/multiprocessing/process.py b/Lib/multiprocessing/process.py index be13c079bb89b3..b095fcc615b70a 100644 --- a/Lib/multiprocessing/process.py +++ b/Lib/multiprocessing/process.py @@ -78,7 +78,7 @@ def _Popen(self): raise NotImplementedError def __init__(self, group=None, target=None, name=None, args=(), kwargs={}, - *, daemon=None): + *, daemon=None, ctx=None): assert group is None, 'group argument must be None for now' count = next(_process_counter) self._identity = _current_process._identity + (count,) @@ -95,6 +95,7 @@ def __init__(self, group=None, target=None, name=None, args=(), kwargs={}, if daemon is not None: self.daemon = daemon _dangling.add(self) + self._ctx = ctx def _check_closed(self): if self._closed: @@ -118,7 +119,7 @@ def start(self): assert not _current_process._config.get('daemon'), \ 'daemonic processes are not allowed to have children' _cleanup() - self._popen = self._Popen(self) + self._popen = self._Popen(self, ctx=self._ctx) self._sentinel = self._popen.sentinel # Avoid a refcycle if the target function holds an indirect # reference to the process object (see bpo-30775) diff --git a/Lib/test/_test_multiprocessing.py b/Lib/test/_test_multiprocessing.py index 27fded583eda2b..a81e97740157d3 100644 --- a/Lib/test/_test_multiprocessing.py +++ b/Lib/test/_test_multiprocessing.py @@ -5594,7 +5594,7 @@ def test_queue_custom_reducer_over_custom_context(self): queue = self.custom_ctx.Queue() p = self.custom_ctx.Process(target=self._put_and_get_in_queue, - args=(queue,)) + args=(queue,), ctx=self.custom_ctx) p.start() element = queue.get(timeout=TIMEOUT) self.assertEqual(element, "Something") From 8bb0d0d08513a765a0e92d3ec53f20c97406ae5b Mon Sep 17 00:00:00 2001 From: Pierre Glaser Date: Fri, 19 Jul 2019 18:59:16 +0200 Subject: [PATCH 14/32] ENH backward-compat friendly context injection --- Lib/multiprocessing/context.py | 23 +++++++++++++++++++---- 1 file changed, 19 insertions(+), 4 deletions(-) diff --git a/Lib/multiprocessing/context.py b/Lib/multiprocessing/context.py index b152c6ba5bbfba..a968a9d54f7037 100644 --- a/Lib/multiprocessing/context.py +++ b/Lib/multiprocessing/context.py @@ -41,6 +41,7 @@ class BaseContext(object): parent_process = staticmethod(process.parent_process) active_children = staticmethod(process.active_children) _reducer = None + _custom_reduction_enabled = False def cpu_count(self): '''Returns the number of CPUs in the system''' @@ -50,6 +51,19 @@ def cpu_count(self): else: return num + @property + def Process(self): + if not self._custom_reduction_enabled: + # Ensure backward compatibility by returning a class when no + # custom reducer was specified + return self._Process + else: + return self.process_factory + + def process_factory(self, *args, **kwargs): + p = self._Process(*args, **kwargs) + p._ctx = self.get_context() + def Manager(self): '''Returns a manager associated with a running server process @@ -238,6 +252,7 @@ def set_reducer(self, reduction): "in the get_pickler_class() method") ctx._reducer = reduction + ctx._custom_reduction_enabled = True def _check_available(self): pass @@ -255,7 +270,7 @@ def _Popen(process_obj, ctx=None): return ctx.Process._Popen(process_obj, ctx) class DefaultContext(BaseContext): - Process = Process + _Process = Process def __init__(self, context): self._default_context = context @@ -322,15 +337,15 @@ def _Popen(process_obj, ctx=None): class ForkContext(BaseContext): _name = 'fork' - Process = ForkProcess + _Process = ForkProcess class SpawnContext(BaseContext): _name = 'spawn' - Process = SpawnProcess + _Process = SpawnProcess class ForkServerContext(BaseContext): _name = 'forkserver' - Process = ForkServerProcess + _Process = ForkServerProcess def _check_available(self): if not reduction.HAVE_SEND_HANDLE: raise ValueError('forkserver start method not available') From 508a992935f56ef38179156bd1569d5765719297 Mon Sep 17 00:00:00 2001 From: Pierre Glaser Date: Mon, 29 Jul 2019 16:50:15 +0200 Subject: [PATCH 15/32] make process_factory actually return a Process --- Lib/multiprocessing/context.py | 1 + 1 file changed, 1 insertion(+) diff --git a/Lib/multiprocessing/context.py b/Lib/multiprocessing/context.py index a968a9d54f7037..e7fb91663b2a98 100644 --- a/Lib/multiprocessing/context.py +++ b/Lib/multiprocessing/context.py @@ -63,6 +63,7 @@ def Process(self): def process_factory(self, *args, **kwargs): p = self._Process(*args, **kwargs) p._ctx = self.get_context() + return p def Manager(self): '''Returns a manager associated with a running server process From 984cdd9b874aa3549b3e8e42d50681681fbb9a29 Mon Sep 17 00:00:00 2001 From: Pierre Glaser Date: Wed, 31 Jul 2019 16:51:40 +0200 Subject: [PATCH 16/32] multiprocessing mirrors dynamically default_context --- Lib/multiprocessing/__init__.py | 6 +++++- 1 file changed, 5 insertions(+), 1 deletion(-) diff --git a/Lib/multiprocessing/__init__.py b/Lib/multiprocessing/__init__.py index 8336f381decaac..e1ab28a543d959 100644 --- a/Lib/multiprocessing/__init__.py +++ b/Lib/multiprocessing/__init__.py @@ -20,7 +20,11 @@ # __all__ = [x for x in dir(context._default_context) if not x.startswith('_')] -globals().update((name, getattr(context._default_context, name)) for name in __all__) + + +def __getattr__(name: str): + return getattr(context._default_context, name) + # # XXX These should not really be documented or public. From 6ce49bfbe88c7ffe73bd90050bafa19ff226a84b Mon Sep 17 00:00:00 2001 From: Pierre Glaser Date: Wed, 31 Jul 2019 16:54:18 +0200 Subject: [PATCH 17/32] ENH pass the reducer to Process (not the context) --- Lib/multiprocessing/context.py | 35 ++++++++++-------------- Lib/multiprocessing/popen_fork.py | 3 +- Lib/multiprocessing/popen_forkserver.py | 8 +++--- Lib/multiprocessing/popen_spawn_posix.py | 9 +++--- Lib/multiprocessing/popen_spawn_win32.py | 7 ++--- Lib/multiprocessing/process.py | 11 ++++++-- 6 files changed, 35 insertions(+), 38 deletions(-) diff --git a/Lib/multiprocessing/context.py b/Lib/multiprocessing/context.py index e7fb91663b2a98..a06f149ce4f114 100644 --- a/Lib/multiprocessing/context.py +++ b/Lib/multiprocessing/context.py @@ -41,7 +41,6 @@ class BaseContext(object): parent_process = staticmethod(process.parent_process) active_children = staticmethod(process.active_children) _reducer = None - _custom_reduction_enabled = False def cpu_count(self): '''Returns the number of CPUs in the system''' @@ -53,7 +52,7 @@ def cpu_count(self): @property def Process(self): - if not self._custom_reduction_enabled: + if self._reducer is None: # Ensure backward compatibility by returning a class when no # custom reducer was specified return self._Process @@ -61,8 +60,7 @@ def Process(self): return self.process_factory def process_factory(self, *args, **kwargs): - p = self._Process(*args, **kwargs) - p._ctx = self.get_context() + p = self._Process(*args, reducer=self.get_reducer(), **kwargs) return p def Manager(self): @@ -232,16 +230,16 @@ def set_start_method(self, method, force=False): def get_reducer(self): '''Controls how objects will be reduced to a form that can be shared with other processes.''' - ctx = self.get_context() - reducer = ctx._reducer + # It does not matter that _reducer binds to self or self.get_context() + # as self.set_start_method() can only be called once. + reducer = self._reducer if reducer is None: return reduction return reducer def set_reducer(self, reduction): - ctx = self.get_context() if reduction is original_reducer: - ctx._reducer = None + self._reducer = None return if not isinstance(reduction, original_reducer.AbstractReducer): @@ -252,8 +250,7 @@ def set_reducer(self, reduction): "multiprocessing.reduction.AbstractPickler " "in the get_pickler_class() method") - ctx._reducer = reduction - ctx._custom_reduction_enabled = True + self._reducer = reduction def _check_available(self): pass @@ -265,10 +262,8 @@ def _check_available(self): class Process(process.BaseProcess): _start_method = None @staticmethod - def _Popen(process_obj, ctx=None): - if ctx is None: - ctx = _default_context.get_context() - return ctx.Process._Popen(process_obj, ctx) + def _Popen(process_obj): + return _default_context.get_context()._Process._Popen(process_obj) class DefaultContext(BaseContext): _Process = Process @@ -318,23 +313,23 @@ def get_all_start_methods(self): class ForkProcess(process.BaseProcess): _start_method = 'fork' @staticmethod - def _Popen(process_obj, ctx=None): + def _Popen(process_obj): from .popen_fork import Popen - return Popen(process_obj, ctx=ctx) + return Popen(process_obj) class SpawnProcess(process.BaseProcess): _start_method = 'spawn' @staticmethod - def _Popen(process_obj, ctx=None): + def _Popen(process_obj): from .popen_spawn_posix import Popen - return Popen(process_obj, ctx=ctx) + return Popen(process_obj) class ForkServerProcess(process.BaseProcess): _start_method = 'forkserver' @staticmethod - def _Popen(process_obj, ctx=None): + def _Popen(process_obj): from .popen_forkserver import Popen - return Popen(process_obj, ctx=ctx) + return Popen(process_obj) class ForkContext(BaseContext): _name = 'fork' diff --git a/Lib/multiprocessing/popen_fork.py b/Lib/multiprocessing/popen_fork.py index 69ae4574c47919..7291513820e8b0 100644 --- a/Lib/multiprocessing/popen_fork.py +++ b/Lib/multiprocessing/popen_fork.py @@ -13,11 +13,10 @@ class Popen(object): method = 'fork' - def __init__(self, process_obj, ctx=None): + def __init__(self, process_obj): util._flush_std_streams() self.returncode = None self.finalizer = None - self.ctx = ctx or multiprocessing.get_context() self._launch(process_obj) def duplicate_for_child(self, fd): diff --git a/Lib/multiprocessing/popen_forkserver.py b/Lib/multiprocessing/popen_forkserver.py index 0902a427299c9a..9f748314371fc4 100644 --- a/Lib/multiprocessing/popen_forkserver.py +++ b/Lib/multiprocessing/popen_forkserver.py @@ -31,9 +31,9 @@ class Popen(popen_fork.Popen): method = 'forkserver' DupFd = _DupFd - def __init__(self, process_obj, ctx=None): + def __init__(self, process_obj): self._fds = [] - super().__init__(process_obj, ctx) + super().__init__(process_obj) def duplicate_for_child(self, fd): self._fds.append(fd) @@ -44,8 +44,8 @@ def _launch(self, process_obj): buf = io.BytesIO() set_spawning_popen(self) try: - self.ctx.get_reducer().dump(prep_data, buf) - self.ctx.get_reducer().dump(process_obj, buf) + process_obj.reducer.dump(prep_data, buf) + process_obj.reducer.dump(process_obj, buf) finally: set_spawning_popen(None) diff --git a/Lib/multiprocessing/popen_spawn_posix.py b/Lib/multiprocessing/popen_spawn_posix.py index 5b588f74c10f91..c9617d51d5c28a 100644 --- a/Lib/multiprocessing/popen_spawn_posix.py +++ b/Lib/multiprocessing/popen_spawn_posix.py @@ -28,10 +28,9 @@ class Popen(popen_fork.Popen): method = 'spawn' DupFd = _DupFd - def __init__(self, process_obj, ctx=None): + def __init__(self, process_obj): self._fds = [] - ctx = ctx or multiprocessing.get_context() - super().__init__(process_obj, ctx) + super().__init__(process_obj) def duplicate_for_child(self, fd): self._fds.append(fd) @@ -45,8 +44,8 @@ def _launch(self, process_obj): fp = io.BytesIO() set_spawning_popen(self) try: - self.ctx.get_reducer().dump(prep_data, fp) - self.ctx.get_reducer().dump(process_obj, fp) + process_obj.reducer.dump(prep_data, fp) + process_obj.reducer.dump(process_obj, fp) finally: set_spawning_popen(None) diff --git a/Lib/multiprocessing/popen_spawn_win32.py b/Lib/multiprocessing/popen_spawn_win32.py index 83112027a26e91..08c508c549ada5 100644 --- a/Lib/multiprocessing/popen_spawn_win32.py +++ b/Lib/multiprocessing/popen_spawn_win32.py @@ -43,7 +43,7 @@ class Popen(object): ''' method = 'spawn' - def __init__(self, process_obj, ctx=None): + def __init__(self, process_obj): prep_data = spawn.get_preparation_data(process_obj._name) # read end of pipe will be duplicated by the child process @@ -52,7 +52,6 @@ def __init__(self, process_obj, ctx=None): # bpo-33929: Previously, the read end of pipe was "stolen" by the child # process, but it leaked a handle if the child process had been # terminated before it could steal the handle from the parent process. - self.ctx = ctx or multiprocessing.get_context() rhandle, whandle = _winapi.CreatePipe(None, 0) wfd = msvcrt.open_osfhandle(whandle, 0) cmd = spawn.get_command_line(parent_pid=os.getpid(), @@ -92,8 +91,8 @@ def __init__(self, process_obj, ctx=None): # send information to child set_spawning_popen(self) try: - self.ctx.get_reducer().dump(prep_data, to_child) - self.ctx.get_reducer().dump(process_obj, to_child) + process_obj.reducer.dump(prep_data, to_child) + process_obj.reducer.dump(process_obj, to_child) finally: set_spawning_popen(None) diff --git a/Lib/multiprocessing/process.py b/Lib/multiprocessing/process.py index b095fcc615b70a..af3c5cff890992 100644 --- a/Lib/multiprocessing/process.py +++ b/Lib/multiprocessing/process.py @@ -21,6 +21,7 @@ import threading from _weakrefset import WeakSet +from . import context # # # @@ -78,7 +79,7 @@ def _Popen(self): raise NotImplementedError def __init__(self, group=None, target=None, name=None, args=(), kwargs={}, - *, daemon=None, ctx=None): + *, daemon=None, reducer=None): assert group is None, 'group argument must be None for now' count = next(_process_counter) self._identity = _current_process._identity + (count,) @@ -95,12 +96,16 @@ def __init__(self, group=None, target=None, name=None, args=(), kwargs={}, if daemon is not None: self.daemon = daemon _dangling.add(self) - self._ctx = ctx + self._reducer = reducer def _check_closed(self): if self._closed: raise ValueError("process object is closed") + @property + def reducer(self): + return self._reducer or context.reduction + def run(self): ''' Method to be run in sub-process; can be overridden in sub-class @@ -119,7 +124,7 @@ def start(self): assert not _current_process._config.get('daemon'), \ 'daemonic processes are not allowed to have children' _cleanup() - self._popen = self._Popen(self, ctx=self._ctx) + self._popen = self._Popen(self) self._sentinel = self._popen.sentinel # Avoid a refcycle if the target function holds an indirect # reference to the process object (see bpo-30775) From e2cedd1ed9db04b29ae20704a8288bc98ad499ab Mon Sep 17 00:00:00 2001 From: Pierre Glaser Date: Wed, 31 Jul 2019 17:24:32 +0200 Subject: [PATCH 18/32] adapt the tests to new changes --- Lib/test/_test_multiprocessing.py | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/Lib/test/_test_multiprocessing.py b/Lib/test/_test_multiprocessing.py index a81e97740157d3..7253113ee81bb5 100644 --- a/Lib/test/_test_multiprocessing.py +++ b/Lib/test/_test_multiprocessing.py @@ -5562,7 +5562,7 @@ def test_listener_custom_reduction(self): class CustomContext(multiprocessing.context.BaseContext): _name = "custom" - Process = multiprocessing.get_context().Process + _Process = multiprocessing.get_context().Process class _TestCustomReducerWithContext(BaseTestCase): @@ -5594,7 +5594,7 @@ def test_queue_custom_reducer_over_custom_context(self): queue = self.custom_ctx.Queue() p = self.custom_ctx.Process(target=self._put_and_get_in_queue, - args=(queue,), ctx=self.custom_ctx) + args=(queue,)) p.start() element = queue.get(timeout=TIMEOUT) self.assertEqual(element, "Something") From c99a54292881e47202077ca0c8ebac0484d085af Mon Sep 17 00:00:00 2001 From: Pierre Glaser Date: Wed, 31 Jul 2019 09:05:25 -0700 Subject: [PATCH 19/32] fix Popen/Process in windows --- Lib/multiprocessing/context.py | 6 +++--- Lib/test/_test_multiprocessing.py | 3 ++- 2 files changed, 5 insertions(+), 4 deletions(-) diff --git a/Lib/multiprocessing/context.py b/Lib/multiprocessing/context.py index a06f149ce4f114..4781d17490f4fa 100644 --- a/Lib/multiprocessing/context.py +++ b/Lib/multiprocessing/context.py @@ -363,13 +363,13 @@ def _check_available(self): class SpawnProcess(process.BaseProcess): _start_method = 'spawn' @staticmethod - def _Popen(process_obj, ctx=None): + def _Popen(process_obj,): from .popen_spawn_win32 import Popen - return Popen(process_obj, ctx) + return Popen(process_obj) class SpawnContext(BaseContext): _name = 'spawn' - Process = SpawnProcess + _Process = SpawnProcess _concrete_contexts = { 'spawn': SpawnContext(), diff --git a/Lib/test/_test_multiprocessing.py b/Lib/test/_test_multiprocessing.py index 7253113ee81bb5..fe9190ec6b6260 100644 --- a/Lib/test/_test_multiprocessing.py +++ b/Lib/test/_test_multiprocessing.py @@ -5605,7 +5605,8 @@ def test_queue_custom_reducer_over_custom_context(self): self.assertEqual(p.exitcode, 0) self.assertEqual(element, "Something") self.assertEqual(self.custom_spy['loads'], 1) - self.assertEqual(self.custom_spy['dump'], 1) + # Fork and spawn call this different ammount of times + self.assertGreaterEqual(self.custom_spy['dump'], 1) self.assertEqual(self.default_spy['loads'], 0) self.assertEqual(self.default_spy['dump'], 0) From aa9215f017741ffbbec149c42d966ff4dc835ad9 Mon Sep 17 00:00:00 2001 From: Pierre Glaser Date: Wed, 31 Jul 2019 18:08:58 +0200 Subject: [PATCH 20/32] CLN unused import --- Lib/multiprocessing/popen_fork.py | 1 - 1 file changed, 1 deletion(-) diff --git a/Lib/multiprocessing/popen_fork.py b/Lib/multiprocessing/popen_fork.py index 7291513820e8b0..a65b06f1b2c75e 100644 --- a/Lib/multiprocessing/popen_fork.py +++ b/Lib/multiprocessing/popen_fork.py @@ -1,7 +1,6 @@ import os import signal -import multiprocessing from . import util __all__ = ['Popen'] From bb7d03e0b8703c9cf9b1916b084dbcb1bf14af10 Mon Sep 17 00:00:00 2001 From: Pierre Glaser Date: Thu, 1 Aug 2019 18:42:05 +0200 Subject: [PATCH 21/32] ENH pass separately pickling and unpickling routines --- Lib/multiprocessing/context.py | 7 +++++++ Lib/multiprocessing/reduction.py | 17 +++++++++++++---- 2 files changed, 20 insertions(+), 4 deletions(-) diff --git a/Lib/multiprocessing/context.py b/Lib/multiprocessing/context.py index 4781d17490f4fa..ea604e024c0dbd 100644 --- a/Lib/multiprocessing/context.py +++ b/Lib/multiprocessing/context.py @@ -1,5 +1,6 @@ import os import pickle +import _pickle import sys import threading @@ -249,6 +250,12 @@ def set_reducer(self, reduction): raise TypeError("Custom reducers must return an subclass of " "multiprocessing.reduction.AbstractPickler " "in the get_pickler_class() method") + if not issubclass(reduction.get_unpickler_class(), + original_reducer.AbstractUnpickler): + raise TypeError( + "Custom reducers must return an subclass of " + "multiprocessing.reduction.AbstractUnpickler in the " + "get_unpickler_class() method") self._reducer = reduction diff --git a/Lib/multiprocessing/reduction.py b/Lib/multiprocessing/reduction.py index 72d80fbb3729e8..17d9b554f8152b 100644 --- a/Lib/multiprocessing/reduction.py +++ b/Lib/multiprocessing/reduction.py @@ -62,6 +62,10 @@ class AbstractPickler(ForkingPickler): pass +class AbstractUnpickler(pickle.Unpickler): + pass + + def register(type_, reduce_): return ForkingPickler.register(type_, reduce_) @@ -306,9 +310,11 @@ def HAVE_SEND_HANDLE(self): hasattr(socket, 'SCM_RIGHTS') and hasattr(socket.socket, 'sendmsg'))) - @abc.abstractmethod def get_pickler_class(self): - pass + return AbstractPickler + + def get_unpickler_class(self): + return AbstractUnpickler def register(self, type_, reduce_): return self.get_pickler_class().register(type_, reduce_) @@ -320,5 +326,8 @@ def dumps(self, obj, protocol=None): return self.get_pickler_class().dumps(obj, protocol=protocol) def loads(self, s, *, fix_imports=True, encoding="ASCII", errors="strict"): - return self.get_pickler_class().loads(s, fix_imports=fix_imports, - encoding=encoding, errors=errors) + unpickler_cls = self.get_unpickler_class() + file = io.BytesIO(s) + unpickler = unpickler_cls( + file, fix_imports=fix_imports, encoding=encoding, errors=errors) + return unpickler.load() From 693e2b66efb89f8c7e792a2979db0b4d2abc170a Mon Sep 17 00:00:00 2001 From: Pierre Glaser Date: Thu, 1 Aug 2019 18:42:32 +0200 Subject: [PATCH 22/32] update the tests to test pickling and unpickling --- Lib/test/_test_multiprocessing.py | 291 +++++++++++++++++++----------- 1 file changed, 189 insertions(+), 102 deletions(-) diff --git a/Lib/test/_test_multiprocessing.py b/Lib/test/_test_multiprocessing.py index fe9190ec6b6260..f7145f61cc391b 100644 --- a/Lib/test/_test_multiprocessing.py +++ b/Lib/test/_test_multiprocessing.py @@ -5476,52 +5476,92 @@ def test__all__(self): blacklist=['SUBDEBUG', 'SUBWARNING']) -class SpyReducer(reduction.AbstractReducer): +class SpyReducerBase(reduction.AbstractReducer): + def __init__(self): + self.spy = {'load': 0, 'dump': 0} + + +class SpyReducerWithPickler(SpyReducerBase): """Custom reducer that records the call to the Pickler class it returns. Mocks are not used because they do not play good with pickling""" - def __init__(self, spy): - self.spy = spy def get_pickler_class(self): spy = self.spy class Pickler(multiprocessing.reduction.AbstractPickler): def dump(self, obj, protocol=None): spy["dump"] += 1 super().dump(obj) - @classmethod - def loads(cls, *args, **kwargs): - spy["loads"] += 1 - return pickle.loads(*args, **kwargs) return Pickler + +class SpyReducerWithUnpickler(SpyReducerBase): + """Custom reducer that records the call to the Unpickler class it returns. + """ + def get_unpickler_class(self): + spy = self.spy + + class Unpickler(multiprocessing.reduction.AbstractUnpickler): + def load(self): + spy["load"] += 1 + return super().load() + return Unpickler + +class SpyReducerWithPicklerAndUnpickler(SpyReducerWithUnpickler, + SpyReducerWithPickler): + pass + + class _TestCustomReducer(BaseTestCase): - """Test case for the global reducer""" + """Test custom reducer usage. + + The goal of these tests is to make sure that multiprocessing actually uses + the Reducer given as input of the set_reducer() methods of multiprocessing + and of context objects. This is done by counting each call of load/dump + methods of the custom pickler/unpickler classes returned by the reducer, + and making sure, when expected, that these counts are greater than 0 at the + end. + """ ALLOWED_TYPES = ('processes',) def setUp(self): self.context = multiprocessing.get_context() self.original_reducer = multiprocessing.get_reducer() - self.spy = {'loads': 0, 'dump': 0} - self.context.set_reducer(SpyReducer(self.spy)) def tearDown(self): self.context.set_reducer(self.original_reducer) - del self.spy + + def _check_pickler_hits(self, reducer): + custom_pickler = ( + reducer.get_pickler_class() is not reduction.AbstractPickler) + custom_unpickler = ( + reducer.get_unpickler_class() is not reduction.AbstractUnpickler) + + self.assertEqual(reducer.spy['dump'], int(custom_pickler)) + self.assertEqual(reducer.spy['load'], int(custom_unpickler)) @unittest.skipUnless(HAS_REDUCTION, "test needs multiprocessing.reduction") def test_connection_custom_reducer(self): - with self.connection.Listener() as l: - with self.connection.Client(l.address) as c: - with l.accept() as d: - c.send(1729) - self.assertEqual(d.recv(), 1729) + for reducer_cls in [ + SpyReducerWithPickler, + SpyReducerWithUnpickler, + SpyReducerWithPicklerAndUnpickler + ]: + with self.subTest(reducer_cls=reducer_cls): + with self.connection.Listener() as l: + reducer = reducer_cls() + self.context.set_reducer(reducer) + with self.connection.Client(l.address) as c: + with l.accept() as d: + c.send(1729) + self.assertEqual(d.recv(), 1729) + + if self.TYPE == 'processes': + self.assertRaises(OSError, l.accept) + + self._check_pickler_hits(reducer) - if self.TYPE == 'processes': - self.assertRaises(OSError, l.accept) - self.assertEqual(self.spy['loads'], 1) - self.assertEqual(self.spy['dump'], 1) @classmethod def _put_and_get_in_queue(cls, queue,): @@ -5529,36 +5569,50 @@ def _put_and_get_in_queue(cls, queue,): queue.get(timeout=TIMEOUT) @unittest.skipUnless(HAS_REDUCTION, "test needs multiprocessing.reduction") - def test_queue_custom_reducer(self): - queue = self.Queue() - p = self.Process(target=self._put_and_get_in_queue, - args=(queue,)) - p.start() - element = queue.get(timeout=TIMEOUT) - self.assertEqual(element, "Something") - queue.put("Other_Something") - close_queue(queue) - p.join(timeout=TIMEOUT) - p.terminate() - self.assertEqual(element, "Something") - self.assertEqual(self.spy['loads'], 1) - # Fork and spawn call this different ammount of times - self.assertGreaterEqual(self.spy['dump'], 1) - self.assertEqual(p.exitcode, 0) + def _test_queue_custom_reducer(self): + for reducer_cls in [ + SpyReducerWithPickler, + SpyReducerWithUnpickler, + SpyReducerWithPicklerAndUnpickler + ]: + with self.subTest(reducer_cls=reducer_cls): + reducer = reducer_cls() + self.context.set_reducer(reducer) + queue = self.Queue() + p = self.Process(target=self._put_and_get_in_queue, + args=(queue,)) + p.start() + element = queue.get(timeout=TIMEOUT) + self.assertEqual(element, "Something") + queue.put("Other_Something") + close_queue(queue) + p.join(timeout=TIMEOUT) + p.terminate() + self.assertEqual(element, "Something") + self.assertEqual(p.exitcode, 0) + self._check_pickler_hits(reducer) @unittest.skipUnless(HAS_REDUCTION, "test needs multiprocessing.reduction") def test_listener_custom_reduction(self): - with self.connection.Listener() as l: - with self.connection.Client(l.address) as c: - with l.accept() as d: - c.send(1729) - self.assertEqual(d.recv(), 1729) - - if self.TYPE == 'processes': - self.assertRaises(OSError, l.accept) + for reducer_cls in [ + SpyReducerWithPickler, + SpyReducerWithUnpickler, + SpyReducerWithPicklerAndUnpickler + ]: + with self.subTest(reducer_cls=reducer_cls): + reducer = reducer_cls() + self.context.set_reducer(reducer) + with self.connection.Listener() as l: + with self.connection.Client(l.address) as c: + with l.accept() as d: + c.send(1729) + self.assertEqual(d.recv(), 1729) + + if self.TYPE == 'processes': + self.assertRaises(OSError, l.accept) + + self._check_pickler_hits(reducer) - self.assertEqual(self.spy['loads'], 1) - self.assertEqual(self.spy['dump'], 1) class CustomContext(multiprocessing.context.BaseContext): _name = "custom" @@ -5575,13 +5629,24 @@ def setUp(self): self.default_ctx = multiprocessing.get_context() self.original_custom_reducer = self.custom_ctx.get_reducer() self.original_default_reducer = self.default_ctx.get_reducer() - self.default_spy = {'loads': 0, 'dump': 0} - self.custom_spy = {'loads': 0, 'dump': 0} def tearDown(self): self.custom_ctx.set_reducer(self.original_custom_reducer) self.default_ctx.set_reducer(self.original_default_reducer) + def _check_pickler_hits(self, reducer, alternative_reducer): + custom_pickler = ( + reducer.get_pickler_class() is not reduction.AbstractPickler) + custom_unpickler = ( + reducer.get_unpickler_class() is not reduction.AbstractUnpickler) + + + self.assertGreaterEqual(reducer.spy['dump'], int(custom_pickler)) + self.assertGreaterEqual(reducer.spy['load'], int(custom_unpickler)) + + self.assertEqual(alternative_reducer.spy['dump'], 0) + self.assertEqual(alternative_reducer.spy['load'], 0) + @classmethod def _put_and_get_in_queue(cls, queue): queue.put("Something") @@ -5589,68 +5654,83 @@ def _put_and_get_in_queue(cls, queue): @unittest.skipUnless(HAS_REDUCTION, "test needs multiprocessing.reduction") def test_queue_custom_reducer_over_custom_context(self): - self.default_ctx.set_reducer(SpyReducer(self.default_spy)) - self.custom_ctx.set_reducer(SpyReducer(self.custom_spy)) + for reducer_cls in [ + SpyReducerWithPickler, + SpyReducerWithUnpickler, + SpyReducerWithPicklerAndUnpickler + ]: + with self.subTest(reducer_cls=reducer_cls): + default_reducer = reducer_cls() + custom_reducer = reducer_cls() + + self.default_ctx.set_reducer(default_reducer) + self.custom_ctx.set_reducer(custom_reducer) + + queue = self.custom_ctx.Queue() + p = self.custom_ctx.Process(target=self._put_and_get_in_queue, + args=(queue,)) + p.start() + element = queue.get(timeout=TIMEOUT) + self.assertEqual(element, "Something") + queue.put("Other_Something") + close_queue(queue) + p.join(timeout=TIMEOUT) + p.terminate() + self.assertEqual(p.exitcode, 0) + self.assertEqual(element, "Something") + self._check_pickler_hits(custom_reducer, default_reducer) - queue = self.custom_ctx.Queue() - p = self.custom_ctx.Process(target=self._put_and_get_in_queue, - args=(queue,)) - p.start() - element = queue.get(timeout=TIMEOUT) - self.assertEqual(element, "Something") - queue.put("Other_Something") - close_queue(queue) - p.join(timeout=TIMEOUT) - p.terminate() - self.assertEqual(p.exitcode, 0) - self.assertEqual(element, "Something") - self.assertEqual(self.custom_spy['loads'], 1) - # Fork and spawn call this different ammount of times - self.assertGreaterEqual(self.custom_spy['dump'], 1) - self.assertEqual(self.default_spy['loads'], 0) - self.assertEqual(self.default_spy['dump'], 0) @unittest.skipUnless(HAS_REDUCTION, "test needs multiprocessing.reduction") def test_listener_custom_reduction_custom_context(self): - self.default_ctx.set_reducer(SpyReducer(self.default_spy)) - self.custom_ctx.set_reducer(SpyReducer(self.custom_spy)) - - with self.custom_ctx.Listener() as l: - with self.custom_ctx.Client(l.address) as c: - with l.accept() as d: - c.send(1729) - self.assertEqual(d.recv(), 1729) - - if self.TYPE == 'processes': - self.assertRaises(OSError, l.accept) - - self.assertEqual(self.custom_spy['loads'], 1) - self.assertEqual(self.custom_spy['dump'], 1) - self.assertEqual(self.default_spy['loads'], 0) - self.assertEqual(self.default_spy['dump'], 0) + for reducer_cls in [ + SpyReducerWithPickler, + SpyReducerWithUnpickler, + SpyReducerWithPicklerAndUnpickler + ]: + with self.subTest(reducer_cls=reducer_cls): + custom_reducer = reducer_cls() + default_reducer = reducer_cls() + self.default_ctx.set_reducer(default_reducer) + self.custom_ctx.set_reducer(custom_reducer) + + with self.custom_ctx.Listener() as l: + with self.custom_ctx.Client(l.address) as c: + with l.accept() as d: + c.send(1729) + self.assertEqual(d.recv(), 1729) + + if self.TYPE == 'processes': + self.assertRaises(OSError, l.accept) + self._check_pickler_hits(custom_reducer, default_reducer) @unittest.skipUnless(HAS_REDUCTION, "test needs multiprocessing.reduction") def test_queue_custom_reducer_over_default_context(self): - self.default_ctx.set_reducer(SpyReducer(self.default_spy)) - self.custom_ctx.set_reducer(SpyReducer(self.custom_spy)) - - queue = self.default_ctx.Queue() - p = self.default_ctx.Process(target=self._put_and_get_in_queue, - args=(queue, )) - p.start() - element = queue.get(timeout=TIMEOUT) - self.assertEqual(element, "Something") - queue.put("Other_Something") - p.join(timeout=TIMEOUT) - self.assertEqual(p.exitcode, 0) - self.assertEqual(element, "Something") - self.assertEqual(self.custom_spy['loads'], 0) - self.assertEqual(self.custom_spy['dump'], 0) - self.assertEqual(self.default_spy['loads'], 1) - # Fork and spawn call this different ammount of times - self.assertGreaterEqual(self.default_spy['dump'], 1) - close_queue(queue) + for reducer_cls in [ + SpyReducerWithPickler, + SpyReducerWithUnpickler, + SpyReducerWithPicklerAndUnpickler + ]: + with self.subTest(reducer_cls=reducer_cls): + default_reducer = reducer_cls() + custom_reducer = reducer_cls() + + self.default_ctx.set_reducer(default_reducer) + self.custom_ctx.set_reducer(custom_reducer) + + queue = self.default_ctx.Queue() + p = self.default_ctx.Process(target=self._put_and_get_in_queue, + args=(queue, )) + p.start() + element = queue.get(timeout=TIMEOUT) + self.assertEqual(element, "Something") + queue.put("Other_Something") + p.join(timeout=TIMEOUT) + self.assertEqual(p.exitcode, 0) + self.assertEqual(element, "Something") + close_queue(queue) + self._check_pickler_hits(default_reducer, custom_reducer) class _TestCustomReducerInvalidParameters(BaseTestCase): @@ -5659,9 +5739,16 @@ class BadReducer(): def get_pickler_class(self): return + class BadReducerUnpickler(): + def get_unpickler_class(self): + return + with self.assertRaises(TypeError): multiprocessing.set_reducer(BadReducer()) + with self.assertRaises(TypeError): + multiprocessing.set_reducer(BadReducerUnpickler()) + def test_setting_valid_reducer_that_returns_invalid_pickler(self): From ee23671076b752c3e7328947d7e14f52fe55ea1a Mon Sep 17 00:00:00 2001 From: Pierre Glaser Date: Thu, 1 Aug 2019 18:53:44 +0200 Subject: [PATCH 23/32] DOC update the docs to match Unpickler/Pickler changes --- Doc/library/multiprocessing.rst | 43 ++++++++++++++++++++++++--------- 1 file changed, 31 insertions(+), 12 deletions(-) diff --git a/Doc/library/multiprocessing.rst b/Doc/library/multiprocessing.rst index 842a1da38ac7a3..1830e272d53d36 100644 --- a/Doc/library/multiprocessing.rst +++ b/Doc/library/multiprocessing.rst @@ -1212,29 +1212,48 @@ the reduction mechanism: .. currentmodule:: multiprocessing.reduction .. class:: AbstractPickler() - Abstract base class that can be implemented in order to override specidifc - methods of the reduction machinery used by multiprocessing. + + Base class that can be implemented in order to override + serialization methods of the reduction machinery used by multiprocessing. .. method:: dump(obj) + Write a pickled representation of obj to the open file. - Defaults to :meth:`pickle.Pickle.dump` + Defaults to :meth:`pickle.Pickler.dump` + +.. class:: AbstractUnpickler(bytes_object, *, fix_imports=True, encoding="ASCII", errors="strict") + + Base class that can be implemented in order to override + multiprocessing's default unserialization mechanism. + + .. method:: load() - .. classmethod:: loads(bytes_object, *, fix_imports=True, encoding="ASCII", errors="strict") - Read a pickled object hierarchy from a bytes object and return the reconstituted - object hierarchy specified therein. + Read a pickled object hierarchy from the open file and return the + reconstituted object hierarchy specified therein. - Defaults to :func:`pickle.loads` + Defaults to :meth:`pickle.Unpickler.load` .. class:: AbstractReducer() - Abstract base class that can be implemented in order to replace the standard - reduction mechanism used in multiprocessing + Base class providing access to custom ``Pickler`` and ``Unpickler`` + classes to be used by ``multiprocessing`` when serializing objects. .. method:: get_pickler_class(): - This method must return an subclass of :class:`pickler.Pickler` to be used by - the multiprocessing reducer mechanism. + This method must return an subclass of :class:`AbstractPickler` to be used + by ``multiprocessing`` when serializing objects. + + .. method:: get_unpickler_class(): + + This method must return an subclass of :class:`AbstractUnpickler` to be used + by ``multiprocessing`` when unserializing objects. + + +Note that both methods of the :class:`AbstractReducer` are optional. If +:meth:`get_pickler_class` (resp. :meth:`get_unpickler_class`) is not +implemented, multiprocessing will fallback to the :class:`pickle.Pickler` +(resp. :class:`pickle.Unpickler`) class to serialize objects. .. currentmodule:: multiprocessing @@ -1242,7 +1261,7 @@ the reduction mechanism: Sets a reduction instance to be used for serialization and deserialization by the module primitive internals. **reduction** must be an instance of a - subclass of :class:`multiprocessing.reduction.AbstractReducer`. + subclass of :class:`AbstractReducer`. .. method:: get_reducer() From 283c69172b44005e70571258fbf91d89b66403dd Mon Sep 17 00:00:00 2001 From: Pierre Glaser Date: Thu, 1 Aug 2019 18:58:28 +0200 Subject: [PATCH 24/32] CLN cosmetics --- Lib/test/_test_multiprocessing.py | 4 +--- 1 file changed, 1 insertion(+), 3 deletions(-) diff --git a/Lib/test/_test_multiprocessing.py b/Lib/test/_test_multiprocessing.py index f7145f61cc391b..eac6fd29a02c36 100644 --- a/Lib/test/_test_multiprocessing.py +++ b/Lib/test/_test_multiprocessing.py @@ -5569,7 +5569,7 @@ def _put_and_get_in_queue(cls, queue,): queue.get(timeout=TIMEOUT) @unittest.skipUnless(HAS_REDUCTION, "test needs multiprocessing.reduction") - def _test_queue_custom_reducer(self): + def test_queue_custom_reducer(self): for reducer_cls in [ SpyReducerWithPickler, SpyReducerWithUnpickler, @@ -5680,7 +5680,6 @@ def test_queue_custom_reducer_over_custom_context(self): self.assertEqual(element, "Something") self._check_pickler_hits(custom_reducer, default_reducer) - @unittest.skipUnless(HAS_REDUCTION, "test needs multiprocessing.reduction") def test_listener_custom_reduction_custom_context(self): for reducer_cls in [ @@ -5704,7 +5703,6 @@ def test_listener_custom_reduction_custom_context(self): self.assertRaises(OSError, l.accept) self._check_pickler_hits(custom_reducer, default_reducer) - @unittest.skipUnless(HAS_REDUCTION, "test needs multiprocessing.reduction") def test_queue_custom_reducer_over_default_context(self): for reducer_cls in [ From 7154be75700577a4f3570db07f67c835141bfd76 Mon Sep 17 00:00:00 2001 From: Pierre Glaser Date: Thu, 1 Aug 2019 19:37:17 +0200 Subject: [PATCH 25/32] dont try to find the number of dump/load calls just make sure the correct reducer was used --- Lib/test/_test_multiprocessing.py | 9 ++++----- 1 file changed, 4 insertions(+), 5 deletions(-) diff --git a/Lib/test/_test_multiprocessing.py b/Lib/test/_test_multiprocessing.py index eac6fd29a02c36..fb1081ea6060e2 100644 --- a/Lib/test/_test_multiprocessing.py +++ b/Lib/test/_test_multiprocessing.py @@ -5489,7 +5489,7 @@ def get_pickler_class(self): spy = self.spy class Pickler(multiprocessing.reduction.AbstractPickler): def dump(self, obj, protocol=None): - spy["dump"] += 1 + spy["dump"] = 1 super().dump(obj) return Pickler @@ -5502,7 +5502,7 @@ def get_unpickler_class(self): class Unpickler(multiprocessing.reduction.AbstractUnpickler): def load(self): - spy["load"] += 1 + spy["load"] = 1 return super().load() return Unpickler @@ -5640,9 +5640,8 @@ def _check_pickler_hits(self, reducer, alternative_reducer): custom_unpickler = ( reducer.get_unpickler_class() is not reduction.AbstractUnpickler) - - self.assertGreaterEqual(reducer.spy['dump'], int(custom_pickler)) - self.assertGreaterEqual(reducer.spy['load'], int(custom_unpickler)) + self.assertEqual(reducer.spy['dump'], int(custom_pickler)) + self.assertEqual(reducer.spy['load'], int(custom_unpickler)) self.assertEqual(alternative_reducer.spy['dump'], 0) self.assertEqual(alternative_reducer.spy['load'], 0) From 6af4e1c13a33f3041f32baa8bf31c9d8e896f13c Mon Sep 17 00:00:00 2001 From: Pierre Glaser Date: Thu, 1 Aug 2019 22:04:30 +0200 Subject: [PATCH 26/32] TST test custom reducer for Process --- Lib/test/_test_multiprocessing.py | 53 ++++++++++++++++++++++++++++++- 1 file changed, 52 insertions(+), 1 deletion(-) diff --git a/Lib/test/_test_multiprocessing.py b/Lib/test/_test_multiprocessing.py index fb1081ea6060e2..d407e281006c98 100644 --- a/Lib/test/_test_multiprocessing.py +++ b/Lib/test/_test_multiprocessing.py @@ -5561,7 +5561,25 @@ def test_connection_custom_reducer(self): self._check_pickler_hits(reducer) + @unittest.skipUnless(HAS_REDUCTION, "test needs multiprocessing.reduction") + def test_process_custom_reducer(self): + sm = multiprocessing.get_start_method() + if sm == 'fork': + # The fork method does send data to new Process objects + self.skipTest('test not appropriate for {}'.format(sm)) + # It is not possible to customize unpickling of Process objects data + # -- we would need to send to the new process which Unpickler to use, + # which itself would require to unpickle this information before + # unpickling anyting else... using which Unpickler? + # Thus, just test the SpyReducer that customizes pickling. + reducer = SpyReducerWithPickler() + self.context.set_reducer(reducer) + p = self.context.Process(target=id, args=(2, )) + p.start() + p.join() + self.assertEqual(p.exitcode, 0) + self._check_pickler_hits(reducer) @classmethod def _put_and_get_in_queue(cls, queue,): @@ -5616,7 +5634,7 @@ def test_listener_custom_reduction(self): class CustomContext(multiprocessing.context.BaseContext): _name = "custom" - _Process = multiprocessing.get_context().Process + _Process = multiprocessing.Process class _TestCustomReducerWithContext(BaseTestCase): @@ -5651,6 +5669,24 @@ def _put_and_get_in_queue(cls, queue): queue.put("Something") queue.get(timeout=TIMEOUT) + @unittest.skipUnless(HAS_REDUCTION, "test needs multiprocessing.reduction") + def test_process_custom_reducer_over_custom_context(self): + sm = multiprocessing.get_start_method() + if sm == 'fork': + # The fork method does send data to new Process objects + self.skipTest('test not appropriate for {}'.format(sm)) + + default_reducer = SpyReducerWithPickler() + custom_reducer = SpyReducerWithPickler() + + self.default_ctx.set_reducer(default_reducer) + self.custom_ctx.set_reducer(custom_reducer) + p = self.custom_ctx.Process(target=id, args=(2, )) + p.start() + p.join() + self.assertEqual(p.exitcode, 0) + self._check_pickler_hits(custom_reducer, default_reducer) + @unittest.skipUnless(HAS_REDUCTION, "test needs multiprocessing.reduction") def test_queue_custom_reducer_over_custom_context(self): for reducer_cls in [ @@ -5729,6 +5765,21 @@ def test_queue_custom_reducer_over_default_context(self): close_queue(queue) self._check_pickler_hits(default_reducer, custom_reducer) + @unittest.skipUnless(HAS_REDUCTION, "test needs multiprocessing.reduction") + def test_process_custom_reducer_over_default_context(self): + sm = multiprocessing.get_start_method() + if sm == 'fork': + # The fork method does send data to new Process objects + self.skipTest('test not appropriate for {}'.format(sm)) + + default_reducer = SpyReducerWithPickler() + custom_reducer = SpyReducerWithPickler() + self.default_ctx.set_reducer(default_reducer) + p = self.default_ctx.Process(target=id, args=(2, )) + p.start() + p.join() + self.assertEqual(p.exitcode, 0) + self._check_pickler_hits(default_reducer, custom_reducer) class _TestCustomReducerInvalidParameters(BaseTestCase): def test_setting_invalid_reducer(self): From 515bbfdf6571363241fb3bb7223a82c40ecb88d2 Mon Sep 17 00:00:00 2001 From: Pierre Glaser Date: Thu, 1 Aug 2019 22:28:07 +0200 Subject: [PATCH 27/32] DOC adapt example --- Doc/library/multiprocessing.rst | 9 ++++++--- 1 file changed, 6 insertions(+), 3 deletions(-) diff --git a/Doc/library/multiprocessing.rst b/Doc/library/multiprocessing.rst index 1830e272d53d36..960a029a274a61 100644 --- a/Doc/library/multiprocessing.rst +++ b/Doc/library/multiprocessing.rst @@ -1274,9 +1274,12 @@ version 2 to be able to communicate with a Python 2.x programs.:: from multiprocessing.reduction import AbstractReducer, AbstractPickler class PicklerProtocol2(AbstractPickler): - @classmethod - def dumps(cls, obj, protocol=2): - return super().dumps(obj, protocol=protocol) + def __init__(self, file, protocol=2, **kwargs): + super().__init__(file, protocol, **kwargs) + + def dump(self, obj): + return super().dump(obj) + class PickleProtocol2Reducer(AbstractReducer): def get_pickler_class(self): From 525239a91cadbfe287e3136e3cfea72bd80d2d9f Mon Sep 17 00:00:00 2001 From: Pierre Glaser Date: Thu, 1 Aug 2019 22:36:14 +0200 Subject: [PATCH 28/32] CLN cosmetics, namings, comments --- Lib/multiprocessing/__init__.py | 2 +- Lib/multiprocessing/context.py | 9 +++------ 2 files changed, 4 insertions(+), 7 deletions(-) diff --git a/Lib/multiprocessing/__init__.py b/Lib/multiprocessing/__init__.py index e1ab28a543d959..2b70c044177e51 100644 --- a/Lib/multiprocessing/__init__.py +++ b/Lib/multiprocessing/__init__.py @@ -22,7 +22,7 @@ __all__ = [x for x in dir(context._default_context) if not x.startswith('_')] -def __getattr__(name: str): +def __getattr__(name): return getattr(context._default_context, name) diff --git a/Lib/multiprocessing/context.py b/Lib/multiprocessing/context.py index ea604e024c0dbd..8f2f320735a64b 100644 --- a/Lib/multiprocessing/context.py +++ b/Lib/multiprocessing/context.py @@ -58,11 +58,10 @@ def Process(self): # custom reducer was specified return self._Process else: - return self.process_factory + return self.process - def process_factory(self, *args, **kwargs): - p = self._Process(*args, reducer=self.get_reducer(), **kwargs) - return p + def process(self, *args, **kwargs): + return self._Process(*args, reducer=self.get_reducer(), **kwargs) def Manager(self): '''Returns a manager associated with a running server process @@ -231,8 +230,6 @@ def set_start_method(self, method, force=False): def get_reducer(self): '''Controls how objects will be reduced to a form that can be shared with other processes.''' - # It does not matter that _reducer binds to self or self.get_context() - # as self.set_start_method() can only be called once. reducer = self._reducer if reducer is None: return reduction From 87cffecc990b4f58a285a1ac37f3da085d7c7694 Mon Sep 17 00:00:00 2001 From: Pierre Glaser Date: Thu, 1 Aug 2019 23:07:03 +0200 Subject: [PATCH 29/32] DOC fix AbstractPickler signature --- Doc/library/multiprocessing.rst | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/Doc/library/multiprocessing.rst b/Doc/library/multiprocessing.rst index 960a029a274a61..a5436933628cc5 100644 --- a/Doc/library/multiprocessing.rst +++ b/Doc/library/multiprocessing.rst @@ -1211,7 +1211,7 @@ the reduction mechanism: .. currentmodule:: multiprocessing.reduction -.. class:: AbstractPickler() +.. class:: AbstractPickler(file, protocol) Base class that can be implemented in order to override serialization methods of the reduction machinery used by multiprocessing. From a31bdae21378aac0221fc7f7433daa4cab2986a6 Mon Sep 17 00:00:00 2001 From: Pierre Glaser Date: Wed, 7 Aug 2019 19:55:11 +0200 Subject: [PATCH 30/32] CLN make Context.process private, unused import --- Lib/multiprocessing/context.py | 4 ++-- Lib/multiprocessing/forkserver.py | 1 - 2 files changed, 2 insertions(+), 3 deletions(-) diff --git a/Lib/multiprocessing/context.py b/Lib/multiprocessing/context.py index 8f2f320735a64b..d7a0dc8dd4578b 100644 --- a/Lib/multiprocessing/context.py +++ b/Lib/multiprocessing/context.py @@ -58,9 +58,9 @@ def Process(self): # custom reducer was specified return self._Process else: - return self.process + return self._process - def process(self, *args, **kwargs): + def _process(self, *args, **kwargs): return self._Process(*args, reducer=self.get_reducer(), **kwargs) def Manager(self): diff --git a/Lib/multiprocessing/forkserver.py b/Lib/multiprocessing/forkserver.py index a644f072dc2981..01806220c013a0 100644 --- a/Lib/multiprocessing/forkserver.py +++ b/Lib/multiprocessing/forkserver.py @@ -12,7 +12,6 @@ from . import process from . import resource_tracker from . import context -from .context import reduction from . import spawn from . import util From cd505583f158587fc458e6d3f7266ecc82e60ad0 Mon Sep 17 00:00:00 2001 From: Pierre Glaser Date: Sun, 2 Feb 2020 21:43:11 +0100 Subject: [PATCH 31/32] unused imports and unnecessary changes --- Lib/multiprocessing/context.py | 4 +--- 1 file changed, 1 insertion(+), 3 deletions(-) diff --git a/Lib/multiprocessing/context.py b/Lib/multiprocessing/context.py index d7a0dc8dd4578b..824d6dc91a71c0 100644 --- a/Lib/multiprocessing/context.py +++ b/Lib/multiprocessing/context.py @@ -1,6 +1,4 @@ import os -import pickle -import _pickle import sys import threading @@ -367,7 +365,7 @@ def _check_available(self): class SpawnProcess(process.BaseProcess): _start_method = 'spawn' @staticmethod - def _Popen(process_obj,): + def _Popen(process_obj): from .popen_spawn_win32 import Popen return Popen(process_obj) From 2f117ec40d71462c8dbd4f50689eb323767141af Mon Sep 17 00:00:00 2001 From: Pierre Glaser Date: Sun, 2 Feb 2020 22:16:13 +0100 Subject: [PATCH 32/32] update to use new timeout constants --- Lib/test/_test_multiprocessing.py | 16 ++++++++-------- 1 file changed, 8 insertions(+), 8 deletions(-) diff --git a/Lib/test/_test_multiprocessing.py b/Lib/test/_test_multiprocessing.py index d407e281006c98..58201cc197ad2d 100644 --- a/Lib/test/_test_multiprocessing.py +++ b/Lib/test/_test_multiprocessing.py @@ -5584,7 +5584,7 @@ def test_process_custom_reducer(self): @classmethod def _put_and_get_in_queue(cls, queue,): queue.put("Something") - queue.get(timeout=TIMEOUT) + queue.get(timeout=support.SHORT_TIMEOUT) @unittest.skipUnless(HAS_REDUCTION, "test needs multiprocessing.reduction") def test_queue_custom_reducer(self): @@ -5600,11 +5600,11 @@ def test_queue_custom_reducer(self): p = self.Process(target=self._put_and_get_in_queue, args=(queue,)) p.start() - element = queue.get(timeout=TIMEOUT) + element = queue.get(timeout=support.SHORT_TIMEOUT) self.assertEqual(element, "Something") queue.put("Other_Something") close_queue(queue) - p.join(timeout=TIMEOUT) + p.join(timeout=support.SHORT_TIMEOUT) p.terminate() self.assertEqual(element, "Something") self.assertEqual(p.exitcode, 0) @@ -5667,7 +5667,7 @@ def _check_pickler_hits(self, reducer, alternative_reducer): @classmethod def _put_and_get_in_queue(cls, queue): queue.put("Something") - queue.get(timeout=TIMEOUT) + queue.get(timeout=support.SHORT_TIMEOUT) @unittest.skipUnless(HAS_REDUCTION, "test needs multiprocessing.reduction") def test_process_custom_reducer_over_custom_context(self): @@ -5705,11 +5705,11 @@ def test_queue_custom_reducer_over_custom_context(self): p = self.custom_ctx.Process(target=self._put_and_get_in_queue, args=(queue,)) p.start() - element = queue.get(timeout=TIMEOUT) + element = queue.get(timeout=support.SHORT_TIMEOUT) self.assertEqual(element, "Something") queue.put("Other_Something") close_queue(queue) - p.join(timeout=TIMEOUT) + p.join(timeout=support.SHORT_TIMEOUT) p.terminate() self.assertEqual(p.exitcode, 0) self.assertEqual(element, "Something") @@ -5756,10 +5756,10 @@ def test_queue_custom_reducer_over_default_context(self): p = self.default_ctx.Process(target=self._put_and_get_in_queue, args=(queue, )) p.start() - element = queue.get(timeout=TIMEOUT) + element = queue.get(timeout=support.SHORT_TIMEOUT) self.assertEqual(element, "Something") queue.put("Other_Something") - p.join(timeout=TIMEOUT) + p.join(timeout=support.SHORT_TIMEOUT) self.assertEqual(p.exitcode, 0) self.assertEqual(element, "Something") close_queue(queue)