comparison mercurial/wireproto.py @ 28438:48fd02dac1d4

wireproto: make iterbatcher behave streamily over http(s)

Unfortunately, the ssh and http implementations are slightly different
due to differences in their _callstream implementations, which
prevents ssh from behaving streamily. We should probably introduce a
new batch command that can stream results over ssh at some point in
the near future.

The streamy behavior of batch over http(s) is an enormous win for
remotefilelog over http: in my testing, it's saving about 40% on file
fetches with a cold cache against a server on localhost.

28437:c3eacee01c7e

# This software may be used and distributed according to the terms of the
# GNU General Public License version 2 or any later version.

from __future__ import absolute_import

import os
import sys
import tempfile
import urllib

class remoteiterbatcher(peer.iterbatcher):
    def __init__(self, remote):
        super(remoteiterbatcher, self).__init__()
        self._remote = remote

    def submit(self):
        """Break the batch request into many patch calls and pipeline them.

        This is mostly valuable over http where request sizes can be
        limited, but can be used in other places as well.
        """
        rb = self._remote.batch()
        rb.calls = self.calls
        rb.submit()

    def results(self):
        for name, args, opts, resref in self.calls:
            yield resref.value

# Forward a couple of names from peer to make wireproto interactions
# slightly more sensible.
batchable = peer.batchable
future = peer.future

        if self.capable('batch'):
            return remotebatch(self)
        else:
            return peer.localbatch(self)
    def _submitbatch(self, req):
        cmds = []
        for op, argsdict in req:
            args = ','.join('%s=%s' % (escapearg(k), escapearg(v))
                            for k, v in argsdict.iteritems())
            cmds.append('%s %s' % (op, args))
        rsp = self._call("batch", cmds=';'.join(cmds))
        return [unescapearg(r) for r in rsp.split(';')]
    def _submitone(self, op, args):
        return self._call(op, **args)

    def iterbatch(self):
        return remoteiterbatcher(self)

28438:48fd02dac1d4

# This software may be used and distributed according to the terms of the
# GNU General Public License version 2 or any later version.

from __future__ import absolute_import

import itertools
import os
import sys
import tempfile
import urllib

class remoteiterbatcher(peer.iterbatcher):
    def __init__(self, remote):
        super(remoteiterbatcher, self).__init__()
        self._remote = remote

    def __getattr__(self, name):
        if not getattr(self._remote, name, False):
            raise AttributeError(
                'Attempted to iterbatch non-batchable call to %r' % name)
        return super(remoteiterbatcher, self).__getattr__(name)

    def submit(self):
        """Break the batch request into many patch calls and pipeline them.

        This is mostly valuable over http where request sizes can be
        limited, but can be used in other places as well.
        """
        req, rsp = [], []
        for name, args, opts, resref in self.calls:
            mtd = getattr(self._remote, name)
            batchable = mtd.batchable(mtd.im_self, *args, **opts)
            encargsorres, encresref = batchable.next()
            assert encresref
            req.append((name, encargsorres))
            rsp.append((batchable, encresref))
        if req:
            self._resultiter = self._remote._submitbatch(req)
        self._rsp = rsp

    def results(self):
        for (batchable, encresref), encres in itertools.izip(
                self._rsp, self._resultiter):
            encresref.set(encres)
            yield batchable.next()

# Forward a couple of names from peer to make wireproto interactions
# slightly more sensible.
batchable = peer.batchable
future = peer.future

        if self.capable('batch'):
            return remotebatch(self)
        else:
            return peer.localbatch(self)
    def _submitbatch(self, req):
        """run batch request <req> on the server

        Returns an iterator of the raw responses from the server.
        """
        cmds = []
        for op, argsdict in req:
            args = ','.join('%s=%s' % (escapearg(k), escapearg(v))
                            for k, v in argsdict.iteritems())
            cmds.append('%s %s' % (op, args))
        rsp = self._callstream("batch", cmds=';'.join(cmds))
        # TODO this response parsing is probably suboptimal for large
        # batches with large responses.
        work = rsp.read(1024)
        chunk = work
        while chunk:
            while ';' in work:
                one, work = work.split(';', 1)
                yield unescapearg(one)
            chunk = rsp.read(1024)
            work += chunk
        yield unescapearg(work)

    def _submitone(self, op, args):
        return self._call(op, **args)

    def iterbatch(self):
        return remoteiterbatcher(self)