D2380: sshpeer: return framed file object when needed

[indygreg (Gregory Szorc)]

indygreg created this revision.
Herald added a subscriber: mercurial-devel.
Herald added a reviewer: hg-reviewers.

REVISION SUMMARY
  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.

REPOSITORY
  rHG Mercurial

REVISION DETAIL
  https://phab.mercurial-scm.org/D2380

AFFECTED FILES
  mercurial/sshpeer.py

CHANGE DETAILS

diff --git a/mercurial/sshpeer.py b/mercurial/sshpeer.py
--- a/mercurial/sshpeer.py
+++ b/mercurial/sshpeer.py
@@ -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)



To: indygreg, #hg-reviewers
Cc: mercurial-devel