Merge branch 'nonblocking-ops'

Fixes #145

6 files changed, 201 insertions, 59 deletions

diff --git a/doc/source/changes.rst b/doc/source/changes.rst
index b7479e4f..4983b3d0 100644
--- a/doc/source/changes.rst
+++ b/doc/source/changes.rst
@@ -2,6 +2,11 @@
 Changelog
 =========
 
+0.3.5 - Bugfixes
+================
+* push/pull/fetch operations will not block anymore
+* A list of all fixed issues can be found here: https://github.com/gitpython-developers/GitPython/issues?q=milestone%3A%22v0.3.5+-+bugfixes%22+
+
 0.3.4 - Python 3 Support
 ========================
 * Internally, hexadecimal SHA1 are treated as ascii encoded strings. Binary SHA1 are treated as bytes.
diff --git a/git/cmd.py b/git/cmd.py
index ef370fe2..f847166c 100644
--- a/git/cmd.py
+++ b/git/cmd.py
@@ -6,7 +6,9 @@
 
 import os
 import sys
+import select
 import logging
+import threading
 from subprocess import (
     call,
     Popen,
@@ -16,7 +18,8 @@ from subprocess import (
 
 from .util import (
     LazyMixin,
-    stream_copy
+    stream_copy,
+    WaitGroup
 )
 from .exc import GitCommandError
 from git.compat import (
@@ -36,9 +39,121 @@ log = logging.getLogger('git.cmd')
 __all__ = ('Git', )
 
 
+# ==============================================================================
+## @name Utilities
+# ------------------------------------------------------------------------------
+# Documentation
+## @{
+
+def handle_process_output(process, stdout_handler, stderr_handler, finalizer):
+    """Registers for notifications to lean that process output is ready to read, and dispatches lines to
+    the respective line handlers. We are able to handle carriage returns in case progress is sent by that
+    mean. For performance reasons, we only apply this to stderr.
+    This function returns once the finalizer returns
+    :return: result of finalizer
+    :param process: subprocess.Popen instance
+    :param stdout_handler: f(stdout_line_string), or None
+    :param stderr_hanlder: f(stderr_line_string), or None
+    :param finalizer: f(proc) - wait for proc to finish"""
+    def read_line_fast(stream):
+        return stream.readline()
+
+    def read_line_slow(stream):
+        line = b''
+        while True:
+            char = stream.read(1)       # reads individual single byte strings
+            if not char:
+                break
+
+            if char in (b'\r', b'\n') and line:
+                break
+            else:
+                line += char
+            # END process parsed line
+        # END while file is not done reading
+        return line
+    # end
+
+    def dispatch_line(fno):
+            stream, handler, readline = fdmap[fno]
+            # this can possibly block for a while, but since we wake-up with at least one or more lines to handle,
+            # we are good ...
+            line = readline(stream).decode(defenc)
+            if line and handler:
+                handler(line)
+            return line
+        # end dispatch helper
+    # end
+
+    def deplete_buffer(fno, wg=None):
+        while True:
+            line = dispatch_line(fno)
+            if not line:
+                break
+        # end deplete buffer
+        if wg:
+            wg.done()
+    # end
+
+    fdmap = {process.stdout.fileno(): (process.stdout, stdout_handler, read_line_fast),
+             process.stderr.fileno(): (process.stderr, stderr_handler, read_line_slow)}
+
+    if hasattr(select, 'poll'):
+        # poll is preferred, as select is limited to file handles up to 1024 ... . This could otherwise be
+        # an issue for us, as it matters how many handles or own process has
+        poll = select.poll()
+        READ_ONLY = select.POLLIN | select.POLLPRI | select.POLLHUP | select.POLLERR
+        CLOSED = select.POLLHUP | select.POLLERR
+
+        poll.register(process.stdout, READ_ONLY)
+        poll.register(process.stderr, READ_ONLY)
+
+        closed_streams = set()
+        while True:
+            # no timeout
+            poll_result = poll.poll()
+            for fd, result in poll_result:
+                if result & CLOSED:
+                    closed_streams.add(fd)
+                else:
+                    dispatch_line(fd)
+                # end handle closed stream
+            # end for each poll-result tuple
+
+            if len(closed_streams) == len(fdmap):
+                break
+            # end its all done
+        # end endless loop
+
+        # Depelete all remaining buffers
+        for fno in fdmap.keys():
+            deplete_buffer(fno)
+        # end for each file handle
+    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
+        # 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 in fdmap.keys():
+            wg.add(1)
+            t = threading.Thread(target=lambda: deplete_buffer(fno, wg))
+            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
+        wg.wait()
+    # end
+
+    return finalizer(process)
+
+
 def dashify(string):
     return string.replace('_', '-')
 
+## -- End Utilities -- @}
+
 
 class Git(LazyMixin):
 
diff --git a/git/index/base.py b/git/index/base.py
index cc883469..66fd5b1f 100644
--- a/git/index/base.py
+++ b/git/index/base.py
@@ -287,11 +287,11 @@ class IndexFile(LazyMixin, diff.Diffable, Serializable):
             changes according to the amount of trees.
             If 1 Tree is given, it will just be read into a new index
             If 2 Trees are given, they will be merged into a new index using a
-             two way merge algorithm. Tree 1 is the 'current' tree, tree 2 is the 'other'
-             one. It behaves like a fast-forward.
-             If 3 Trees are given, a 3-way merge will be performed with the first tree
-             being the common ancestor of tree 2 and tree 3. Tree 2 is the 'current' tree,
-             tree 3 is the 'other' one
+            two way merge algorithm. Tree 1 is the 'current' tree, tree 2 is the 'other'
+            one. It behaves like a fast-forward.
+            If 3 Trees are given, a 3-way merge will be performed with the first tree
+            being the common ancestor of tree 2 and tree 3. Tree 2 is the 'current' tree,
+            tree 3 is the 'other' one
 
         :param kwargs:
             Additional arguments passed to git-read-tree
@@ -882,14 +882,11 @@ class IndexFile(LazyMixin, diff.Diffable, Serializable):
 
     def commit(self, message, parent_commits=None, head=True, author=None, committer=None):
         """Commit the current default index file, creating a commit object.
-
         For more information on the arguments, see tree.commit.
-        :note:
-            If you have manually altered the .entries member of this instance,
-            don't forget to write() your changes to disk beforehand.
 
-        :return:
-            Commit object representing the new commit"""
+        :note: If you have manually altered the .entries member of this instance,
+               don't forget to write() your changes to disk beforehand.
+        :return: Commit object representing the new commit"""
         tree = self.write_tree()
         return Commit.create_from_tree(self.repo, tree, message, parent_commits,
                                        head, author=author, committer=committer)
diff --git a/git/remote.py b/git/remote.py
index 484bc031..87db5dd4 100644
--- a/git/remote.py
+++ b/git/remote.py
@@ -31,6 +31,7 @@ from git.util import (
     join_path,
     finalize_process
 )
+from git.cmd import handle_process_output
 from gitdb.util import join
 from git.compat import defenc
 
@@ -40,30 +41,6 @@ __all__ = ('RemoteProgress', 'PushInfo', 'FetchInfo', 'Remote')
 #{ Utilities
 
 
-def digest_process_messages(fh, progress):
-    """Read progress messages from file-like object fh, supplying the respective
-    progress messages to the progress instance.
-
-    :param fh: File handle to read from
-    :return: list(line, ...) list of lines without linebreaks that did
-        not contain progress information"""
-    line_so_far = b''
-    dropped_lines = list()
-    while True:
-        char = fh.read(1)       # reads individual single byte strings
-        if not char:
-            break
-
-        if char in (b'\r', b'\n') and line_so_far:
-            dropped_lines.extend(progress._parse_progress_line(line_so_far.decode(defenc)))
-            line_so_far = b''
-        else:
-            line_so_far += char
-        # END process parsed line
-    # END while file is not done reading
-    return dropped_lines
-
-
 def add_progress(kwargs, git, progress):
     """Add the --progress flag to the given kwargs dict if supported by the
     git command. If the actual progress in the given progress instance is not
@@ -532,17 +509,25 @@ class Remote(LazyMixin, Iterable):
         # Basically we want all fetch info lines which appear to be in regular form, and thus have a
         # command character. Everything else we ignore,
         cmds = set(PushInfo._flag_map.keys()) & set(FetchInfo._flag_map.keys())
-        for line in digest_process_messages(proc.stderr, progress):
-            if line.startswith('fatal:'):
-                raise GitCommandError(("Error when fetching: %s" % line,), 2)
-            # END handle special messages
-            for cmd in cmds:
-                if line[1] == cmd:
-                    fetch_info_lines.append(line)
-                    continue
-                # end find command code
-            # end for each comand code we know
-        # END for each line
+
+        progress_handler = progress.new_message_handler()
+
+        def my_progress_handler(line):
+            for pline in progress_handler(line):
+                if line.startswith('fatal:'):
+                    raise GitCommandError(("Error when fetching: %s" % line,), 2)
+                # END handle special messages
+                for cmd in cmds:
+                    if line[1] == cmd:
+                        fetch_info_lines.append(line)
+                        continue
+                    # end find command code
+                # end for each comand code we know
+            # end for each line progress didn't handle
+        # end
+
+        # We are only interested in stderr here ...
+        handle_process_output(proc, None, my_progress_handler, finalize_process)
 
         # read head information
         fp = open(join(self.repo.git_dir, 'FETCH_HEAD'), 'rb')
@@ -555,7 +540,6 @@ class Remote(LazyMixin, Iterable):
 
         output.extend(FetchInfo._from_line(self.repo, err_line, fetch_line)
                       for err_line, fetch_line in zip(fetch_info_lines, fetch_head_info))
-        finalize_process(proc)
         return output
 
     def _get_push_info(self, proc, progress):
@@ -564,11 +548,10 @@ class Remote(LazyMixin, Iterable):
         # read the lines manually as it will use carriage returns between the messages
         # to override the previous one. This is why we read the bytes manually
         # TODO: poll() on file descriptors to know what to read next, process streams concurrently
-        digest_process_messages(proc.stderr, progress)
-
+        progress_handler = progress.new_message_handler()
         output = IterableList('name')
-        for line in proc.stdout.readlines():
-            line = line.decode(defenc)
+
+        def stdout_handler(line):
             try:
                 output.append(PushInfo._from_line(self, line))
             except ValueError:
@@ -576,7 +559,8 @@ class Remote(LazyMixin, Iterable):
                 pass
             # END exception handling
         # END for each line
-        finalize_process(proc)
+
+        handle_process_output(proc, stdout_handler, progress_handler, finalize_process)
         return output
 
     def fetch(self, refspec=None, progress=None, **kwargs):
diff --git a/git/repo/base.py b/git/repo/base.py
index 2a63492b..e8db3540 100644
--- a/git/repo/base.py
+++ b/git/repo/base.py
@@ -5,7 +5,10 @@
 # the BSD License: http://www.opensource.org/licenses/bsd-license.php
 
 from git.exc import InvalidGitRepositoryError, NoSuchPathError
-from git.cmd import Git
+from git.cmd import (
+    Git,
+    handle_process_output
+)
 from git.refs import (
     HEAD,
     Head,
@@ -25,7 +28,6 @@ from git.index import IndexFile
 from git.config import GitConfigParser
 from git.remote import (
     Remote,
-    digest_process_messages,
     add_progress
 )
 
@@ -711,9 +713,10 @@ class Repo(object):
             proc = git.clone(url, path, with_extended_output=True, as_process=True,
                              v=True, **add_progress(kwargs, git, progress))
             if progress:
-                digest_process_messages(proc.stderr, progress)
-            # END handle progress
-            finalize_process(proc)
+                handle_process_output(proc, None, progress.new_message_handler(), finalize_process)
+            else:
+                finalize_process(proc)
+            # end handle progress
         finally:
             if prev_cwd is not None:
                 os.chdir(prev_cwd)
diff --git a/git/util.py b/git/util.py
index 4de736d3..4d1ea8d6 100644
--- a/git/util.py
+++ b/git/util.py
@@ -12,6 +12,7 @@ import stat
 import shutil
 import platform
 import getpass
+import threading
 
 # NOTE:  Some of the unused imports might be used/imported by others.
 # Handle once test-cases are back up and running.
@@ -32,7 +33,7 @@ from gitdb.util import (  # NOQA
 __all__ = ("stream_copy", "join_path", "to_native_path_windows", "to_native_path_linux",
            "join_path_native", "Stats", "IndexFileSHA1Writer", "Iterable", "IterableList",
            "BlockingLockFile", "LockFile", 'Actor', 'get_user_id', 'assure_directory_exists',
-           'RemoteProgress', 'rmtree')
+           'RemoteProgress', 'rmtree', 'WaitGroup')
 
 #{ Utility Methods
 
@@ -249,6 +250,14 @@ class RemoteProgress(object):
         # END for each sub line
         return failed_lines
 
+    def new_message_handler(self):
+        """:return: a progress handler suitable for handle_process_output(), passing lines on to this Progress
+        handler in a suitable format"""
+        def handler(line):
+            return self._parse_progress_line(line.rstrip())
+        # end
+        return handler
+
     def line_dropped(self, line):
         """Called whenever a line could not be understood and was therefore dropped."""
         pass
@@ -691,3 +700,32 @@ class Iterable(object):
         raise NotImplementedError("To be implemented by Subclass")
 
 #} END classes
+
+
+class WaitGroup(object):
+    """WaitGroup is like Go sync.WaitGroup.
+
+    Without all the useful corner cases.
+    By Peter Teichman, taken from https://gist.github.com/pteichman/84b92ae7cef0ab98f5a8
+    """
+    def __init__(self):
+        self.count = 0
+        self.cv = threading.Condition()
+
+    def add(self, n):
+        self.cv.acquire()
+        self.count += n
+        self.cv.release()
+
+    def done(self):
+        self.cv.acquire()
+        self.count -= 1
+        if self.count == 0:
+            self.cv.notify_all()
+        self.cv.release()
+
+    def wait(self):
+        self.cv.acquire()
+        while self.count > 0:
+            self.cv.wait()
+        self.cv.release()