bpo-5001: More-informative multiprocessing error messages (#3079)
https://github.com/python/cpython/commit/bd73e72b4a9f019be514954b1d40e64dc3a... commit: bd73e72b4a9f019be514954b1d40e64dc3a5e81c branch: master author: Allen W. Smith, Ph.D <drallensmith@users.noreply.github.com> committer: Antoine Pitrou <pitrou@free.fr> date: 2017-08-30T00:52:18+02:00 summary: bpo-5001: More-informative multiprocessing error messages (#3079) * Make error message more informative Replace assertions in error-reporting code with more-informative version that doesn't cause confusion over where and what the error is. * Additional clarification + get travis to check * Change from SystemError to TypeError As suggested in PR comment by @pitrou, changing from SystemError; TypeError appears appropriate. * NEWS file installation; ACKS addition (will do my best to justify it by additional work) * Making current AssertionErrors in multiprocessing more informative * Blurb added re multiprocessing managers.py, queues.py cleanup * Further multiprocessing cleanup - went through pool.py * Fix two asserts in multiprocessing/util.py * Most asserts in multiprocessing more informative * Didn't save right version * Further work on multiprocessing error messages * Correct typo * Correct typo v2 * Blasted colon... serves me right for trying to work on two things at once * Simplify NEWS entry * Update 2017-08-18-17-16-38.bpo-5001.gwnthq.rst * Update 2017-08-18-17-16-38.bpo-5001.gwnthq.rst OK, never mind. * Corrected (thanks to pitrou) error messages for notify * Remove extraneous backslash in docstring. files: A Misc/NEWS.d/next/Library/2017-08-18-17-16-38.bpo-5001.gwnthq.rst M Lib/multiprocessing/connection.py M Lib/multiprocessing/dummy/__init__.py M Lib/multiprocessing/forkserver.py M Lib/multiprocessing/heap.py M Lib/multiprocessing/managers.py M Lib/multiprocessing/pool.py M Lib/multiprocessing/popen_fork.py M Lib/multiprocessing/queues.py M Lib/multiprocessing/reduction.py M Lib/multiprocessing/resource_sharer.py M Lib/multiprocessing/semaphore_tracker.py M Lib/multiprocessing/spawn.py M Lib/multiprocessing/synchronize.py M Lib/multiprocessing/util.py diff --git a/Lib/multiprocessing/connection.py b/Lib/multiprocessing/connection.py index ba9b17cee1b..7a621a55f46 100644 --- a/Lib/multiprocessing/connection.py +++ b/Lib/multiprocessing/connection.py @@ -720,7 +720,9 @@ def PipeClient(address): def deliver_challenge(connection, authkey): import hmac - assert isinstance(authkey, bytes) + if not isinstance(authkey, bytes): + raise ValueError( + "Authkey must be bytes, not {0!s}".format(type(authkey))) message = os.urandom(MESSAGE_LENGTH) connection.send_bytes(CHALLENGE + message) digest = hmac.new(authkey, message, 'md5').digest() @@ -733,7 +735,9 @@ def deliver_challenge(connection, authkey): def answer_challenge(connection, authkey): import hmac - assert isinstance(authkey, bytes) + if not isinstance(authkey, bytes): + raise ValueError( + "Authkey must be bytes, not {0!s}".format(type(authkey))) message = connection.recv_bytes(256) # reject large message assert message[:len(CHALLENGE)] == CHALLENGE, 'message = %r' % message message = message[len(CHALLENGE):] diff --git a/Lib/multiprocessing/dummy/__init__.py b/Lib/multiprocessing/dummy/__init__.py index cbb7f4909cb..403f5e5198e 100644 --- a/Lib/multiprocessing/dummy/__init__.py +++ b/Lib/multiprocessing/dummy/__init__.py @@ -41,7 +41,10 @@ def __init__(self, group=None, target=None, name=None, args=(), kwargs={}): self._parent = current_process() def start(self): - assert self._parent is current_process() + if self._parent is not current_process(): + raise RuntimeError( + "Parent is {0!r} but current_process is {1!r}".format( + self._parent, current_process())) self._start_called = True if hasattr(self._parent, '_children'): self._parent._children[self] = None diff --git a/Lib/multiprocessing/forkserver.py b/Lib/multiprocessing/forkserver.py index 69b842aa939..7a952e2ba69 100644 --- a/Lib/multiprocessing/forkserver.py +++ b/Lib/multiprocessing/forkserver.py @@ -189,7 +189,7 @@ def sigchld_handler(*_unused): if alive_r in rfds: # EOF because no more client processes left - assert os.read(alive_r, 1) == b'' + assert os.read(alive_r, 1) == b'', "Not at EOF?" raise SystemExit if sig_r in rfds: @@ -208,7 +208,10 @@ def sigchld_handler(*_unused): if os.WIFSIGNALED(sts): returncode = -os.WTERMSIG(sts) else: - assert os.WIFEXITED(sts) + if not os.WIFEXITED(sts): + raise AssertionError( + "Child {0:n} status is {1:n}".format( + pid,sts)) returncode = os.WEXITSTATUS(sts) # Send exit code to client process try: @@ -227,7 +230,10 @@ def sigchld_handler(*_unused): with listener.accept()[0] as s: # Receive fds from client fds = reduction.recvfds(s, MAXFDS_TO_SEND + 1) - assert len(fds) <= MAXFDS_TO_SEND + if len(fds) > MAXFDS_TO_SEND: + raise RuntimeError( + "Too many ({0:n}) fds to send".format( + len(fds))) child_r, child_w, *fds = fds s.close() pid = os.fork() diff --git a/Lib/multiprocessing/heap.py b/Lib/multiprocessing/heap.py index ee3ed551d0c..566173a1b0a 100644 --- a/Lib/multiprocessing/heap.py +++ b/Lib/multiprocessing/heap.py @@ -211,7 +211,10 @@ def free(self, block): # synchronously sometimes later from malloc() or free(), by calling # _free_pending_blocks() (appending and retrieving from a list is not # strictly thread-safe but under cPython it's atomic thanks to the GIL). - assert os.getpid() == self._lastpid + if os.getpid() != self._lastpid: + raise ValueError( + "My pid ({0:n}) is not last pid {1:n}".format( + os.getpid(),self._lastpid)) if not self._lock.acquire(False): # can't acquire the lock right now, add the block to the list of # pending blocks to free @@ -227,7 +230,10 @@ def free(self, block): def malloc(self, size): # return a block of right size (possibly rounded up) - assert 0 <= size < sys.maxsize + if size < 0: + raise ValueError("Size {0:n} out of range".format(size)) + if sys.maxsize <= size: + raise OverflowError("Size {0:n} too large".format(size)) if os.getpid() != self._lastpid: self.__init__() # reinitialize after fork with self._lock: @@ -250,7 +256,10 @@ class BufferWrapper(object): _heap = Heap() def __init__(self, size): - assert 0 <= size < sys.maxsize + if size < 0: + raise ValueError("Size {0:n} out of range".format(size)) + if sys.maxsize <= size: + raise OverflowError("Size {0:n} too large".format(size)) block = BufferWrapper._heap.malloc(size) self._state = (block, size) util.Finalize(self, BufferWrapper._heap.free, args=(block,)) diff --git a/Lib/multiprocessing/managers.py b/Lib/multiprocessing/managers.py index c6722771b05..04df26bac66 100644 --- a/Lib/multiprocessing/managers.py +++ b/Lib/multiprocessing/managers.py @@ -23,7 +23,7 @@ from traceback import format_exc from . import connection -from .context import reduction, get_spawning_popen +from .context import reduction, get_spawning_popen, ProcessError from . import pool from . import process from . import util @@ -133,7 +133,10 @@ class Server(object): 'debug_info', 'number_of_objects', 'dummy', 'incref', 'decref'] def __init__(self, registry, address, authkey, serializer): - assert isinstance(authkey, bytes) + if not isinstance(authkey, bytes): + raise TypeError( + "Authkey {0!r} is type {1!s}, not bytes".format( + authkey, type(authkey))) self.registry = registry self.authkey = process.AuthenticationString(authkey) Listener, Client = listener_client[serializer] @@ -163,7 +166,7 @@ def serve_forever(self): except (KeyboardInterrupt, SystemExit): pass finally: - if sys.stdout != sys.__stdout__: + if sys.stdout != sys.__stdout__: # what about stderr? util.debug('resetting stdout, stderr') sys.stdout = sys.__stdout__ sys.stderr = sys.__stderr__ @@ -316,6 +319,7 @@ def debug_info(self, c): ''' Return some info --- useful to spot problems with refcounting ''' + # Perhaps include debug info about 'c'? with self.mutex: result = [] keys = list(self.id_to_refcount.keys()) @@ -356,7 +360,9 @@ def create(self, c, typeid, *args, **kwds): self.registry[typeid] if callable is None: - assert len(args) == 1 and not kwds + if kwds or (len(args) != 1): + raise ValueError( + "Without callable, must have one non-keyword argument") obj = args[0] else: obj = callable(*args, **kwds) @@ -364,7 +370,10 @@ def create(self, c, typeid, *args, **kwds): if exposed is None: exposed = public_methods(obj) if method_to_typeid is not None: - assert type(method_to_typeid) is dict + if not isinstance(method_to_typeid, dict): + raise TypeError( + "Method_to_typeid {0!r}: type {1!s}, not dict".format( + method_to_typeid, type(method_to_typeid))) exposed = list(exposed) + list(method_to_typeid) ident = '%x' % id(obj) # convert to string because xmlrpclib @@ -417,7 +426,11 @@ def decref(self, c, ident): return with self.mutex: - assert self.id_to_refcount[ident] >= 1 + if self.id_to_refcount[ident] <= 0: + raise AssertionError( + "Id {0!s} ({1!r}) has refcount {2:n}, not 1+".format( + ident, self.id_to_obj[ident], + self.id_to_refcount[ident])) self.id_to_refcount[ident] -= 1 if self.id_to_refcount[ident] == 0: del self.id_to_refcount[ident] @@ -480,7 +493,14 @@ def get_server(self): ''' Return server object with serve_forever() method and address attribute ''' - assert self._state.value == State.INITIAL + if self._state.value != State.INITIAL: + if self._state.value == State.STARTED: + raise ProcessError("Already started server") + elif self._state.value == State.SHUTDOWN: + raise ProcessError("Manager has shut down") + else: + raise ProcessError( + "Unknown state {!r}".format(self._state.value)) return Server(self._registry, self._address, self._authkey, self._serializer) @@ -497,7 +517,14 @@ def start(self, initializer=None, initargs=()): ''' Spawn a server process for this manager object ''' - assert self._state.value == State.INITIAL + if self._state.value != State.INITIAL: + if self._state.value == State.STARTED: + raise ProcessError("Already started server") + elif self._state.value == State.SHUTDOWN: + raise ProcessError("Manager has shut down") + else: + raise ProcessError( + "Unknown state {!r}".format(self._state.value)) if initializer is not None and not callable(initializer): raise TypeError('initializer must be a callable') @@ -593,7 +620,14 @@ def _number_of_objects(self): def __enter__(self): if self._state.value == State.INITIAL: self.start() - assert self._state.value == State.STARTED + if self._state.value != State.STARTED: + if self._state.value == State.INITIAL: + raise ProcessError("Unable to start server") + elif self._state.value == State.SHUTDOWN: + raise ProcessError("Manager has shut down") + else: + raise ProcessError( + "Unknown state {!r}".format(self._state.value)) return self def __exit__(self, exc_type, exc_val, exc_tb): @@ -653,7 +687,7 @@ def register(cls, typeid, callable=None, proxytype=None, exposed=None, getattr(proxytype, '_method_to_typeid_', None) if method_to_typeid: - for key, value in list(method_to_typeid.items()): + for key, value in list(method_to_typeid.items()): # isinstance? assert type(key) is str, '%r is not a string' % key assert type(value) is str, '%r is not a string' % value diff --git a/Lib/multiprocessing/pool.py b/Lib/multiprocessing/pool.py index c2364ab186c..e457f0a576f 100644 --- a/Lib/multiprocessing/pool.py +++ b/Lib/multiprocessing/pool.py @@ -92,7 +92,9 @@ def __repr__(self): def worker(inqueue, outqueue, initializer=None, initargs=(), maxtasks=None, wrap_exception=False): - assert maxtasks is None or (type(maxtasks) == int and maxtasks > 0) + if (maxtasks is not None) and not (isinstance(maxtasks, int) + and maxtasks >= 1): + raise AssertionError("Maxtasks {!r} is not valid".format(maxtasks)) put = outqueue.put get = inqueue.get if hasattr(inqueue, '_writer'): @@ -254,8 +256,8 @@ def _setup_queues(self): def apply(self, func, args=(), kwds={}): ''' Equivalent of `func(*args, **kwds)`. + Pool must be running. ''' - assert self._state == RUN return self.apply_async(func, args, kwds).get() def map(self, func, iterable, chunksize=None): @@ -307,6 +309,10 @@ def imap(self, func, iterable, chunksize=1): )) return result else: + if chunksize < 1: + raise ValueError( + "Chunksize must be 1+, not {0:n}".format( + chunksize)) assert chunksize > 1 task_batches = Pool._get_tasks(func, iterable, chunksize) result = IMapIterator(self._cache) @@ -334,7 +340,9 @@ def imap_unordered(self, func, iterable, chunksize=1): )) return result else: - assert chunksize > 1 + if chunksize < 1: + raise ValueError( + "Chunksize must be 1+, not {0!r}".format(chunksize)) task_batches = Pool._get_tasks(func, iterable, chunksize) result = IMapUnorderedIterator(self._cache) self._taskqueue.put( @@ -466,7 +474,7 @@ def _handle_results(outqueue, get, cache): return if thread._state: - assert thread._state == TERMINATE + assert thread._state == TERMINATE, "Thread not in TERMINATE" util.debug('result handler found thread._state=TERMINATE') break @@ -542,7 +550,10 @@ def terminate(self): def join(self): util.debug('joining pool') - assert self._state in (CLOSE, TERMINATE) + if self._state == RUN: + raise ValueError("Pool is still running") + elif self._state not in (CLOSE, TERMINATE): + raise ValueError("In unknown state") self._worker_handler.join() self._task_handler.join() self._result_handler.join() @@ -570,7 +581,9 @@ def _terminate_pool(cls, taskqueue, inqueue, outqueue, pool, util.debug('helping task handler/workers to finish') cls._help_stuff_finish(inqueue, task_handler, len(pool)) - assert result_handler.is_alive() or len(cache) == 0 + if (not result_handler.is_alive()) and (len(cache) != 0): + raise AssertionError( + "Cannot have cache with result_hander not alive") result_handler._state = TERMINATE outqueue.put(None) # sentinel @@ -628,7 +641,8 @@ def ready(self): return self._event.is_set() def successful(self): - assert self.ready() + if not self.ready(): + raise ValueError("{0!r} not ready".format(self)) return self._success def wait(self, timeout=None): diff --git a/Lib/multiprocessing/popen_fork.py b/Lib/multiprocessing/popen_fork.py index 44ce9a9e79b..cbdbfa79cf3 100644 --- a/Lib/multiprocessing/popen_fork.py +++ b/Lib/multiprocessing/popen_fork.py @@ -35,7 +35,7 @@ def poll(self, flag=os.WNOHANG): if os.WIFSIGNALED(sts): self.returncode = -os.WTERMSIG(sts) else: - assert os.WIFEXITED(sts) + assert os.WIFEXITED(sts), "Status is {:n}".format(sts) self.returncode = os.WEXITSTATUS(sts) return self.returncode diff --git a/Lib/multiprocessing/queues.py b/Lib/multiprocessing/queues.py index 513807cafec..328efbd95fe 100644 --- a/Lib/multiprocessing/queues.py +++ b/Lib/multiprocessing/queues.py @@ -78,7 +78,7 @@ def _after_fork(self): self._poll = self._reader.poll def put(self, obj, block=True, timeout=None): - assert not self._closed + assert not self._closed, "Queue {0!r} has been closed".format(self) if not self._sem.acquire(block, timeout): raise Full @@ -140,7 +140,7 @@ def close(self): def join_thread(self): debug('Queue.join_thread()') - assert self._closed + assert self._closed, "Queue {0!r} not closed".format(self) if self._jointhread: self._jointhread() @@ -281,7 +281,7 @@ def __setstate__(self, state): self._cond, self._unfinished_tasks = state[-2:] def put(self, obj, block=True, timeout=None): - assert not self._closed + assert not self._closed, "Queue {0!r} is closed".format(self) if not self._sem.acquire(block, timeout): raise Full diff --git a/Lib/multiprocessing/reduction.py b/Lib/multiprocessing/reduction.py index 7f65947379f..deca19ccad7 100644 --- a/Lib/multiprocessing/reduction.py +++ b/Lib/multiprocessing/reduction.py @@ -165,7 +165,10 @@ def recvfds(sock, size): if len(cmsg_data) % a.itemsize != 0: raise ValueError a.frombytes(cmsg_data) - assert len(a) % 256 == msg[0] + if len(a) % 256 != msg[0]: + raise AssertionError( + "Len is {0:n} but msg[0] is {1!r}".format( + len(a), msg[0])) return list(a) except (ValueError, IndexError): pass diff --git a/Lib/multiprocessing/resource_sharer.py b/Lib/multiprocessing/resource_sharer.py index e44a728fa9a..6d99da102ff 100644 --- a/Lib/multiprocessing/resource_sharer.py +++ b/Lib/multiprocessing/resource_sharer.py @@ -125,7 +125,7 @@ def _afterfork(self): def _start(self): from .connection import Listener - assert self._listener is None + assert self._listener is None, "Already have Listener" util.debug('starting listener and thread for sending handles') self._listener = Listener(authkey=process.current_process().authkey) self._address = self._listener.address diff --git a/Lib/multiprocessing/semaphore_tracker.py b/Lib/multiprocessing/semaphore_tracker.py index de7738eeee8..d5f259c246b 100644 --- a/Lib/multiprocessing/semaphore_tracker.py +++ b/Lib/multiprocessing/semaphore_tracker.py @@ -80,7 +80,8 @@ def _send(self, cmd, name): # bytes are atomic, and that PIPE_BUF >= 512 raise ValueError('name too long') nbytes = os.write(self._fd, msg) - assert nbytes == len(msg) + assert nbytes == len(msg), "nbytes {0:n} but len(msg) {1:n}".format( + nbytes, len(msg)) _semaphore_tracker = SemaphoreTracker() diff --git a/Lib/multiprocessing/spawn.py b/Lib/multiprocessing/spawn.py index 4aba372e48e..1f4f3f496f5 100644 --- a/Lib/multiprocessing/spawn.py +++ b/Lib/multiprocessing/spawn.py @@ -93,7 +93,7 @@ def spawn_main(pipe_handle, parent_pid=None, tracker_fd=None): ''' Run code specified by data received over pipe ''' - assert is_forking(sys.argv) + assert is_forking(sys.argv), "Not forking" if sys.platform == 'win32': import msvcrt new_handle = reduction.steal_handle(parent_pid, pipe_handle) diff --git a/Lib/multiprocessing/synchronize.py b/Lib/multiprocessing/synchronize.py index 0590ed68f53..038f73f6b76 100644 --- a/Lib/multiprocessing/synchronize.py +++ b/Lib/multiprocessing/synchronize.py @@ -270,13 +270,16 @@ def wait(self, timeout=None): def notify(self, n=1): assert self._lock._semlock._is_mine(), 'lock is not owned' - assert not self._wait_semaphore.acquire(False) + assert not self._wait_semaphore.acquire( + False), ('notify: Should not have been able to acquire' + + '_wait_semaphore') # to take account of timeouts since last notify*() we subtract # woken_count from sleeping_count and rezero woken_count while self._woken_count.acquire(False): res = self._sleeping_count.acquire(False) - assert res + assert res, ('notify: Bug in sleeping_count.acquire' + + '- res should not be False') sleepers = 0 while sleepers < n and self._sleeping_count.acquire(False): diff --git a/Lib/multiprocessing/util.py b/Lib/multiprocessing/util.py index b490caa7e64..f0827f0a969 100644 --- a/Lib/multiprocessing/util.py +++ b/Lib/multiprocessing/util.py @@ -149,12 +149,15 @@ class Finalize(object): Class which supports object finalization using weakrefs ''' def __init__(self, obj, callback, args=(), kwargs=None, exitpriority=None): - assert exitpriority is None or type(exitpriority) is int + if (exitpriority is not None) and not isinstance(exitpriority,int): + raise TypeError( + "Exitpriority ({0!r}) must be None or int, not {1!s}".format( + exitpriority, type(exitpriority))) if obj is not None: self._weakref = weakref.ref(obj, self) - else: - assert exitpriority is not None + elif exitpriority is None: + raise ValueError("Without object, exitpriority cannot be None") self._callback = callback self._args = args diff --git a/Misc/NEWS.d/next/Library/2017-08-18-17-16-38.bpo-5001.gwnthq.rst b/Misc/NEWS.d/next/Library/2017-08-18-17-16-38.bpo-5001.gwnthq.rst new file mode 100644 index 00000000000..640b801c9b3 --- /dev/null +++ b/Misc/NEWS.d/next/Library/2017-08-18-17-16-38.bpo-5001.gwnthq.rst @@ -0,0 +1 @@ +Many asserts in `multiprocessing` are now more informative, and some error types have been changed to more specific ones.
participants (1)
-
Antoine Pitrou