1 files changed, 72 insertions, 46 deletions

diff --git a/bitbake/lib/bb/asyncrpc/serv.py b/bitbake/lib/bb/asyncrpc/serv.py
index 4084f300df..45628698b6 100644
--- a/bitbake/lib/bb/asyncrpc/serv.py
+++ b/bitbake/lib/bb/asyncrpc/serv.py
@@ -131,53 +131,58 @@ class AsyncServerConnection(object):
 
 
 class AsyncServer(object):
-    def __init__(self, logger, loop=None):
-        if loop is None:
-            self.loop = asyncio.new_event_loop()
-            self.close_loop = True
-        else:
-            self.loop = loop
-            self.close_loop = False
-
+    def __init__(self, logger):
         self._cleanup_socket = None
         self.logger = logger
+        self.start = None
+        self.address = None
+
+    @property
+    def loop(self):
+        return asyncio.get_event_loop()
 
     def start_tcp_server(self, host, port):
-        self.server = self.loop.run_until_complete(
-            asyncio.start_server(self.handle_client, host, port, loop=self.loop)
-        )
-
-        for s in self.server.sockets:
-            self.logger.debug('Listening on %r' % (s.getsockname(),))
-            # Newer python does this automatically. Do it manually here for
-            # maximum compatibility
-            s.setsockopt(socket.SOL_TCP, socket.TCP_NODELAY, 1)
-            s.setsockopt(socket.SOL_TCP, socket.TCP_QUICKACK, 1)
-
-        name = self.server.sockets[0].getsockname()
-        if self.server.sockets[0].family == socket.AF_INET6:
-            self.address = "[%s]:%d" % (name[0], name[1])
-        else:
-            self.address = "%s:%d" % (name[0], name[1])
+        def start_tcp():
+            self.server = self.loop.run_until_complete(
+                asyncio.start_server(self.handle_client, host, port)
+            )
+
+            for s in self.server.sockets:
+                self.logger.debug('Listening on %r' % (s.getsockname(),))
+                # Newer python does this automatically. Do it manually here for
+                # maximum compatibility
+                s.setsockopt(socket.SOL_TCP, socket.TCP_NODELAY, 1)
+                s.setsockopt(socket.SOL_TCP, socket.TCP_QUICKACK, 1)
+
+            name = self.server.sockets[0].getsockname()
+            if self.server.sockets[0].family == socket.AF_INET6:
+                self.address = "[%s]:%d" % (name[0], name[1])
+            else:
+                self.address = "%s:%d" % (name[0], name[1])
+
+        self.start = start_tcp
 
     def start_unix_server(self, path):
         def cleanup():
             os.unlink(path)
 
-        cwd = os.getcwd()
-        try:
-            # Work around path length limits in AF_UNIX
-            os.chdir(os.path.dirname(path))
-            self.server = self.loop.run_until_complete(
-                asyncio.start_unix_server(self.handle_client, os.path.basename(path), loop=self.loop)
-            )
-        finally:
-            os.chdir(cwd)
+        def start_unix():
+            cwd = os.getcwd()
+            try:
+                # Work around path length limits in AF_UNIX
+                os.chdir(os.path.dirname(path))
+                self.server = self.loop.run_until_complete(
+                    asyncio.start_unix_server(self.handle_client, os.path.basename(path))
+                )
+            finally:
+                os.chdir(cwd)
 
-        self.logger.debug('Listening on %r' % path)
+            self.logger.debug('Listening on %r' % path)
 
-        self._cleanup_socket = cleanup
-        self.address = "unix://%s" % os.path.abspath(path)
+            self._cleanup_socket = cleanup
+            self.address = "unix://%s" % os.path.abspath(path)
+
+        self.start = start_unix
 
     @abc.abstractmethod
     def accept_client(self, reader, writer):
@@ -205,8 +210,7 @@ class AsyncServer(object):
         self.logger.debug("Got exit signal")
         self.loop.stop()
 
-    def serve_forever(self):
-        asyncio.set_event_loop(self.loop)
+    def _serve_forever(self):
         try:
             self.loop.add_signal_handler(signal.SIGTERM, self.signal_handler)
             signal.pthread_sigmask(signal.SIG_UNBLOCK, [signal.SIGTERM])
@@ -217,28 +221,50 @@ class AsyncServer(object):
             self.loop.run_until_complete(self.server.wait_closed())
             self.logger.debug('Server shutting down')
         finally:
-            if self.close_loop:
-                if sys.version_info >= (3, 6):
-                    self.loop.run_until_complete(self.loop.shutdown_asyncgens())
-                self.loop.close()
-
             if self._cleanup_socket is not None:
                 self._cleanup_socket()
 
+    def serve_forever(self):
+        """
+        Serve requests in the current process
+        """
+        self.start()
+        self._serve_forever()
+
     def serve_as_process(self, *, prefunc=None, args=()):
-        def run():
+        """
+        Serve requests in a child process
+        """
+        def run(queue):
+            try:
+                self.start()
+            finally:
+                queue.put(self.address)
+                queue.close()
+
             if prefunc is not None:
                 prefunc(self, *args)
-            self.serve_forever()
+
+            self._serve_forever()
+
+            if sys.version_info >= (3, 6):
+                self.loop.run_until_complete(self.loop.shutdown_asyncgens())
+            self.loop.close()
+
+        queue = multiprocessing.Queue()
 
         # Temporarily block SIGTERM. The server process will inherit this
         # block which will ensure it doesn't receive the SIGTERM until the
         # handler is ready for it
         mask = signal.pthread_sigmask(signal.SIG_BLOCK, [signal.SIGTERM])
         try:
-            self.process = multiprocessing.Process(target=run)
+            self.process = multiprocessing.Process(target=run, args=(queue,))
             self.process.start()
 
+            self.address = queue.get()
+            queue.close()
+            queue.join_thread()
+
             return self.process
         finally:
             signal.pthread_sigmask(signal.SIG_SETMASK, mask)