WIP

Author: moreati

mitogen/core.py

@@ -2224,7 +2224,19 @@ def read(self, n=CHUNK_SIZE):
             # Refuse to touch the handle after closed, it may have been reused
             # by another thread. TODO: synchronize read()/write()/close().
             return b('')
-        s, disconnected = io_op(os.read, self.fd, n)
+
+        # FIXME What should this do with BlockingIOError?
+        try:
+            s, disconnected = io_op(os.read, self.fd, n)
+        except IOError:
+            exc = sys.exc_info()[1]
+            if exc.errno in (errno.EAGAIN, errno.EWOULDBLOCK):
+                LOG.warning(
+                    '%r.read() BlockingIOError %r fp=%r\n%s',
+                    self, exc, self.fp, traceback.format_exc(),
+                )
+            raise
+
         if disconnected:
             LOG.debug('%r: disconnected during read: %s', self, disconnected)
             return b('')
@@ -2244,7 +2256,19 @@ def write(self, s):
             # Don't touch the handle after close, it may be reused elsewhere.
             return None
 
-        written, disconnected = io_op(os.write, self.fd, s)
+        # FIXME Does BlockingIOError *always* mean 0 bytes written?
+        # FIXME Are callers handling a return value of 0?
+        try:
+            written, disconnected = io_op(os.write, self.fd, s)
+        except IOError:
+            exc = sys.exc_info()[1]
+            if exc.errno in (errno.EAGAIN, errno.EWOULDBLOCK):
+                LOG.warning(
+                    '%r.write() BlockingIOError %r fp=%r len(s)=%d\n%s',
+                    self, exc, self.fp, len(s), traceback.format_exc(),
+                )
+            return 0
+
         if disconnected:
             LOG.debug('%r: disconnected during write: %s', self, disconnected)
             return None
@@ -2933,7 +2957,16 @@ def put(self, obj=None):
             self._wake(wsock, cookie)
 
     def _wake(self, wsock, cookie):
-        written, disconnected = io_op(os.write, wsock.fileno(), cookie)
+        try:
+            written, disconnected = io_op(os.write, wsock.fileno(), cookie)
+        except IOError:
+            exc = sys.exc_info()[1]
+            if exc.errno in (errno.EAGAIN, errno.EWOULDBLOCK):
+                LOG.warning(
+                    '%r._wake() BlockingIOError %r wsock=%r len(cookie)=%d\n%s',
+                    self, exc, wsock, len(cookie), traceback.format_exc(),
+                )
+            raise
         assert written == len(cookie) and not disconnected
 
     def __repr__(self):

tests/ansible/regression/files/write_512mb_test_file

@@ -0,0 +1,36 @@
+# 288230376151711744 mitogen-test-file ABCDEFGHIJKLMNOPQRSTUVWXYZ
+
+import sys
+
+PADDING_POOL = b'ABCDEFGHIJKLMNOPQRSTUVWXYZ' * 2
+PADDING_SIZE = len(PADDING_POOL) // 2
+FILE_SIZE = 512 * 2**20
+LINE_SIZE = 64
+
+try:
+    xrange = range
+except NameError:
+    pass
+
+def format_line(lineno):
+    # type: (int) -> bytes
+    line_offset = lineno * LINE_SIZE
+    padding_offset = lineno % PADDING_SIZE
+    padding = PADDING_POOL[padding_offset:padding_offset + PADDING_SIZE]
+    return b'%018d mitogen-test-file %s\x0a' % (line_offset, padding)
+
+
+def main():
+    output_file = sys.argv[1]
+    written_bytes_count = 0
+    with open(output_file, 'wb') as f:
+        for lineno in xrange(0, FILE_SIZE // LINE_SIZE):
+            line = format_line(lineno)
+            f.write(line)
+            written_bytes_count += len(line)
+
+    print('Wrote %d bytes to %s' % (written_bytes_count, output_file))
+
+
+if __name__ == '__main__':
+    main()

tests/ansible/regression/issue_615__streaming_transfer.yml

@@ -7,30 +7,35 @@
   become: true
   vars:
     mitogen_ssh_compression: false
+    controller_file_path: "/tmp/fetch-{{ inventory_hostname }}-512mb.txt"
+    target_file_path: /tmp/mitogen-test-512mb.txt
   tasks:
     - include_tasks: _mitogen_only.yml
     - block:
-        - name: Create /tmp/512mb.zero
-          shell: |
-            dd if=/dev/zero of=/tmp/512mb.zero bs=1048576 count=512;
-            chmod go= /tmp/512mb.zero
-          args:
-            creates: /tmp/512mb.zero
+        - name: Create test file on target
+          script:
+            cmd: write_512mb_test_file "{{ target_file_path }}"
+            executable: "{{ ansible_python_interpreter | default(ansible_facts.discovered_interpreter_python) }}"
+            creates: "{{ target_file_path }}"
+          register: target_file_task
 
-        - name: Fetch /tmp/512mb.zero
+        - debug:
+            var: target_file_task
+
+        - name: Fetch test file
           fetch:
-            src: /tmp/512mb.zero
-            dest: /tmp/fetch-{{ inventory_hostname }}-512mb.zero
+            src: "{{ target_file_path }}"
+            dest: "{{ controller_file_path }}"
             flat: true
 
-        - name: Cleanup /tmp/512mb.zero
+        - name: Cleanup target
           file:
-            path: /tmp/512mb.zero
+            path: "{{ target_file_path }}"
             state: absent
 
-        - name: Cleanup fetched file
+        - name: Cleanup controller
           file:
-            path: /tmp/fetch-{{ inventory_hostname }}-512mb.zero
+            path: "{{ controller_file_path }}"
             state: absent
           become: false
           delegate_to: localhost