apveyor, #519: FIX incomplete Popen pump

ankostis · ankostis · commit 022ce82de4cc · 2016-09-26T02:20:59.000+02:00
+ Pump code reads only once streams per `_read_lines_from_fno()`
invocation, so when input larger than `mmap.PAGESIZE`, bytes are
forgotten in the stream.
diff --git a/git/cmd.py b/git/cmd.py
@@ -100,16 +100,16 @@ def _parse_lines_from_buffer(buf):
     # end
 
     def _read_lines_from_fno(fno, last_buf_list):
-        buf = os.read(fno, mmap.PAGESIZE)
-        buf = last_buf_list[0] + buf
+        for buf in iter(lambda: os.read(fno, mmap.PAGESIZE), b''):
+            buf = last_buf_list[0] + buf
 
-        bi = 0
-        for bi, line in _parse_lines_from_buffer(buf):
-            yield line
-        # for each line to parse from the buffer
+            bi = 0
+            for bi, line in _parse_lines_from_buffer(buf):
+                yield line
+            # for each line to parse from the buffer
 
-        # keep remainder
-        last_buf_list[0] = buf[bi:]
+            # keep remainder
+            last_buf_list[0] = buf[bi:]
 
     def _dispatch_single_line(line, handler):
         line = line.decode(defenc)
@@ -193,14 +193,24 @@ def _deplete_buffer(fno, handler, buf_list, wg=None):
     else:
         # Oh ... probably we are on windows. select.select() can only handle sockets, we have files
         # The only reliable way to do this now is to use threads and wait for both to finish
+        def _handle_lines(fd, handler, wg):
+            for line in fd:
+                line = line.decode(defenc)
+                if line and handler:
+                    handler(line)
+            if wg:
+                wg.done()
+
         # Since the finalizer is expected to wait, we don't have to introduce our own wait primitive
         # NO: It's not enough unfortunately, and we will have to sync the threads
         wg = WaitGroup()
-        for fno, (handler, buf_list) in fdmap.items():
+        for fd, handler in zip((process.stdout, process.stderr),
+                    (stdout_handler, stderr_handler)):
             wg.add(1)
-            t = threading.Thread(target=lambda: _deplete_buffer(fno, handler, buf_list, wg))
+            t = threading.Thread(target=_handle_lines, args=(fd, handler, wg))
+            t.setDaemon(True)
             t.start()
-        # end
+
         # NOTE: Just joining threads can possibly fail as there is a gap between .start() and when it's
         # actually started, which could make the wait() call to just return because the thread is not yet
         # active
diff --git a/git/test/test_git.py b/git/test/test_git.py
@@ -238,7 +238,9 @@ def counter_stderr(line):
                                 stdin=None,
                                 stdout=subprocess.PIPE,
                                 stderr=subprocess.PIPE,
-                                shell=False)
+                                shell=False,
+                                creationflags=Git.CREATE_NO_WINDOW if sys.platform == 'win32' else 0,
+                                )
 
         handle_process_output(proc, counter_stdout, counter_stderr, lambda proc: proc.wait())
 
