changeset 36367:043e77f3be09

sshpeer: return framed file object when needed Currently, wireproto.wirepeer has a default implementation of _submitbatch() and sshv1peer has a very similar implementation. The main difference is that sshv1peer is aware of the total amount of bytes it can read whereas the default implementation reads the stream until no more data is returned. The default implementation works for HTTP, since there is a known end to HTTP responses (either Content-Length or 0 sized chunk). This commit teaches sshv1peer to use our just-introduced "cappedreader" class for wrapping a file object to limit the number of bytes that can be read. We do this by introducing an argument to specify whether the response is framed. If set, we returned a cappedreader instance instead of the raw pipe. _call() always has framed responses. So we set this argument unconditionally and then .read() the entirety of the result. Strictly speaking, we don't need to use cappedreader in this case and can inline frame decoding/read logic. But I like when things are consistent. The overhead should be negligible. _callstream() and _callcompressable() are special: whether framing is used depends on the specific command. So, we define a set of commands that have framed response. It currently only contains "batch." As a result of this change, the one-off implementation of _submitbatch() in sshv1peer can be removed since it is now safe to .read() the response's file object until end of stream. cappedreader takes care of not overrunning the frame. Differential Revision: https://phab.mercurial-scm.org/D2380
author Gregory Szorc <gregory.szorc@gmail.com>
date Wed, 21 Feb 2018 08:35:48 -0800
parents a34d5ef53c2e
children 02782e6e2c38
files mercurial/sshpeer.py
diffstat 1 files changed, 18 insertions(+), 24 deletions(-) [+]
line wrap: on
line diff
--- a/mercurial/sshpeer.py	Wed Feb 21 08:33:50 2018 -0800
+++ b/mercurial/sshpeer.py	Wed Feb 21 08:35:48 2018 -0800
@@ -349,6 +349,12 @@
         self._pipee = stderr
         self._caps = caps
 
+    # Commands that have a "framed" response where the first line of the
+    # response contains the length of that response.
+    _FRAMED_COMMANDS = {
+        'batch',
+    }
+
     # Begin of _basepeer interface.
 
     @util.propertycache
@@ -391,26 +397,7 @@
 
     __del__ = _cleanup
 
-    def _submitbatch(self, req):
-        rsp = self._callstream("batch", cmds=wireproto.encodebatchcmds(req))
-        available = self._getamount()
-        # TODO this response parsing is probably suboptimal for large
-        # batches with large responses.
-        toread = min(available, 1024)
-        work = rsp.read(toread)
-        available -= toread
-        chunk = work
-        while chunk:
-            while ';' in work:
-                one, work = work.split(';', 1)
-                yield wireproto.unescapearg(one)
-            toread = min(available, 1024)
-            chunk = rsp.read(toread)
-            available -= toread
-            work += chunk
-        yield wireproto.unescapearg(work)
-
-    def _sendrequest(self, cmd, args):
+    def _sendrequest(self, cmd, args, framed=False):
         if (self.ui.debugflag
             and self.ui.configbool('devel', 'debug.peer-request')):
             dbg = self.ui.debug
@@ -444,20 +431,27 @@
                 self._pipeo.write(v)
         self._pipeo.flush()
 
+        # We know exactly how many bytes are in the response. So return a proxy
+        # around the raw output stream that allows reading exactly this many
+        # bytes. Callers then can read() without fear of overrunning the
+        # response.
+        if framed:
+            amount = self._getamount()
+            return util.cappedreader(self._pipei, amount)
+
         return self._pipei
 
     def _callstream(self, cmd, **args):
         args = pycompat.byteskwargs(args)
-        return self._sendrequest(cmd, args)
+        return self._sendrequest(cmd, args, framed=cmd in self._FRAMED_COMMANDS)
 
     def _callcompressable(self, cmd, **args):
         args = pycompat.byteskwargs(args)
-        return self._sendrequest(cmd, args)
+        return self._sendrequest(cmd, args, framed=cmd in self._FRAMED_COMMANDS)
 
     def _call(self, cmd, **args):
         args = pycompat.byteskwargs(args)
-        self._sendrequest(cmd, args)
-        return self._readframed()
+        return self._sendrequest(cmd, args, framed=True).read()
 
     def _callpush(self, cmd, fp, **args):
         r = self._call(cmd, **args)