Skip to content

Commit 4914405

Browse files
ByronByron
committed
Implemented non-blocking operations using poll()
Next up is using threads
1 parent d83f6e8 commit 4914405

File tree

6 files changed

+149
-58
lines changed

6 files changed

+149
-58
lines changed

‎doc/source/changes.rst‎

Lines changed: 5 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -2,6 +2,11 @@
22
Changelog
33
=========
44

5+
0.3.5 - Bugfixes
6+
================
7+
* push/pull/fetch operations will not block anymore
8+
* 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+
9+
510
0.3.4 - Python 3 Support
611
========================
712
* Internally, hexadecimal SHA1 are treated as ascii encoded strings. Binary SHA1 are treated as bytes.

‎git/cmd.py‎

Lines changed: 96 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -6,6 +6,7 @@
66

77
import os
88
import sys
9+
import select
910
import logging
1011
from subprocess import (
1112
call,
@@ -36,9 +37,104 @@
3637
__all__ = ('Git', )
3738

3839

40+
# ==============================================================================
41+
## @name Utilities
42+
# ------------------------------------------------------------------------------
43+
# Documentation
44+
## @{
45+
46+
def handle_process_output(process, stdout_handler, stderr_handler, finalizer):
47+
"""Registers for notifications to lean that process output is ready to read, and dispatches lines to
48+
the respective line handlers. We are able to handle carriage returns in case progress is sent by that
49+
mean. For performance reasons, we only apply this to stderr.
50+
This function returns once the finalizer returns
51+
:return: result of finalizer
52+
:param process: subprocess.Popen instance
53+
:param stdout_handler: f(stdout_line_string), or None
54+
:param stderr_hanlder: f(stderr_line_string), or None
55+
:param finalizer: f(proc) - wait for proc to finish"""
56+
def read_line_fast(stream):
57+
return stream.readline()
58+
59+
def read_line_slow(stream):
60+
line = b''
61+
while True:
62+
char = stream.read(1) # reads individual single byte strings
63+
if not char:
64+
break
65+
66+
if char in (b'\r', b'\n') and line:
67+
break
68+
else:
69+
line += char
70+
# END process parsed line
71+
# END while file is not done reading
72+
return line
73+
# end
74+
75+
fdmap = { process.stdout.fileno() : (process.stdout, stdout_handler, read_line_fast),
76+
process.stderr.fileno() : (process.stderr, stderr_handler, read_line_slow) }
77+
78+
if hasattr(select, 'poll'):
79+
def dispatch_line(fd):
80+
stream, handler, readline = fdmap[fd]
81+
# this can possibly block for a while, but since we wake-up with at least one or more lines to handle,
82+
# we are good ...
83+
line = readline(stream).decode(defenc)
84+
if line and handler:
85+
handler(line)
86+
return line
87+
# end dispatch helper
88+
89+
# poll is preferred, as select is limited to file handles up to 1024 ... . Not an issue for us though,
90+
# as we deal with relatively blank processes
91+
poll = select.poll()
92+
READ_ONLY = select.POLLIN | select.POLLPRI | select.POLLHUP | select.POLLERR
93+
CLOSED = select.POLLHUP | select.POLLERR
94+
95+
poll.register(process.stdout, READ_ONLY)
96+
poll.register(process.stderr, READ_ONLY)
97+
98+
closed_streams = set()
99+
while True:
100+
# no timeout
101+
poll_result = poll.poll()
102+
for fd, result in poll_result:
103+
if result & CLOSED:
104+
closed_streams.add(fd)
105+
else:
106+
dispatch_line(fd)
107+
# end handle closed stream
108+
# end for each poll-result tuple
109+
110+
if len(closed_streams) == len(fdmap):
111+
break
112+
# end its all done
113+
# end endless loop
114+
115+
# Depelete all remaining buffers
116+
for fno, _ in fdmap.items():
117+
while True:
118+
line = dispatch_line(fno)
119+
if not line:
120+
break
121+
# end deplete buffer
122+
# end for each file handle
123+
else:
124+
# Oh ... probably we are on windows. select.select() can only handle sockets, we have files
125+
# The only reliable way to do this now is to use threads and wait for both to finish
126+
# Since the finalizer is expected to wait, we don't have to introduce our own wait primitive
127+
raise NotImplementedError()
128+
# end
129+
130+
return finalizer(process)
131+
132+
39133
def dashify(string):
40134
return string.replace('_', '-')
41135

136+
## -- End Utilities -- @}
137+
42138

43139
class Git(LazyMixin):
44140

‎git/index/base.py‎

Lines changed: 8 additions & 11 deletions
Original file line numberDiff line numberDiff line change
@@ -287,11 +287,11 @@ def from_tree(cls, repo, *treeish, **kwargs):
287287
changes according to the amount of trees.
288288
If 1 Tree is given, it will just be read into a new index
289289
If 2 Trees are given, they will be merged into a new index using a
290-
two way merge algorithm. Tree 1 is the 'current' tree, tree 2 is the 'other'
291-
one. It behaves like a fast-forward.
292-
If 3 Trees are given, a 3-way merge will be performed with the first tree
293-
being the common ancestor of tree 2 and tree 3. Tree 2 is the 'current' tree,
294-
tree 3 is the 'other' one
290+
two way merge algorithm. Tree 1 is the 'current' tree, tree 2 is the 'other'
291+
one. It behaves like a fast-forward.
292+
If 3 Trees are given, a 3-way merge will be performed with the first tree
293+
being the common ancestor of tree 2 and tree 3. Tree 2 is the 'current' tree,
294+
tree 3 is the 'other' one
295295
296296
:param kwargs:
297297
Additional arguments passed to git-read-tree
@@ -882,14 +882,11 @@ def move(self, items, skip_errors=False, **kwargs):
882882

883883
def commit(self, message, parent_commits=None, head=True, author=None, committer=None):
884884
"""Commit the current default index file, creating a commit object.
885-
886885
For more information on the arguments, see tree.commit.
887-
:note:
888-
If you have manually altered the .entries member of this instance,
889-
don't forget to write() your changes to disk beforehand.
890886
891-
:return:
892-
Commit object representing the new commit"""
887+
:note: If you have manually altered the .entries member of this instance,
888+
don't forget to write() your changes to disk beforehand.
889+
:return: Commit object representing the new commit"""
893890
tree = self.write_tree()
894891
return Commit.create_from_tree(self.repo, tree, message, parent_commits,
895892
head, author=author, committer=committer)

‎git/remote.py‎

Lines changed: 24 additions & 42 deletions
Original file line numberDiff line numberDiff line change
@@ -31,6 +31,7 @@
3131
join_path,
3232
finalize_process
3333
)
34+
from git.cmd import handle_process_output
3435
from gitdb.util import join
3536
from git.compat import defenc
3637

@@ -39,31 +40,6 @@
3940

4041
#{ Utilities
4142

42-
43-
def digest_process_messages(fh, progress):
44-
"""Read progress messages from file-like object fh, supplying the respective
45-
progress messages to the progress instance.
46-
47-
:param fh: File handle to read from
48-
:return: list(line, ...) list of lines without linebreaks that did
49-
not contain progress information"""
50-
line_so_far = b''
51-
dropped_lines = list()
52-
while True:
53-
char = fh.read(1) # reads individual single byte strings
54-
if not char:
55-
break
56-
57-
if char in (b'\r', b'\n') and line_so_far:
58-
dropped_lines.extend(progress._parse_progress_line(line_so_far.decode(defenc)))
59-
line_so_far = b''
60-
else:
61-
line_so_far += char
62-
# END process parsed line
63-
# END while file is not done reading
64-
return dropped_lines
65-
66-
6743
def add_progress(kwargs, git, progress):
6844
"""Add the --progress flag to the given kwargs dict if supported by the
6945
git command. If the actual progress in the given progress instance is not
@@ -532,17 +508,24 @@ def _get_fetch_info_from_stderr(self, proc, progress):
532508
# Basically we want all fetch info lines which appear to be in regular form, and thus have a
533509
# command character. Everything else we ignore,
534510
cmds = set(PushInfo._flag_map.keys()) & set(FetchInfo._flag_map.keys())
535-
for line in digest_process_messages(proc.stderr, progress):
536-
if line.startswith('fatal:'):
537-
raise GitCommandError(("Error when fetching: %s" % line,), 2)
538-
# END handle special messages
539-
for cmd in cmds:
540-
if line[1] == cmd:
541-
fetch_info_lines.append(line)
542-
continue
543-
# end find command code
544-
# end for each comand code we know
545-
# END for each line
511+
512+
progress_handler = progress.new_message_handler()
513+
def my_progress_handler(line):
514+
for pline in progress_handler(line):
515+
if line.startswith('fatal:'):
516+
raise GitCommandError(("Error when fetching: %s" % line,), 2)
517+
# END handle special messages
518+
for cmd in cmds:
519+
if line[1] == cmd:
520+
fetch_info_lines.append(line)
521+
continue
522+
# end find command code
523+
# end for each comand code we know
524+
# end for each line progress didn't handle
525+
# end
526+
527+
# We are only interested in stderr here ...
528+
handle_process_output(proc, None, my_progress_handler, finalize_process)
546529

547530
# read head information
548531
fp = open(join(self.repo.git_dir, 'FETCH_HEAD'), 'rb')
@@ -555,7 +538,6 @@ def _get_fetch_info_from_stderr(self, proc, progress):
555538

556539
output.extend(FetchInfo._from_line(self.repo, err_line, fetch_line)
557540
for err_line, fetch_line in zip(fetch_info_lines, fetch_head_info))
558-
finalize_process(proc)
559541
return output
560542

561543
def _get_push_info(self, proc, progress):
@@ -564,19 +546,19 @@ def _get_push_info(self, proc, progress):
564546
# read the lines manually as it will use carriage returns between the messages
565547
# to override the previous one. This is why we read the bytes manually
566548
# TODO: poll() on file descriptors to know what to read next, process streams concurrently
567-
digest_process_messages(proc.stderr, progress)
568-
549+
progress_handler = progress.new_message_handler()
569550
output = IterableList('name')
570-
for line in proc.stdout.readlines():
571-
line = line.decode(defenc)
551+
552+
def stdout_handler(line):
572553
try:
573554
output.append(PushInfo._from_line(self, line))
574555
except ValueError:
575556
# if an error happens, additional info is given which we cannot parse
576557
pass
577558
# END exception handling
578559
# END for each line
579-
finalize_process(proc)
560+
561+
handle_process_output(proc, stdout_handler, progress_handler, finalize_process)
580562
return output
581563

582564
def fetch(self, refspec=None, progress=None, **kwargs):

‎git/repo/base.py‎

Lines changed: 8 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -5,7 +5,10 @@
55
# the BSD License: http://www.opensource.org/licenses/bsd-license.php
66

77
from git.exc import InvalidGitRepositoryError, NoSuchPathError
8-
from git.cmd import Git
8+
from git.cmd import (
9+
Git,
10+
handle_process_output
11+
)
912
from git.refs import (
1013
HEAD,
1114
Head,
@@ -25,7 +28,6 @@
2528
from git.config import GitConfigParser
2629
from git.remote import (
2730
Remote,
28-
digest_process_messages,
2931
add_progress
3032
)
3133

@@ -711,9 +713,10 @@ def _clone(cls, git, url, path, odb_default_type, progress, **kwargs):
711713
proc = git.clone(url, path, with_extended_output=True, as_process=True,
712714
v=True, **add_progress(kwargs, git, progress))
713715
if progress:
714-
digest_process_messages(proc.stderr, progress)
715-
# END handle progress
716-
finalize_process(proc)
716+
handle_process_output(proc, None, progress.new_message_handler(), finalize_process)
717+
else:
718+
finalize_process(proc)
719+
# end handle progress
717720
finally:
718721
if prev_cwd is not None:
719722
os.chdir(prev_cwd)

‎git/util.py‎

Lines changed: 8 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -249,6 +249,14 @@ def _parse_progress_line(self, line):
249249
# END for each sub line
250250
return failed_lines
251251

252+
def new_message_handler(self):
253+
""":return: a progress handler suitable for handle_process_output(), passing lines on to this Progress
254+
handler in a suitable format"""
255+
def handler(line):
256+
return self._parse_progress_line(line.rstrip())
257+
# end
258+
return handler
259+
252260
def line_dropped(self, line):
253261
"""Called whenever a line could not be understood and was therefore dropped."""
254262
pass

0 commit comments

Comments
 (0)