test, gitpython-developers#519: Popen() pump: remove WaitGroup

Author: ankostis

git/cmd.py

@@ -25,7 +25,6 @@
 from .util import (
     LazyMixin,
     stream_copy,
-    WaitGroup
 )
 from .exc import (
     GitCommandError,
@@ -193,28 +192,22 @@ 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):
+        def _handle_lines(fd, handler):
             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()
+        threads = []
         for fd, handler in zip((process.stdout, process.stderr),
                                (stdout_handler, stderr_handler)):
-            wg.add(1)
-            t = threading.Thread(target=_handle_lines, args=(fd, handler, wg))
+            t = threading.Thread(target=_handle_lines, args=(fd, handler))
             t.setDaemon(True)
             t.start()
+            threads.append(t)
 
-        # 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()
+        for t in threads:
+            t.join()
     # end
 
     return finalizer(process)

git/util.py

@@ -12,7 +12,6 @@
 import shutil
 import platform
 import getpass
-import threading
 import logging
 
 # NOTE:  Some of the unused imports might be used/imported by others.
@@ -39,7 +38,7 @@
 __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', 'CallableRemoteProgress', 'rmtree', 'WaitGroup', 'unbare_repo')
+           'RemoteProgress', 'CallableRemoteProgress', 'rmtree', 'unbare_repo')
 
 #{ Utility Methods
 
@@ -324,12 +323,12 @@ def update(self, op_code, cur_count, max_count=None, message=''):
 
         You may read the contents of the current line in self._cur_line"""
         pass
-        
+
 
 class CallableRemoteProgress(RemoteProgress):
     """An implementation forwarding updates to any callable"""
     __slots__ = ('_callable')
-    
+
     def __init__(self, fn):
         self._callable = fn
         super(CallableRemoteProgress, self).__init__()
@@ -754,35 +753,6 @@ def iter_items(cls, repo, *args, **kwargs):
 #} 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, stderr=b''):
-        self.cv.acquire()
-        while self.count > 0:
-            self.cv.wait()
-        self.cv.release()
-
-
 class NullHandler(logging.Handler):
     def emit(self, record):
         pass