| ryanmce |
| Restricted Project |
| Automatic diff as part of commit; lint not applicable. |
| Automatic diff as part of commit; unit tests not applicable. |
cc @kulshrax -- coordinate landing of this with update of upstream version
| Path | Packages | |||
|---|---|---|---|---|
| M | hgext3rd/extutil.py (2 lines) | |||
| M | hgext3rd/fastannotate/protocol.py (11 lines) | |||
| M | remotefilelog/__init__.py (3 lines) | |||
| M | remotefilelog/connectionpool.py (14 lines) | |||
| M | remotefilelog/fileserverclient.py (55 lines) | |||
| M | remotefilelog/remotefilelogserver.py (5 lines) | |||
| M | remotefilelog/shallowutil.py (14 lines) | |||
| M | tests/test-remotefilelog-http.t (2 lines) |
| # extutil.py - useful utility methods for extensions | # extutil.py - useful utility methods for extensions | ||||
| # | # | ||||
| # Copyright 2016 Facebook | # Copyright 2016 Facebook | ||||
| # | # | ||||
| # This software may be used and distributed according to the terms of the | # This software may be used and distributed according to the terms of the | ||||
| # GNU General Public License version 2 or any later version. | # GNU General Public License version 2 or any later version. | ||||
| from __future__ import absolute_import | |||||
| import errno | import errno | ||||
| import os | import os | ||||
| import platform | import platform | ||||
| import subprocess | import subprocess | ||||
| if platform.system() == 'Windows': | if platform.system() == 'Windows': | ||||
| # no fork on Windows, but we can create a detached process | # no fork on Windows, but we can create a detached process | ||||
| # https://msdn.microsoft.com/en-us/library/windows/desktop/ms684863.aspx | # https://msdn.microsoft.com/en-us/library/windows/desktop/ms684863.aspx | ||||
| # Copyright 2016-present Facebook. All Rights Reserved. | # Copyright 2016-present Facebook. All Rights Reserved. | ||||
| # | # | ||||
| # protocol: logic for a server providing fastannotate support | # protocol: logic for a server providing fastannotate support | ||||
| # | # | ||||
| # This software may be used and distributed according to the terms of the | # This software may be used and distributed according to the terms of the | ||||
| # GNU General Public License version 2 or any later version. | # GNU General Public License version 2 or any later version. | ||||
| from mercurial import ( | from mercurial import ( | ||||
| error, | error, | ||||
| extensions, | extensions, | ||||
| hg, | hg, | ||||
| localrepo, | localrepo, | ||||
| peer as peermod, | |||||
| scmutil, | scmutil, | ||||
| wireproto, | wireproto, | ||||
| ) | ) | ||||
| from mercurial.i18n import _ | from mercurial.i18n import _ | ||||
| import contextlib | import contextlib | ||||
| import os | import os | ||||
| if peer is None: | if peer is None: | ||||
| with annotatepeer(repo) as peer: | with annotatepeer(repo) as peer: | ||||
| return clientfetch(repo, paths, lastnodemap, peer) | return clientfetch(repo, paths, lastnodemap, peer) | ||||
| if lastnodemap is None: | if lastnodemap is None: | ||||
| lastnodemap = {} | lastnodemap = {} | ||||
| ui = repo.ui | ui = repo.ui | ||||
| batcher = peer.batch() | batcher = peer.iterbatch() | ||||
| ui.debug('fastannotate: requesting %d files\n' % len(paths)) | ui.debug('fastannotate: requesting %d files\n' % len(paths)) | ||||
| results = [batcher.getannotate(p, lastnodemap.get(p)) for p in paths] | for p in paths: | ||||
| batcher.getannotate(p, lastnodemap.get(p)) | |||||
| # Note: This is the only place that fastannotate sends a request via SSH. | # Note: This is the only place that fastannotate sends a request via SSH. | ||||
| # The SSH stream should not be in the remotefilelog "getfiles" loop. | # The SSH stream should not be in the remotefilelog "getfiles" loop. | ||||
| batcher.submit() | batcher.submit() | ||||
| results = list(batcher.results()) | |||||
| ui.debug('fastannotate: server returned\n') | ui.debug('fastannotate: server returned\n') | ||||
| for result in results: | for result in results: | ||||
| if result.value is None: | for path, content in result.iteritems(): | ||||
| continue | |||||
| for path, content in result.value.iteritems(): | |||||
| # ignore malicious paths | # ignore malicious paths | ||||
| if not path.startswith('fastannotate/') or '/../' in (path + '/'): | if not path.startswith('fastannotate/') or '/../' in (path + '/'): | ||||
| ui.debug('fastannotate: ignored malicious path %s\n' % path) | ui.debug('fastannotate: ignored malicious path %s\n' % path) | ||||
| continue | continue | ||||
| if ui.debugflag: | if ui.debugflag: | ||||
| ui.debug('fastannotate: writing %d bytes to %s\n' | ui.debug('fastannotate: writing %d bytes to %s\n' | ||||
| % (len(content), path)) | % (len(content), path)) | ||||
| repo.vfs.makedirs(os.path.dirname(path)) | repo.vfs.makedirs(os.path.dirname(path)) | ||||
| wrapfunction(exchange, 'pull', pull_shallow) | wrapfunction(exchange, 'pull', pull_shallow) | ||||
| # Wrap the stream logic to add requirements and to pass include/exclude | # Wrap the stream logic to add requirements and to pass include/exclude | ||||
| # patterns around. | # patterns around. | ||||
| def setup_streamout(repo, remote): | def setup_streamout(repo, remote): | ||||
| # Replace remote.stream_out with a version that sends file | # Replace remote.stream_out with a version that sends file | ||||
| # patterns. | # patterns. | ||||
| def stream_out_shallow(orig): | def stream_out_shallow(orig): | ||||
| if shallowrepo.requirement in remote._capabilities(): | caps = shallowutil.peercapabilities(remote) | ||||
| if shallowrepo.requirement in caps: | |||||
| opts = {} | opts = {} | ||||
| if repo.includepattern: | if repo.includepattern: | ||||
| opts['includepattern'] = '\0'.join(repo.includepattern) | opts['includepattern'] = '\0'.join(repo.includepattern) | ||||
| if repo.excludepattern: | if repo.excludepattern: | ||||
| opts['excludepattern'] = '\0'.join(repo.excludepattern) | opts['excludepattern'] = '\0'.join(repo.excludepattern) | ||||
| return remote._callstream('stream_out_shallow', **opts) | return remote._callstream('stream_out_shallow', **opts) | ||||
| else: | else: | ||||
| return orig() | return orig() | ||||
| # connectionpool.py - class for pooling peer connections for reuse | # connectionpool.py - class for pooling peer connections for reuse | ||||
| # | # | ||||
| # Copyright 2017 Facebook, Inc. | # Copyright 2017 Facebook, Inc. | ||||
| # | # | ||||
| # This software may be used and distributed according to the terms of the | # This software may be used and distributed according to the terms of the | ||||
| # GNU General Public License version 2 or any later version. | # GNU General Public License version 2 or any later version. | ||||
| from __future__ import absolute_import | |||||
| from mercurial import ( | from mercurial import ( | ||||
| extensions, | extensions, | ||||
| hg, | hg, | ||||
| sshpeer, | sshpeer, | ||||
| util, | util, | ||||
| ) | ) | ||||
| from . import ( | |||||
| shallowutil, | |||||
| ) | |||||
| class connectionpool(object): | class connectionpool(object): | ||||
| def __init__(self, repo): | def __init__(self, repo): | ||||
| self._repo = repo | self._repo = repo | ||||
| self._pool = dict() | self._pool = dict() | ||||
| def get(self, path): | def get(self, path): | ||||
| pathpool = self._pool.get(path) | pathpool = self._pool.get(path) | ||||
| if pathpool is None: | if pathpool is None: | ||||
| pathpool = list() | pathpool = list() | ||||
| self._pool[path] = pathpool | self._pool[path] = pathpool | ||||
| conn = None | conn = None | ||||
| if len(pathpool) > 0: | if len(pathpool) > 0: | ||||
| try: | try: | ||||
| conn = pathpool.pop() | conn = pathpool.pop() | ||||
| peer = conn.peer | peer = conn.peer | ||||
| # If the connection has died, drop it | # If the connection has died, drop it | ||||
| if (isinstance(peer, sshpeer.sshpeer) and | if isinstance(peer, sshpeer.sshpeer): | ||||
| peer.subprocess.poll() is not None): | proc = shallowutil.trygetattr( | ||||
| peer, ('_subprocess', 'subprocess')) | |||||
| if proc.poll() is not None: | |||||
| conn = None | conn = None | ||||
| except IndexError: | except IndexError: | ||||
| pass | pass | ||||
| if conn is None: | if conn is None: | ||||
| def _cleanup(orig): | def _cleanup(orig): | ||||
| # close pipee first so peer.cleanup reading it won't deadlock, | # close pipee first so peer.cleanup reading it won't deadlock, | ||||
| # if there are other processes with pipeo open (i.e. us). | # if there are other processes with pipeo open (i.e. us). | ||||
| peer = orig.im_self | peer = orig.im_self | ||||
| # fileserverclient.py - client for communicating with the cache process | # fileserverclient.py - client for communicating with the cache process | ||||
| # | # | ||||
| # Copyright 2013 Facebook, Inc. | # Copyright 2013 Facebook, Inc. | ||||
| # | # | ||||
| # This software may be used and distributed according to the terms of the | # This software may be used and distributed according to the terms of the | ||||
| # GNU General Public License version 2 or any later version. | # GNU General Public License version 2 or any later version. | ||||
| from mercurial.i18n import _ | from __future__ import absolute_import | ||||
| from mercurial.node import hex, bin, nullid | |||||
| from mercurial import util, sshpeer, error, util, wireproto, httppeer | |||||
| import hashlib, os, time, io, struct | import hashlib, os, time, io, struct | ||||
| import itertools | import itertools | ||||
| from mercurial.i18n import _ | |||||
| from mercurial.node import hex, bin, nullid | |||||
| from mercurial import ( | |||||
| error, | |||||
| httppeer, | |||||
| peer as peermod, | |||||
| sshpeer, | |||||
| util, | |||||
| util, | |||||
| wireproto, | |||||
| ) | |||||
| from . import ( | from . import ( | ||||
| connectionpool, | connectionpool, | ||||
| constants, | constants, | ||||
| shallowutil, | shallowutil, | ||||
| wirepack, | wirepack, | ||||
| ) | ) | ||||
| from .contentstore import unioncontentstore | from .contentstore import unioncontentstore | ||||
| from .metadatastore import unionmetadatastore | from .metadatastore import unionmetadatastore | ||||
| from lz4wrapper import lz4decompress | from .lz4wrapper import lz4decompress | ||||
| # Statistics for debugging | # Statistics for debugging | ||||
| fetchcost = 0 | fetchcost = 0 | ||||
| fetches = 0 | fetches = 0 | ||||
| fetched = 0 | fetched = 0 | ||||
| fetchmisses = 0 | fetchmisses = 0 | ||||
| _downloading = _('downloading') | _downloading = _('downloading') | ||||
| f = wireproto.future() | f = wireproto.future() | ||||
| yield {'path': path}, f | yield {'path': path}, f | ||||
| heads = f.value.split('\n') if f.value else [] | heads = f.value.split('\n') if f.value else [] | ||||
| yield heads | yield heads | ||||
| def _updatecallstreamopts(self, command, opts): | def _updatecallstreamopts(self, command, opts): | ||||
| if command != 'getbundle': | if command != 'getbundle': | ||||
| return | return | ||||
| if 'remotefilelog' not in self._capabilities(): | if 'remotefilelog' not in shallowutil.peercapabilities(self): | ||||
| return | return | ||||
| if not util.safehasattr(self, '_localrepo'): | if not util.safehasattr(self, '_localrepo'): | ||||
| return | return | ||||
| if constants.REQUIREMENT not in self._localrepo.requirements: | if constants.REQUIREMENT not in self._localrepo.requirements: | ||||
| return | return | ||||
| bundlecaps = opts.get('bundlecaps') | bundlecaps = opts.get('bundlecaps') | ||||
| if bundlecaps: | if bundlecaps: | ||||
| for m, r in itertools.izip(missed, b.results()): | for m, r in itertools.izip(missed, b.results()): | ||||
| file_ = idmap[m] | file_ = idmap[m] | ||||
| node = m[-40:] | node = m[-40:] | ||||
| receivemissing(io.BytesIO('%d\n%s' % (len(r), r)), file_, node) | receivemissing(io.BytesIO('%d\n%s' % (len(r), r)), file_, node) | ||||
| progresstick() | progresstick() | ||||
| return | return | ||||
| while missed: | while missed: | ||||
| chunk, missed = missed[:batchsize], missed[batchsize:] | chunk, missed = missed[:batchsize], missed[batchsize:] | ||||
| b = remote.batch() | b = remote.iterbatch() | ||||
| futures = {} | |||||
| for m in chunk: | for m in chunk: | ||||
| file_ = idmap[m] | file_ = idmap[m] | ||||
| node = m[-40:] | node = m[-40:] | ||||
| futures[m] = b.getfile(file_, node) | b.getfile(file_, node) | ||||
| b.submit() | b.submit() | ||||
| for m in chunk: | for m, v in zip(chunk, b.results()): | ||||
| v = futures[m].value | |||||
| file_ = idmap[m] | file_ = idmap[m] | ||||
| node = m[-40:] | node = m[-40:] | ||||
| receivemissing(io.BytesIO('%d\n%s' % (len(v), v)), file_, node) | receivemissing(io.BytesIO('%d\n%s' % (len(v), v)), file_, node) | ||||
| progresstick() | progresstick() | ||||
| def _getfiles( | def _getfiles( | ||||
| remote, receivemissing, progresstick, missed, idmap, step): | remote, receivemissing, progresstick, missed, idmap, step): | ||||
| remote._callstream("getfiles") | remote._callstream("getfiles") | ||||
| i = 0 | i = 0 | ||||
| pipeo = shallowutil.trygetattr(remote, ('_pipeo', 'pipeo')) | |||||
| pipei = shallowutil.trygetattr(remote, ('_pipei', 'pipei')) | |||||
| while i < len(missed): | while i < len(missed): | ||||
| # issue a batch of requests | # issue a batch of requests | ||||
| start = i | start = i | ||||
| end = min(len(missed), start + step) | end = min(len(missed), start + step) | ||||
| i = end | i = end | ||||
| for missingid in missed[start:end]: | for missingid in missed[start:end]: | ||||
| # issue new request | # issue new request | ||||
| versionid = missingid[-40:] | versionid = missingid[-40:] | ||||
| file = idmap[missingid] | file = idmap[missingid] | ||||
| sshrequest = "%s%s\n" % (versionid, file) | sshrequest = "%s%s\n" % (versionid, file) | ||||
| remote.pipeo.write(sshrequest) | pipeo.write(sshrequest) | ||||
| remote.pipeo.flush() | pipeo.flush() | ||||
| # receive batch results | # receive batch results | ||||
| for missingid in missed[start:end]: | for missingid in missed[start:end]: | ||||
| versionid = missingid[-40:] | versionid = missingid[-40:] | ||||
| file = idmap[missingid] | file = idmap[missingid] | ||||
| receivemissing(remote.pipei, file, versionid) | receivemissing(pipei, file, versionid) | ||||
| progresstick() | progresstick() | ||||
| # End the command | # End the command | ||||
| remote.pipeo.write('\n') | pipeo.write('\n') | ||||
| remote.pipeo.flush() | pipeo.flush() | ||||
| class fileserverclient(object): | class fileserverclient(object): | ||||
| """A client for requesting files from the remote file server. | """A client for requesting files from the remote file server. | ||||
| """ | """ | ||||
| def __init__(self, repo): | def __init__(self, repo): | ||||
| ui = repo.ui | ui = repo.ui | ||||
| self.repo = repo | self.repo = repo | ||||
| self.ui = ui | self.ui = ui | ||||
| remote = conn.peer | remote = conn.peer | ||||
| remote._callstream("getpackv1") | remote._callstream("getpackv1") | ||||
| self._sendpackrequest(remote, fileids) | self._sendpackrequest(remote, fileids) | ||||
| packpath = shallowutil.getcachepackpath( | packpath = shallowutil.getcachepackpath( | ||||
| self.repo, constants.FILEPACK_CATEGORY) | self.repo, constants.FILEPACK_CATEGORY) | ||||
| pipei = shallowutil.trygetattr(remote, ('_pipei', 'pipei')) | |||||
| receiveddata, receivedhistory = wirepack.receivepack( | receiveddata, receivedhistory = wirepack.receivepack( | ||||
| self.repo.ui, remote.pipei, packpath) | self.repo.ui, pipei, packpath) | ||||
| rcvd = len(receiveddata) | rcvd = len(receiveddata) | ||||
| self.ui.log("remotefilefetchlog", | self.ui.log("remotefilefetchlog", | ||||
| "Success(pack)" if (rcvd==total) else "Fail(pack)", | "Success(pack)" if (rcvd==total) else "Fail(pack)", | ||||
| fetched_files = rcvd, | fetched_files = rcvd, | ||||
| total_to_fetch = total) | total_to_fetch = total) | ||||
| except Exception: | except Exception: | ||||
| self.ui.log("remotefilefetchlog", | self.ui.log("remotefilefetchlog", | ||||
| "Fail(pack)", | "Fail(pack)", | ||||
| fetched_files = rcvd, | fetched_files = rcvd, | ||||
| total_to_fetch = total) | total_to_fetch = total) | ||||
| raise | raise | ||||
| def _sendpackrequest(self, remote, fileids): | def _sendpackrequest(self, remote, fileids): | ||||
| """Formats and writes the given fileids to the remote as part of a | """Formats and writes the given fileids to the remote as part of a | ||||
| getpackv1 call. | getpackv1 call. | ||||
| """ | """ | ||||
| # Sort the requests by name, so we receive requests in batches by name | # Sort the requests by name, so we receive requests in batches by name | ||||
| grouped = {} | grouped = {} | ||||
| for filename, node in fileids: | for filename, node in fileids: | ||||
| grouped.setdefault(filename, set()).add(node) | grouped.setdefault(filename, set()).add(node) | ||||
| # Issue request | # Issue request | ||||
| pipeo = shallowutil.trygetattr(remote, ('_pipeo', 'pipeo')) | |||||
| for filename, nodes in grouped.iteritems(): | for filename, nodes in grouped.iteritems(): | ||||
| filenamelen = struct.pack(constants.FILENAMESTRUCT, len(filename)) | filenamelen = struct.pack(constants.FILENAMESTRUCT, len(filename)) | ||||
| countlen = struct.pack(constants.PACKREQUESTCOUNTSTRUCT, len(nodes)) | countlen = struct.pack(constants.PACKREQUESTCOUNTSTRUCT, len(nodes)) | ||||
| rawnodes = ''.join(bin(n) for n in nodes) | rawnodes = ''.join(bin(n) for n in nodes) | ||||
| remote.pipeo.write('%s%s%s%s' % (filenamelen, filename, countlen, | pipeo.write('%s%s%s%s' % (filenamelen, filename, countlen, | ||||
| rawnodes)) | rawnodes)) | ||||
| remote.pipeo.flush() | pipeo.flush() | ||||
| remote.pipeo.write(struct.pack(constants.FILENAMESTRUCT, 0)) | pipeo.write(struct.pack(constants.FILENAMESTRUCT, 0)) | ||||
| remote.pipeo.flush() | pipeo.flush() | ||||
| def connect(self): | def connect(self): | ||||
| if self.cacheprocess: | if self.cacheprocess: | ||||
| cmd = "%s %s" % (self.cacheprocess, self.writedata._path) | cmd = "%s %s" % (self.cacheprocess, self.writedata._path) | ||||
| self.remotecache.connect(cmd) | self.remotecache.connect(cmd) | ||||
| else: | else: | ||||
| # If no cache process is specified, we fake one that always | # If no cache process is specified, we fake one that always | ||||
| # returns cache misses. This enables tests to run easily | # returns cache misses. This enables tests to run easily | ||||
| if ((shallowrepo.requirement in repo.requirements or | if ((shallowrepo.requirement in repo.requirements or | ||||
| ui.configbool('remotefilelog', 'server'))): | ui.configbool('remotefilelog', 'server'))): | ||||
| if isinstance(proto, sshserver.sshserver): | if isinstance(proto, sshserver.sshserver): | ||||
| # legacy getfiles method which only works over ssh | # legacy getfiles method which only works over ssh | ||||
| caps.append(shallowrepo.requirement) | caps.append(shallowrepo.requirement) | ||||
| caps.append('getflogheads') | caps.append('getflogheads') | ||||
| caps.append('getfile') | caps.append('getfile') | ||||
| return caps | return caps | ||||
| if util.safehasattr(wireproto, '_capabilities'): | |||||
| wrapfunction(wireproto, '_capabilities', _capabilities) | wrapfunction(wireproto, '_capabilities', _capabilities) | ||||
| else: | |||||
| wrapfunction(wireproto, 'capabilities', _capabilities) | |||||
| def _adjustlinkrev(orig, self, *args, **kwargs): | def _adjustlinkrev(orig, self, *args, **kwargs): | ||||
| # When generating file blobs, taking the real path is too slow on large | # When generating file blobs, taking the real path is too slow on large | ||||
| # repos, so force it to just return the linkrev directly. | # repos, so force it to just return the linkrev directly. | ||||
| repo = self._repo | repo = self._repo | ||||
| if util.safehasattr(repo, 'forcelinkrev') and repo.forcelinkrev: | if util.safehasattr(repo, 'forcelinkrev') and repo.forcelinkrev: | ||||
| return self._filelog.linkrev(self._filelog.rev(self._filenode)) | return self._filelog.linkrev(self._filelog.rev(self._filenode)) | ||||
| return orig(self, *args, **kwargs) | return orig(self, *args, **kwargs) | ||||
| for path in reversed(missingdirs): | for path in reversed(missingdirs): | ||||
| os.mkdir(path) | os.mkdir(path) | ||||
| for path in missingdirs: | for path in missingdirs: | ||||
| setstickygroupdir(path, gid, ui.warn) | setstickygroupdir(path, gid, ui.warn) | ||||
| finally: | finally: | ||||
| os.umask(oldumask) | os.umask(oldumask) | ||||
| def trygetattr(obj, names): | |||||
| """try different attribute names, return the first matched attribute, | |||||
| or raise if no names are matched. | |||||
| """ | |||||
| for name in names: | |||||
| result = getattr(obj, name, None) | |||||
| if result is not None: | |||||
| return result | |||||
| raise AttributeError | |||||
| def peercapabilities(peer): | |||||
| """return capabilities of a peer""" | |||||
| return trygetattr(peer, ('_capabilities', 'capabilities'))() | |||||
| $ cd .. | $ cd .. | ||||
| $ cat hg1.pid >> $DAEMON_PIDS | $ cat hg1.pid >> $DAEMON_PIDS | ||||
| $ hgcloneshallow http://localhost:$HGPORT/ shallow -q | $ hgcloneshallow http://localhost:$HGPORT/ shallow -q | ||||
| 1 files fetched over 1 fetches - (1 misses, 0.00% hit ratio) over *s (glob) | 1 files fetched over 1 fetches - (1 misses, 0.00% hit ratio) over *s (glob) | ||||
| $ grep batch access.log | grep getfile | $ grep batch access.log | grep getfile | ||||
| * "GET /?cmd=batch HTTP/1.1" 200 - x-hgarg-1:cmds=getfile+*node%3D1406e74118627694268417491f018a4a883152f0* (glob) | * "GET /?cmd=batch HTTP/1.1" 200 - x-hgarg-1:cmds=getfile+*node%3D1406e74118627694268417491f018a4a883152f0* (glob) | ||||
| $ grep batch access.log | grep getfile | |||||
| * "GET /?cmd=batch HTTP/1.1" 200 - x-hgarg-1:cmds=getfile+*file%3Dx* (glob) | |||||
| Clear filenode cache so we can test fetching with a modified batch size | Clear filenode cache so we can test fetching with a modified batch size | ||||
| $ rm -r $TESTTMP/hgcache | $ rm -r $TESTTMP/hgcache | ||||
| Now do a fetch with a large batch size so we're sure it works | Now do a fetch with a large batch size so we're sure it works | ||||
| $ hgcloneshallow http://localhost:$HGPORT/ shallow-large-batch \ | $ hgcloneshallow http://localhost:$HGPORT/ shallow-large-batch \ | ||||
| > --config remotefilelog.batchsize=1000 -q | > --config remotefilelog.batchsize=1000 -q | ||||
| 1 files fetched over 1 fetches - (1 misses, 0.00% hit ratio) over *s (glob) | 1 files fetched over 1 fetches - (1 misses, 0.00% hit ratio) over *s (glob) | ||||