Method test_rapid_restart

Lib/test/_test_multiprocessing.py:3587–3626

(self)

    @warnings_helper.ignore_fork_in_thread_deprecation_warnings()
    @support.skip_if_sanitizer("TSan: leaks threads", thread=True)
    def test_rapid_restart(self):
        authkey = os.urandom(32)
        manager = QueueManager(
            address=(socket_helper.HOST, 0), authkey=authkey,
            serializer=SERIALIZER, shutdown_timeout=SHUTDOWN_TIMEOUT)
        try:
            srvr = manager.get_server()
            addr = srvr.address
            # Close the connection.Listener socket which gets opened as a part
            # of manager.get_server(). It's not needed for the test.
            srvr.listener.close()
            manager.start()

            p = self.Process(target=self._putter, args=(manager.address, authkey))
            p.start()
            p.join()
            queue = manager.get_queue()
            self.assertEqual(queue.get(), 'hello world')
            del queue
        finally:
            if hasattr(manager, "shutdown"):
                manager.shutdown()

        manager = QueueManager(
            address=addr, authkey=authkey, serializer=SERIALIZER,
            shutdown_timeout=SHUTDOWN_TIMEOUT)
        try:
            manager.start()
            self.addCleanup(manager.shutdown)
        except OSError as e:
            if e.errno != errno.EADDRINUSE:
                raise
            # Retry after some time, in case the old socket was lingering
            # (sporadic failure on buildbots)
            time.sleep(1.0)
            manager = QueueManager(
                address=addr, authkey=authkey, serializer=SERIALIZER,
                shutdown_timeout=SHUTDOWN_TIMEOUT)
            if hasattr(manager, "shutdown"):
                self.addCleanup(manager.shutdown)


class FakeConnection: