Merge lp:~mvo/click/acquire-asyncio into lp:click/devel

[Michael Vogt]

Use asyncio for the method communication.

Simplify the non-blocking code in the subprocess handling by using the new asyncio. This means we will be python3+ only (unless we use the python-trollius compat module for python2 and older python3).

[PS Jenkins bot (ps-jenkins)]

FAILED: Continuous integration, rev:550
No commit message was specified in the merge proposal. Click on the following link and set the commit message (if you want a jenkins rebuild you need to trigger it yourself):
https://code.launchpad.net/~mvo/click/acquire-asyncio/+merge/236499/+edit-commit-message

http://jenkins.qa.ubuntu.com/job/click-devel-ci/89/
Executed test runs:
    SUCCESS: http://jenkins.qa.ubuntu.com/job/click-devel-utopic-amd64-ci/91
    SUCCESS: http://jenkins.qa.ubuntu.com/job/click-devel-utopic-armhf-ci/89
        deb: http://jenkins.qa.ubuntu.com/job/click-devel-utopic-armhf-ci/89/artifact/work/output/*zip*/output.zip
    SUCCESS: http://jenkins.qa.ubuntu.com/job/click-devel-utopic-i386-ci/89

Click here to trigger a rebuild:
http://s-jenkins.ubuntu-ci:8080/job/click-devel-ci/89/rebuild

review: Needs Fixing (continuous-integration)

Unmerged revisions

550. By Michael Vogt on 2014-09-30

refactor so that AcquireMethod.run() also uses asyncio

549. By Michael Vogt on 2014-09-30

port to asyncio

548. By Michael Vogt on 2014-09-30

add missing python3-dbus build-dependency

547. By Michael Vogt on 2014-09-30

rename read_messages() -> _read_messages()

546. By Michael Vogt on 2014-09-30

click/acquire.py: add note about ssl options for curl

545. By Michael Vogt on 2014-09-30

first round of addressing review points from Colin

544. By Michael Vogt on 2014-09-25

add python3-pycurl to b-d (needed during the tests

543. By Michael Vogt on 2014-09-24

add missing python3-pycurl depdendency

542. By Michael Vogt on 2014-09-24

click/acquire.py: make dbus optional

541. By Michael Vogt on 2014-09-24

remove debug message

=== modified file 'click/acquire.py'
--- click/acquire.py	2014-09-30 12:27:37 +0000
+++ click/acquire.py	2014-09-30 12:27:37 +0000
@@ -28,8 +28,8 @@
     'ClickAcquireStatusText',
     ]
 
+import asyncio
 import os
-import select
 import subprocess
 import sys
 from textwrap import dedent
@@ -49,47 +49,32 @@
     pass
 
 
-def extract_one_message(lines):
+# hrm, why is this not in the asyncio directly?
+def get_asyncio_stream_reader_from_file(loop, file_like):
+    """Return (yield from) a StreamReader for the given file like object"""
+    stream_reader = asyncio.streams.StreamReader(loop=loop)
+    transport, protocol = yield from loop.connect_read_pipe(
+        lambda: asyncio.streams.StreamReaderProtocol(stream_reader),
+        file_like)
+    return stream_reader
+
+
[e-mail]
+def _read_message(input_file):
     msg = []
     while True:
-        try:
-            line = lines.pop(0)
-        except IndexError:
+        raw_line = yield from input_file.readline()
+        line = raw_line.decode("utf-8")
+        # consume all top "\n"
+        if len(msg) == 0 and line == "\n":
+            continue
+        if line == "\n":
             break
-        if os.environ.get("CLICK_DEBUG_ACQUIRE", ""):
-            sys.stderr.write("[%s] raw_line: '%s'\n" % (
-                os.getpid(), line.replace("\n", "\\n")))
         msg.append(line)
-        if len(msg) > 0 and line == "":
-            # we are done, collect all remaining "\n" and stop
-            while len(lines) > 0 and lines[0] == "":
-                lines.pop(0)
-            break
     if len(msg) < 2:
         return -1 , ""
-    if os.environ.get("CLICK_DEBUG_ACQUIRE", ""):
-        sys.stderr.write("[%s] msg: '%s'\n" % (os.getpid(), str(msg)))
     number = int(msg[0].split()[0])
-    return number, "\n".join(msg)
-
-
-def _read_messages(input_file, timeout=None):
-    infd = input_file.fileno()
-    msgs = []
-    rl, wl, xl = select.select([infd], [], [], timeout)
-    if rl:
-        # FIXME: 16k message limit is arbitrary
-        buf = os.read(infd, 16*1024).decode("utf-8")
-        if os.environ.get("CLICK_DEBUG_ACQUIRE", ""):
-            sys.stderr.write("[%s] read buf: '%s'\n" % (
-                os.getpid(), buf))
-        lines = buf.split("\n")
-        while True:
-            number, msg = extract_one_message(lines)
-            if number < 0:
-                break
-            msgs.append( (number, msg) )
-    return msgs
+    return number, "".join(msg)
 
 
 class ClickAcquireStatus:
@@ -149,44 +134,49 @@
         Filename: {destfile}
 
         """).format(uri=uri, destfile=destfile)
-        pipe.write(cmd)
-        pipe.flush()
+        pipe.write(cmd.encode("utf-8"))
+        yield from pipe.drain()
 
     def _redirect(self, pipe, new_uri, destfile):
         self._uri_acquire(pipe, new_uri, destfile)
 
+    @asyncio.coroutine
+    def _communicate_with_method(self, cmd, uri, destfile):
+        p = yield from asyncio.create_subprocess_exec(cmd,
+            stdout=subprocess.PIPE, stdin=subprocess.PIPE)
+        while True:
+            number, raw_message = yield from _read_message(p.stdout)
+            message = Deb822(raw_message)
+            if number == self.M_CAPABILITIES:
+                yield from self._uri_acquire(p.stdin, uri, destfile)
+            elif number == self.M_STATUS:
+                pass
+            elif number == self.M_REDIRECT:
+                self._redirect(p.stdin, message.get("New-URI"), destfile)
+            elif number == self.URI_START:
+                self._status.uri = message.get("URI", 0)
+                self._status.total_bytes = int(message.get("Size", 0))
+            elif number == self.URI_SUCCESS:
+                self._status.fetched_bytes = int(message.get("Size"))
+                self._status.done()
+                return True
+            elif number == self.URI_FAILURE:
+                raise ClickAcquireError()
+
+    def _report_progress(self, loop, destfile):
+        if os.path.exists(destfile):
+            self._status.fetched_bytes = os.path.getsize(destfile)
+            self._status.pulse()
+        loop.call_later(self.TIMEOUT, self._report_progress, loop, destfile)
+
     def _run_acquire_method(self, uri, destfile):
         parsed_uri = urlparse(uri)
+        loop = asyncio.get_event_loop()
         cmd = os.path.join(acquire_methods_dir, parsed_uri.scheme)
-        p = subprocess.Popen([cmd],
-            stdout=subprocess.PIPE, stdin=subprocess.PIPE,
-            universal_newlines=True)
-        while True:
-            for number, raw_message in _read_messages(p.stdout, self.TIMEOUT):
-                message = Deb822(raw_message)
-                if number == self.M_CAPABILITIES:
-                    self._uri_acquire(p.stdin, uri, destfile)
-                elif number == self.M_STATUS:
-                    pass
-                elif number == self.M_REDIRECT:
-                    self._redirect(p.stdin, message.get("New-URI"), destfile)
-                elif number == self.URI_START:
-                    self._status.uri = message.get("URI", 0)
-                    self._status.total_bytes = int(message.get("Size", 0))
-                elif number == self.URI_SUCCESS:
-                    self._status.fetched_bytes = int(message.get("Size"))
-                    self._status.done()
-                    p.stdout.close()
-                    p.stdin.close()
-                    return True
-                elif number == self.URI_FAILURE:
-                    p.stdout.close()
-                    p.stdin.close()
-                    raise ClickAcquireError()
-            # update progress
-            if os.path.exists(destfile):
-                self._status.fetched_bytes = os.path.getsize(destfile)
-                self._status.pulse()
+        loop.call_later(self.TIMEOUT, self._report_progress, loop, destfile)
+        loop.run_until_complete(
+            self._communicate_with_method(cmd, uri, destfile))
+        loop.close()
         return False
 
     def fetch(self, uri, destfile):
@@ -212,16 +202,20 @@
         sys.stdout.flush()
 
     def run(self):
+        loop = asyncio.get_event_loop()
+        loop.run_until_complete(self._run(loop))
+
+    @asyncio.coroutine
+    def _run(self, loop):
+        stdin_reader = yield from get_asyncio_stream_reader_from_file(
+            loop, sys.stdin)
         while True:
-            msgs = _read_messages(sys.stdin)
-            if not msgs:
-                break
-            for number, raw_message in msgs:
-                message = Deb822(raw_message)
-                if number == self.M_CONFIGURATION:
-                    pass
-                elif number == self.M_FETCH:
-                    self.fetch(message.get("URI"), message.get("FileName"))
+            number, raw_message = yield from _read_message(stdin_reader)
+            message = Deb822(raw_message)
+            if number == self.M_CONFIGURATION:
+                pass
+            elif number == self.M_FETCH:
+                self.fetch(message.get("URI"), message.get("FileName"))
 
     def uri_start(self, uri, filename, size):
         # note that apt itself does not use "Filename" here because it

=== modified file 'click/tests/test_acquire.py'
--- click/tests/test_acquire.py	2014-09-30 12:27:37 +0000
+++ click/tests/test_acquire.py	2014-09-30 12:27:37 +0000
@@ -24,7 +24,6 @@
 
 import multiprocessing
 import os.path
-import time
 
 from six.moves import (
     SimpleHTTPServer,
@@ -34,7 +33,7 @@
 from click.acquire import (
     ClickAcquire,
     ClickAcquireError,
-    _read_messages,
+    _read_message,
 )
 
 from click.tests.helpers import (
@@ -108,7 +107,7 @@
         self.assertEqual(canary_str, data)
 
 
-class TestClickAcquireReadMessages(TestCase):
+class TestClickAcquireReadMessage(TestCase):
 
     def test_forked_read_message(self):
         read_end, write_end = os.pipe()
@@ -116,7 +115,7 @@
         if pid == 0:
             os.close(write_end)
             with os.fdopen(read_end) as f:
-                number, msg = _read_messages(f)[0]
+                number, msg = yield from _read_message(f)
                 self.assertEqual(msg, "102 Status\nFoo: Bar\n")
                 self.assertEqual(number, 102)
             os._exit(0)
@@ -131,7 +130,9 @@
         if pid == 0:
             os.close(write_end)
             with os.fdopen(read_end) as f:
-                msgs = _read_messages(f)
+                msg1 = yield from _read_message(f)
+                msg2 = yield from _read_message(f)
+                msgs = [msg1, msg2]
                 self.assertEqual(len(msgs), 2)
                 self.assertEqual(msgs[0][0], 100)
                 self.assertEqual(msgs[1][0], 200)
@@ -147,28 +148,10 @@
         if pid == 0:
             os.close(write_end)
             with os.fdopen(read_end) as f:
-                self.assertEqual(_read_messages(f), [])
+                msg = yield from _read_message(f)
+                self.assertEqual(msg, (-1, []))
             os._exit(0)
         os.close(read_end)
         os.write(write_end, "102 close-before-msg".encode("utf-8"))
         os.close(write_end)
         os.waitpid(pid, 0)
-
-    def test_forked_read_message_with_timeout(self):
-        read_end, write_end = os.pipe()
-        pid = os.fork()
-        if pid == 0:
-            os.close(write_end)
-            with os.fdopen(read_end) as f:
-                self.assertEqual(_read_messages(f, 0.1), [])
-                msgs = _read_messages(f)
-                self.assertEqual(len(msgs), 2)
-                self.assertEqual(msgs[0][1], "100 aaa\n")
-                self.assertEqual(msgs[1][1], "200 bbb\n")
-            os._exit(0)
-        os.close(read_end)
-        # simulate slow method
-        time.sleep(0.2)
-        os.write(write_end, "100 aaa\n\n".encode("utf-8"))
-        os.write(write_end, "200 bbb\n\n".encode("utf-8"))
-        os.waitpid(pid, 0)