rawdata: update caller in remotefilelog

We update callers incrementally because this help bisecting failures. This was
useful during development, so we expect it might be useful again in the future.

hgext/remotefilelog/contentstore.py

@@ -1,376 +1,376 b''
 from __future__ import absolute_import
 
 import threading
 
 from mercurial.node import hex, nullid
 from mercurial import (
     mdiff,
     pycompat,
     revlog,
 )
 from . import (
     basestore,
     constants,
     shallowutil,
 )
 
 class ChainIndicies(object):
     """A static class for easy reference to the delta chain indicies.
     """
     # The filename of this revision delta
     NAME = 0
     # The mercurial file node for this revision delta
     NODE = 1
     # The filename of the delta base's revision. This is useful when delta
     # between different files (like in the case of a move or copy, we can delta
     # against the original file content).
     BASENAME = 2
     # The mercurial file node for the delta base revision. This is the nullid if
     # this delta is a full text.
     BASENODE = 3
     # The actual delta or full text data.
     DATA = 4
 
 class unioncontentstore(basestore.baseunionstore):
     def __init__(self, *args, **kwargs):
         super(unioncontentstore, self).__init__(*args, **kwargs)
 
         self.stores = args
         self.writestore = kwargs.get(r'writestore')
 
         # If allowincomplete==True then the union store can return partial
         # delta chains, otherwise it will throw a KeyError if a full
         # deltachain can't be found.
         self.allowincomplete = kwargs.get(r'allowincomplete', False)
 
     def get(self, name, node):
         """Fetches the full text revision contents of the given name+node pair.
         If the full text doesn't exist, throws a KeyError.
 
         Under the hood, this uses getdeltachain() across all the stores to build
         up a full chain to produce the full text.
         """
         chain = self.getdeltachain(name, node)
 
         if chain[-1][ChainIndicies.BASENODE] != nullid:
             # If we didn't receive a full chain, throw
             raise KeyError((name, hex(node)))
 
         # The last entry in the chain is a full text, so we start our delta
         # applies with that.
         fulltext = chain.pop()[ChainIndicies.DATA]
 
         text = fulltext
         while chain:
             delta = chain.pop()[ChainIndicies.DATA]
             text = mdiff.patches(text, [delta])
 
         return text
 
     @basestore.baseunionstore.retriable
     def getdelta(self, name, node):
         """Return the single delta entry for the given name/node pair.
         """
         for store in self.stores:
             try:
                 return store.getdelta(name, node)
             except KeyError:
                 pass
 
         raise KeyError((name, hex(node)))
 
     def getdeltachain(self, name, node):
         """Returns the deltachain for the given name/node pair.
 
         Returns an ordered list of:
 
           [(name, node, deltabasename, deltabasenode, deltacontent),...]
 
         where the chain is terminated by a full text entry with a nullid
         deltabasenode.
         """
         chain = self._getpartialchain(name, node)
         while chain[-1][ChainIndicies.BASENODE] != nullid:
             x, x, deltabasename, deltabasenode, x = chain[-1]
             try:
                 morechain = self._getpartialchain(deltabasename, deltabasenode)
                 chain.extend(morechain)
             except KeyError:
                 # If we allow incomplete chains, don't throw.
                 if not self.allowincomplete:
                     raise
                 break
 
         return chain
 
     @basestore.baseunionstore.retriable
     def getmeta(self, name, node):
         """Returns the metadata dict for given node."""
         for store in self.stores:
             try:
                 return store.getmeta(name, node)
             except KeyError:
                 pass
         raise KeyError((name, hex(node)))
 
     def getmetrics(self):
         metrics = [s.getmetrics() for s in self.stores]
         return shallowutil.sumdicts(*metrics)
 
     @basestore.baseunionstore.retriable
     def _getpartialchain(self, name, node):
         """Returns a partial delta chain for the given name/node pair.
 
         A partial chain is a chain that may not be terminated in a full-text.
         """
         for store in self.stores:
             try:
                 return store.getdeltachain(name, node)
             except KeyError:
                 pass
 
         raise KeyError((name, hex(node)))
 
     def add(self, name, node, data):
         raise RuntimeError("cannot add content only to remotefilelog "
                            "contentstore")
 
     def getmissing(self, keys):
         missing = keys
         for store in self.stores:
             if missing:
                 missing = store.getmissing(missing)
         return missing
 
     def addremotefilelognode(self, name, node, data):
         if self.writestore:
             self.writestore.addremotefilelognode(name, node, data)
         else:
             raise RuntimeError("no writable store configured")
 
     def markledger(self, ledger, options=None):
         for store in self.stores:
             store.markledger(ledger, options)
 
 class remotefilelogcontentstore(basestore.basestore):
     def __init__(self, *args, **kwargs):
         super(remotefilelogcontentstore, self).__init__(*args, **kwargs)
         self._threaddata = threading.local()
 
     def get(self, name, node):
         # return raw revision text
         data = self._getdata(name, node)
 
         offset, size, flags = shallowutil.parsesizeflags(data)
         content = data[offset:offset + size]
 
         ancestormap = shallowutil.ancestormap(data)
         p1, p2, linknode, copyfrom = ancestormap[node]
         copyrev = None
         if copyfrom:
             copyrev = hex(p1)
 
         self._updatemetacache(node, size, flags)
 
         # lfs tracks renames in its own metadata, remove hg copy metadata,
         # because copy metadata will be re-added by lfs flag processor.
         if flags & revlog.REVIDX_EXTSTORED:
             copyrev = copyfrom = None
         revision = shallowutil.createrevlogtext(content, copyfrom, copyrev)
         return revision
 
     def getdelta(self, name, node):
         # Since remotefilelog content stores only contain full texts, just
         # return that.
         revision = self.get(name, node)
         return revision, name, nullid, self.getmeta(name, node)
 
     def getdeltachain(self, name, node):
         # Since remotefilelog content stores just contain full texts, we return
         # a fake delta chain that just consists of a single full text revision.
         # The nullid in the deltabasenode slot indicates that the revision is a
         # fulltext.
         revision = self.get(name, node)
         return [(name, node, None, nullid, revision)]
 
     def getmeta(self, name, node):
         self._sanitizemetacache()
         if node != self._threaddata.metacache[0]:
             data = self._getdata(name, node)
             offset, size, flags = shallowutil.parsesizeflags(data)
             self._updatemetacache(node, size, flags)
         return self._threaddata.metacache[1]
 
     def add(self, name, node, data):
         raise RuntimeError("cannot add content only to remotefilelog "
                            "contentstore")
 
     def _sanitizemetacache(self):
         metacache = getattr(self._threaddata, 'metacache', None)
         if metacache is None:
             self._threaddata.metacache = (None, None) # (node, meta)
 
     def _updatemetacache(self, node, size, flags):
         self._sanitizemetacache()
         if node == self._threaddata.metacache[0]:
             return
         meta = {constants.METAKEYFLAG: flags,
                 constants.METAKEYSIZE: size}
         self._threaddata.metacache = (node, meta)
 
 class remotecontentstore(object):
     def __init__(self, ui, fileservice, shared):
         self._fileservice = fileservice
         # type(shared) is usually remotefilelogcontentstore
         self._shared = shared
 
     def get(self, name, node):
         self._fileservice.prefetch([(name, hex(node))], force=True,
                                    fetchdata=True)
         return self._shared.get(name, node)
 
     def getdelta(self, name, node):
         revision = self.get(name, node)
         return revision, name, nullid, self._shared.getmeta(name, node)
 
     def getdeltachain(self, name, node):
         # Since our remote content stores just contain full texts, we return a
         # fake delta chain that just consists of a single full text revision.
         # The nullid in the deltabasenode slot indicates that the revision is a
         # fulltext.
         revision = self.get(name, node)
         return [(name, node, None, nullid, revision)]
 
     def getmeta(self, name, node):
         self._fileservice.prefetch([(name, hex(node))], force=True,
                                    fetchdata=True)
         return self._shared.getmeta(name, node)
 
     def add(self, name, node, data):
         raise RuntimeError("cannot add to a remote store")
 
     def getmissing(self, keys):
         return keys
 
     def markledger(self, ledger, options=None):
         pass
 
 class manifestrevlogstore(object):
     def __init__(self, repo):
         self._store = repo.store
         self._svfs = repo.svfs
         self._revlogs = dict()
         self._cl = revlog.revlog(self._svfs, '00changelog.i')
         self._repackstartlinkrev = 0
 
     def get(self, name, node):
-        return self._revlog(name).revision(node, raw=True)
+        return self._revlog(name).rawdata(node)
 
     def getdelta(self, name, node):
         revision = self.get(name, node)
         return revision, name, nullid, self.getmeta(name, node)
 
     def getdeltachain(self, name, node):
         revision = self.get(name, node)
         return [(name, node, None, nullid, revision)]
 
     def getmeta(self, name, node):
         rl = self._revlog(name)
         rev = rl.rev(node)
         return {constants.METAKEYFLAG: rl.flags(rev),
                 constants.METAKEYSIZE: rl.rawsize(rev)}
 
     def getancestors(self, name, node, known=None):
         if known is None:
             known = set()
         if node in known:
             return []
 
         rl = self._revlog(name)
         ancestors = {}
         missing = set((node,))
         for ancrev in rl.ancestors([rl.rev(node)], inclusive=True):
             ancnode = rl.node(ancrev)
             missing.discard(ancnode)
 
             p1, p2 = rl.parents(ancnode)
             if p1 != nullid and p1 not in known:
                 missing.add(p1)
             if p2 != nullid and p2 not in known:
                 missing.add(p2)
 
             linknode = self._cl.node(rl.linkrev(ancrev))
             ancestors[rl.node(ancrev)] = (p1, p2, linknode, '')
             if not missing:
                 break
         return ancestors
 
     def getnodeinfo(self, name, node):
         cl = self._cl
         rl = self._revlog(name)
         parents = rl.parents(node)
         linkrev = rl.linkrev(rl.rev(node))
         return (parents[0], parents[1], cl.node(linkrev), None)
 
     def add(self, *args):
         raise RuntimeError("cannot add to a revlog store")
 
     def _revlog(self, name):
         rl = self._revlogs.get(name)
         if rl is None:
             revlogname = '00manifesttree.i'
             if name != '':
                 revlogname = 'meta/%s/00manifest.i' % name
             rl = revlog.revlog(self._svfs, revlogname)
             self._revlogs[name] = rl
         return rl
 
     def getmissing(self, keys):
         missing = []
         for name, node in keys:
             mfrevlog = self._revlog(name)
             if node not in mfrevlog.nodemap:
                 missing.append((name, node))
 
         return missing
 
     def setrepacklinkrevrange(self, startrev, endrev):
         self._repackstartlinkrev = startrev
         self._repackendlinkrev = endrev
 
     def markledger(self, ledger, options=None):
         if options and options.get(constants.OPTION_PACKSONLY):
             return
         treename = ''
         rl = revlog.revlog(self._svfs, '00manifesttree.i')
         startlinkrev = self._repackstartlinkrev
         endlinkrev = self._repackendlinkrev
         for rev in pycompat.xrange(len(rl) - 1, -1, -1):
             linkrev = rl.linkrev(rev)
             if linkrev < startlinkrev:
                 break
             if linkrev > endlinkrev:
                 continue
             node = rl.node(rev)
             ledger.markdataentry(self, treename, node)
             ledger.markhistoryentry(self, treename, node)
 
         for path, encoded, size in self._store.datafiles():
             if path[:5] != 'meta/' or path[-2:] != '.i':
                 continue
 
             treename = path[5:-len('/00manifest.i')]
 
             rl = revlog.revlog(self._svfs, path)
             for rev in pycompat.xrange(len(rl) - 1, -1, -1):
                 linkrev = rl.linkrev(rev)
                 if linkrev < startlinkrev:
                     break
                 if linkrev > endlinkrev:
                     continue
                 node = rl.node(rev)
                 ledger.markdataentry(self, treename, node)
                 ledger.markhistoryentry(self, treename, node)
 
     def cleanup(self, ledger):
         pass

hgext/remotefilelog/fileserverclient.py

@@ -1,584 +1,584 b''
 # fileserverclient.py - client for communicating with the cache process
 #
 # Copyright 2013 Facebook, Inc.
 #
 # 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 hashlib
 import io
 import os
 import threading
 import time
 import zlib
 
 from mercurial.i18n import _
 from mercurial.node import bin, hex, nullid
 from mercurial import (
     error,
     node,
     pycompat,
     revlog,
     sshpeer,
     util,
     wireprotov1peer,
 )
 from mercurial.utils import procutil
 
 from . import (
     constants,
     contentstore,
     metadatastore,
 )
 
 _sshv1peer = sshpeer.sshv1peer
 
 # Statistics for debugging
 fetchcost = 0
 fetches = 0
 fetched = 0
 fetchmisses = 0
 
 _lfsmod = None
 
 def getcachekey(reponame, file, id):
     pathhash = node.hex(hashlib.sha1(file).digest())
     return os.path.join(reponame, pathhash[:2], pathhash[2:], id)
 
 def getlocalkey(file, id):
     pathhash = node.hex(hashlib.sha1(file).digest())
     return os.path.join(pathhash, id)
 
 def peersetup(ui, peer):
 
     class remotefilepeer(peer.__class__):
         @wireprotov1peer.batchable
         def x_rfl_getfile(self, file, node):
             if not self.capable('x_rfl_getfile'):
                 raise error.Abort(
                     'configured remotefile server does not support getfile')
             f = wireprotov1peer.future()
             yield {'file': file, 'node': node}, f
             code, data = f.value.split('\0', 1)
             if int(code):
                 raise error.LookupError(file, node, data)
             yield data
 
         @wireprotov1peer.batchable
         def x_rfl_getflogheads(self, path):
             if not self.capable('x_rfl_getflogheads'):
                 raise error.Abort('configured remotefile server does not '
                                   'support getflogheads')
             f = wireprotov1peer.future()
             yield {'path': path}, f
             heads = f.value.split('\n') if f.value else []
             yield heads
 
         def _updatecallstreamopts(self, command, opts):
             if command != 'getbundle':
                 return
             if (constants.NETWORK_CAP_LEGACY_SSH_GETFILES
                 not in self.capabilities()):
                 return
             if not util.safehasattr(self, '_localrepo'):
                 return
             if (constants.SHALLOWREPO_REQUIREMENT
                 not in self._localrepo.requirements):
                 return
 
             bundlecaps = opts.get('bundlecaps')
             if bundlecaps:
                 bundlecaps = [bundlecaps]
             else:
                 bundlecaps = []
 
             # shallow, includepattern, and excludepattern are a hacky way of
             # carrying over data from the local repo to this getbundle
             # command. We need to do it this way because bundle1 getbundle
             # doesn't provide any other place we can hook in to manipulate
             # getbundle args before it goes across the wire. Once we get rid
             # of bundle1, we can use bundle2's _pullbundle2extraprepare to
             # do this more cleanly.
             bundlecaps.append(constants.BUNDLE2_CAPABLITY)
             if self._localrepo.includepattern:
                 patterns = '\0'.join(self._localrepo.includepattern)
                 includecap = "includepattern=" + patterns
                 bundlecaps.append(includecap)
             if self._localrepo.excludepattern:
                 patterns = '\0'.join(self._localrepo.excludepattern)
                 excludecap = "excludepattern=" + patterns
                 bundlecaps.append(excludecap)
             opts['bundlecaps'] = ','.join(bundlecaps)
 
         def _sendrequest(self, command, args, **opts):
             self._updatecallstreamopts(command, args)
             return super(remotefilepeer, self)._sendrequest(command, args,
                                                             **opts)
 
         def _callstream(self, command, **opts):
             supertype = super(remotefilepeer, self)
             if not util.safehasattr(supertype, '_sendrequest'):
                 self._updatecallstreamopts(command, pycompat.byteskwargs(opts))
             return super(remotefilepeer, self)._callstream(command, **opts)
 
     peer.__class__ = remotefilepeer
 
 class cacheconnection(object):
     """The connection for communicating with the remote cache. Performs
     gets and sets by communicating with an external process that has the
     cache-specific implementation.
     """
     def __init__(self):
         self.pipeo = self.pipei = self.pipee = None
         self.subprocess = None
         self.connected = False
 
     def connect(self, cachecommand):
         if self.pipeo:
             raise error.Abort(_("cache connection already open"))
         self.pipei, self.pipeo, self.pipee, self.subprocess = (
             procutil.popen4(cachecommand))
         self.connected = True
 
     def close(self):
         def tryclose(pipe):
             try:
                 pipe.close()
             except Exception:
                 pass
         if self.connected:
             try:
                 self.pipei.write("exit\n")
             except Exception:
                 pass
             tryclose(self.pipei)
             self.pipei = None
             tryclose(self.pipeo)
             self.pipeo = None
             tryclose(self.pipee)
             self.pipee = None
             try:
                 # Wait for process to terminate, making sure to avoid deadlock.
                 # See https://docs.python.org/2/library/subprocess.html for
                 # warnings about wait() and deadlocking.
                 self.subprocess.communicate()
             except Exception:
                 pass
             self.subprocess = None
         self.connected = False
 
     def request(self, request, flush=True):
         if self.connected:
             try:
                 self.pipei.write(request)
                 if flush:
                     self.pipei.flush()
             except IOError:
                 self.close()
 
     def receiveline(self):
         if not self.connected:
             return None
         try:
             result = self.pipeo.readline()[:-1]
             if not result:
                 self.close()
         except IOError:
             self.close()
 
         return result
 
 def _getfilesbatch(
         remote, receivemissing, progresstick, missed, idmap, batchsize):
     # Over http(s), iterbatch is a streamy method and we can start
     # looking at results early. This means we send one (potentially
     # large) request, but then we show nice progress as we process
     # file results, rather than showing chunks of $batchsize in
     # progress.
     #
     # Over ssh, iterbatch isn't streamy because batch() wasn't
     # explicitly designed as a streaming method. In the future we
     # should probably introduce a streambatch() method upstream and
     # use that for this.
     with remote.commandexecutor() as e:
         futures = []
         for m in missed:
             futures.append(e.callcommand('x_rfl_getfile', {
                 'file': idmap[m],
                 'node': m[-40:]
             }))
 
         for i, m in enumerate(missed):
             r = futures[i].result()
             futures[i] = None  # release memory
             file_ = idmap[m]
             node = m[-40:]
             receivemissing(io.BytesIO('%d\n%s' % (len(r), r)), file_, node)
             progresstick()
 
 def _getfiles_optimistic(
     remote, receivemissing, progresstick, missed, idmap, step):
     remote._callstream("x_rfl_getfiles")
     i = 0
     pipeo = remote._pipeo
     pipei = remote._pipei
     while i < len(missed):
         # issue a batch of requests
         start = i
         end = min(len(missed), start + step)
         i = end
         for missingid in missed[start:end]:
             # issue new request
             versionid = missingid[-40:]
             file = idmap[missingid]
             sshrequest = "%s%s\n" % (versionid, file)
             pipeo.write(sshrequest)
         pipeo.flush()
 
         # receive batch results
         for missingid in missed[start:end]:
             versionid = missingid[-40:]
             file = idmap[missingid]
             receivemissing(pipei, file, versionid)
             progresstick()
 
     # End the command
     pipeo.write('\n')
     pipeo.flush()
 
 def _getfiles_threaded(
     remote, receivemissing, progresstick, missed, idmap, step):
     remote._callstream("getfiles")
     pipeo = remote._pipeo
     pipei = remote._pipei
 
     def writer():
         for missingid in missed:
             versionid = missingid[-40:]
             file = idmap[missingid]
             sshrequest = "%s%s\n" % (versionid, file)
             pipeo.write(sshrequest)
         pipeo.flush()
     writerthread = threading.Thread(target=writer)
     writerthread.daemon = True
     writerthread.start()
 
     for missingid in missed:
         versionid = missingid[-40:]
         file = idmap[missingid]
         receivemissing(pipei, file, versionid)
         progresstick()
 
     writerthread.join()
     # End the command
     pipeo.write('\n')
     pipeo.flush()
 
 class fileserverclient(object):
     """A client for requesting files from the remote file server.
     """
     def __init__(self, repo):
         ui = repo.ui
         self.repo = repo
         self.ui = ui
         self.cacheprocess = ui.config("remotefilelog", "cacheprocess")
         if self.cacheprocess:
             self.cacheprocess = util.expandpath(self.cacheprocess)
 
         # This option causes remotefilelog to pass the full file path to the
         # cacheprocess instead of a hashed key.
         self.cacheprocesspasspath = ui.configbool(
             "remotefilelog", "cacheprocess.includepath")
 
         self.debugoutput = ui.configbool("remotefilelog", "debug")
 
         self.remotecache = cacheconnection()
 
     def setstore(self, datastore, historystore, writedata, writehistory):
         self.datastore = datastore
         self.historystore = historystore
         self.writedata = writedata
         self.writehistory = writehistory
 
     def _connect(self):
         return self.repo.connectionpool.get(self.repo.fallbackpath)
 
     def request(self, fileids):
         """Takes a list of filename/node pairs and fetches them from the
         server. Files are stored in the local cache.
         A list of nodes that the server couldn't find is returned.
         If the connection fails, an exception is raised.
         """
         if not self.remotecache.connected:
             self.connect()
         cache = self.remotecache
         writedata = self.writedata
 
         repo = self.repo
         total = len(fileids)
         request = "get\n%d\n" % total
         idmap = {}
         reponame = repo.name
         for file, id in fileids:
             fullid = getcachekey(reponame, file, id)
             if self.cacheprocesspasspath:
                 request += file + '\0'
             request += fullid + "\n"
             idmap[fullid] = file
 
         cache.request(request)
 
         progress = self.ui.makeprogress(_('downloading'), total=total)
         progress.update(0)
 
         missed = []
         while True:
             missingid = cache.receiveline()
             if not missingid:
                 missedset = set(missed)
                 for missingid in idmap:
                     if not missingid in missedset:
                         missed.append(missingid)
                 self.ui.warn(_("warning: cache connection closed early - " +
                     "falling back to server\n"))
                 break
             if missingid == "0":
                 break
             if missingid.startswith("_hits_"):
                 # receive progress reports
                 parts = missingid.split("_")
                 progress.increment(int(parts[2]))
                 continue
 
             missed.append(missingid)
 
         global fetchmisses
         fetchmisses += len(missed)
 
         fromcache = total - len(missed)
         progress.update(fromcache, total=total)
         self.ui.log("remotefilelog", "remote cache hit rate is %r of %r\n",
                     fromcache, total, hit=fromcache, total=total)
 
         oldumask = os.umask(0o002)
         try:
             # receive cache misses from master
             if missed:
                 # When verbose is true, sshpeer prints 'running ssh...'
                 # to stdout, which can interfere with some command
                 # outputs
                 verbose = self.ui.verbose
                 self.ui.verbose = False
                 try:
                     with self._connect() as conn:
                         remote = conn.peer
                         if remote.capable(
                                 constants.NETWORK_CAP_LEGACY_SSH_GETFILES):
                             if not isinstance(remote, _sshv1peer):
                                 raise error.Abort('remotefilelog requires ssh '
                                                   'servers')
                             step = self.ui.configint('remotefilelog',
                                                      'getfilesstep')
                             getfilestype = self.ui.config('remotefilelog',
                                                           'getfilestype')
                             if getfilestype == 'threaded':
                                 _getfiles = _getfiles_threaded
                             else:
                                 _getfiles = _getfiles_optimistic
                             _getfiles(remote, self.receivemissing,
                                       progress.increment, missed, idmap, step)
                         elif remote.capable("x_rfl_getfile"):
                             if remote.capable('batch'):
                                 batchdefault = 100
                             else:
                                 batchdefault = 10
                             batchsize = self.ui.configint(
                                 'remotefilelog', 'batchsize', batchdefault)
                             self.ui.debug(
                                 b'requesting %d files from '
                                 b'remotefilelog server...\n' % len(missed))
                             _getfilesbatch(
                                 remote, self.receivemissing, progress.increment,
                                 missed, idmap, batchsize)
                         else:
                             raise error.Abort("configured remotefilelog server"
                                              " does not support remotefilelog")
 
                     self.ui.log("remotefilefetchlog",
                                 "Success\n",
                                 fetched_files = progress.pos - fromcache,
                                 total_to_fetch = total - fromcache)
                 except Exception:
                     self.ui.log("remotefilefetchlog",
                                 "Fail\n",
                                 fetched_files = progress.pos - fromcache,
                                 total_to_fetch = total - fromcache)
                     raise
                 finally:
                     self.ui.verbose = verbose
                 # send to memcache
                 request = "set\n%d\n%s\n" % (len(missed), "\n".join(missed))
                 cache.request(request)
 
             progress.complete()
 
             # mark ourselves as a user of this cache
             writedata.markrepo(self.repo.path)
         finally:
             os.umask(oldumask)
 
     def receivemissing(self, pipe, filename, node):
         line = pipe.readline()[:-1]
         if not line:
             raise error.ResponseError(_("error downloading file contents:"),
                                       _("connection closed early"))
         size = int(line)
         data = pipe.read(size)
         if len(data) != size:
             raise error.ResponseError(_("error downloading file contents:"),
                                       _("only received %s of %s bytes")
                                       % (len(data), size))
 
         self.writedata.addremotefilelognode(filename, bin(node),
                                              zlib.decompress(data))
 
     def connect(self):
         if self.cacheprocess:
             cmd = "%s %s" % (self.cacheprocess, self.writedata._path)
             self.remotecache.connect(cmd)
         else:
             # If no cache process is specified, we fake one that always
             # returns cache misses.  This enables tests to run easily
             # and may eventually allow us to be a drop in replacement
             # for the largefiles extension.
             class simplecache(object):
                 def __init__(self):
                     self.missingids = []
                     self.connected = True
 
                 def close(self):
                     pass
 
                 def request(self, value, flush=True):
                     lines = value.split("\n")
                     if lines[0] != "get":
                         return
                     self.missingids = lines[2:-1]
                     self.missingids.append('0')
 
                 def receiveline(self):
                     if len(self.missingids) > 0:
                         return self.missingids.pop(0)
                     return None
 
             self.remotecache = simplecache()
 
     def close(self):
         if fetches:
             msg = ("%d files fetched over %d fetches - " +
                    "(%d misses, %0.2f%% hit ratio) over %0.2fs\n") % (
                        fetched,
                        fetches,
                        fetchmisses,
                        float(fetched - fetchmisses) / float(fetched) * 100.0,
                        fetchcost)
             if self.debugoutput:
                 self.ui.warn(msg)
             self.ui.log("remotefilelog.prefetch", msg.replace("%", "%%"),
                 remotefilelogfetched=fetched,
                 remotefilelogfetches=fetches,
                 remotefilelogfetchmisses=fetchmisses,
                 remotefilelogfetchtime=fetchcost * 1000)
 
         if self.remotecache.connected:
             self.remotecache.close()
 
     def prefetch(self, fileids, force=False, fetchdata=True,
                  fetchhistory=False):
         """downloads the given file versions to the cache
         """
         repo = self.repo
         idstocheck = []
         for file, id in fileids:
             # hack
             # - we don't use .hgtags
             # - workingctx produces ids with length 42,
             #   which we skip since they aren't in any cache
             if (file == '.hgtags' or len(id) == 42
                 or not repo.shallowmatch(file)):
                 continue
 
             idstocheck.append((file, bin(id)))
 
         datastore = self.datastore
         historystore = self.historystore
         if force:
             datastore = contentstore.unioncontentstore(*repo.shareddatastores)
             historystore = metadatastore.unionmetadatastore(
                 *repo.sharedhistorystores)
 
         missingids = set()
         if fetchdata:
             missingids.update(datastore.getmissing(idstocheck))
         if fetchhistory:
             missingids.update(historystore.getmissing(idstocheck))
 
         # partition missing nodes into nullid and not-nullid so we can
         # warn about this filtering potentially shadowing bugs.
         nullids = len([None for unused, id in missingids if id == nullid])
         if nullids:
             missingids = [(f, id) for f, id in missingids if id != nullid]
             repo.ui.develwarn(
                 ('remotefilelog not fetching %d null revs'
                  ' - this is likely hiding bugs' % nullids),
                 config='remotefilelog-ext')
         if missingids:
             global fetches, fetched, fetchcost
             fetches += 1
 
             # We want to be able to detect excess individual file downloads, so
             # let's log that information for debugging.
             if fetches >= 15 and fetches < 18:
                 if fetches == 15:
                     fetchwarning = self.ui.config('remotefilelog',
                                                   'fetchwarning')
                     if fetchwarning:
                         self.ui.warn(fetchwarning + '\n')
                 self.logstacktrace()
             missingids = [(file, hex(id)) for file, id in sorted(missingids)]
             fetched += len(missingids)
             start = time.time()
             missingids = self.request(missingids)
             if missingids:
                 raise error.Abort(_("unable to download %d files") %
                                   len(missingids))
             fetchcost += time.time() - start
             self._lfsprefetch(fileids)
 
     def _lfsprefetch(self, fileids):
         if not _lfsmod or not util.safehasattr(
                 self.repo.svfs, 'lfslocalblobstore'):
             return
         if not _lfsmod.wrapper.candownload(self.repo):
             return
         pointers = []
         store = self.repo.svfs.lfslocalblobstore
         for file, id in fileids:
             node = bin(id)
             rlog = self.repo.file(file)
             if rlog.flags(node) & revlog.REVIDX_EXTSTORED:
-                text = rlog.revision(node, raw=True)
+                text = rlog.rawdata(node)
                 p = _lfsmod.pointer.deserialize(text)
                 oid = p.oid()
                 if not store.has(oid):
                     pointers.append(p)
         if len(pointers) > 0:
             self.repo.svfs.lfsremoteblobstore.readbatch(pointers, store)
             assert all(store.has(p.oid()) for p in pointers)
 
     def logstacktrace(self):
         import traceback
         self.ui.log('remotefilelog', 'excess remotefilelog fetching:\n%s\n',
                     ''.join(traceback.format_stack()))

hgext/remotefilelog/remotefilelog.py

@@ -1,462 +1,462 b''
 # remotefilelog.py - filelog implementation where filelog history is stored
 #                    remotely
 #
 # Copyright 2013 Facebook, Inc.
 #
 # 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 collections
 import os
 
 from mercurial.node import (
     bin,
     nullid,
     wdirfilenodeids,
     wdirid,
 )
 from mercurial.i18n import _
 from mercurial import (
     ancestor,
     error,
     mdiff,
     revlog,
 )
 from mercurial.utils import storageutil
 
 from . import (
     constants,
     fileserverclient,
     shallowutil,
 )
 
 class remotefilelognodemap(object):
     def __init__(self, filename, store):
         self._filename = filename
         self._store = store
 
     def __contains__(self, node):
         missing = self._store.getmissing([(self._filename, node)])
         return not bool(missing)
 
     def __get__(self, node):
         if node not in self:
             raise KeyError(node)
         return node
 
 class remotefilelog(object):
 
     _generaldelta = True
 
     def __init__(self, opener, path, repo):
         self.opener = opener
         self.filename = path
         self.repo = repo
         self.nodemap = remotefilelognodemap(self.filename, repo.contentstore)
 
         self.version = 1
 
     def read(self, node):
         """returns the file contents at this node"""
         t = self.revision(node)
         if not t.startswith('\1\n'):
             return t
         s = t.index('\1\n', 2)
         return t[s + 2:]
 
     def add(self, text, meta, transaction, linknode, p1=None, p2=None):
         # hash with the metadata, like in vanilla filelogs
         hashtext = shallowutil.createrevlogtext(text, meta.get('copy'),
                                                 meta.get('copyrev'))
         node = storageutil.hashrevisionsha1(hashtext, p1, p2)
         return self.addrevision(hashtext, transaction, linknode, p1, p2,
                                 node=node)
 
     def _createfileblob(self, text, meta, flags, p1, p2, node, linknode):
         # text passed to "_createfileblob" does not include filelog metadata
         header = shallowutil.buildfileblobheader(len(text), flags)
         data = "%s\0%s" % (header, text)
 
         realp1 = p1
         copyfrom = ""
         if meta and 'copy' in meta:
             copyfrom = meta['copy']
             realp1 = bin(meta['copyrev'])
 
         data += "%s%s%s%s%s\0" % (node, realp1, p2, linknode, copyfrom)
 
         visited = set()
 
         pancestors = {}
         queue = []
         if realp1 != nullid:
             p1flog = self
             if copyfrom:
                 p1flog = remotefilelog(self.opener, copyfrom, self.repo)
 
             pancestors.update(p1flog.ancestormap(realp1))
             queue.append(realp1)
             visited.add(realp1)
         if p2 != nullid:
             pancestors.update(self.ancestormap(p2))
             queue.append(p2)
             visited.add(p2)
 
         ancestortext = ""
 
         # add the ancestors in topological order
         while queue:
             c = queue.pop(0)
             pa1, pa2, ancestorlinknode, pacopyfrom = pancestors[c]
 
             pacopyfrom = pacopyfrom or ''
             ancestortext += "%s%s%s%s%s\0" % (
                 c, pa1, pa2, ancestorlinknode, pacopyfrom)
 
             if pa1 != nullid and pa1 not in visited:
                 queue.append(pa1)
                 visited.add(pa1)
             if pa2 != nullid and pa2 not in visited:
                 queue.append(pa2)
                 visited.add(pa2)
 
         data += ancestortext
 
         return data
 
     def addrevision(self, text, transaction, linknode, p1, p2, cachedelta=None,
                     node=None, flags=revlog.REVIDX_DEFAULT_FLAGS):
         # text passed to "addrevision" includes hg filelog metadata header
         if node is None:
             node = storageutil.hashrevisionsha1(text, p1, p2)
 
         meta, metaoffset = storageutil.parsemeta(text)
         rawtext, validatehash = self._processflags(text, flags, 'write')
         return self.addrawrevision(rawtext, transaction, linknode, p1, p2,
                                    node, flags, cachedelta,
                                    _metatuple=(meta, metaoffset))
 
     def addrawrevision(self, rawtext, transaction, linknode, p1, p2, node,
                        flags, cachedelta=None, _metatuple=None):
         if _metatuple:
             # _metatuple: used by "addrevision" internally by remotefilelog
             # meta was parsed confidently
             meta, metaoffset = _metatuple
         else:
             # not from self.addrevision, but something else (repo._filecommit)
             # calls addrawrevision directly. remotefilelog needs to get and
             # strip filelog metadata.
             # we don't have confidence about whether rawtext contains filelog
             # metadata or not (flag processor could replace it), so we just
             # parse it as best-effort.
             # in LFS (flags != 0)'s case, the best way is to call LFS code to
             # get the meta information, instead of storageutil.parsemeta.
             meta, metaoffset = storageutil.parsemeta(rawtext)
         if flags != 0:
             # when flags != 0, be conservative and do not mangle rawtext, since
             # a read flag processor expects the text not being mangled at all.
             metaoffset = 0
         if metaoffset:
             # remotefilelog fileblob stores copy metadata in its ancestortext,
             # not its main blob. so we need to remove filelog metadata
             # (containing copy information) from text.
             blobtext = rawtext[metaoffset:]
         else:
             blobtext = rawtext
         data = self._createfileblob(blobtext, meta, flags, p1, p2, node,
                                     linknode)
         self.repo.contentstore.addremotefilelognode(self.filename, node, data)
 
         return node
 
     def renamed(self, node):
         ancestors = self.repo.metadatastore.getancestors(self.filename, node)
         p1, p2, linknode, copyfrom = ancestors[node]
         if copyfrom:
             return (copyfrom, p1)
 
         return False
 
     def size(self, node):
         """return the size of a given revision"""
         return len(self.read(node))
 
     rawsize = size
 
     def cmp(self, node, text):
         """compare text with a given file revision
 
         returns True if text is different than what is stored.
         """
 
         if node == nullid:
             return True
 
         nodetext = self.read(node)
         return nodetext != text
 
     def __nonzero__(self):
         return True
 
     __bool__ = __nonzero__
 
     def __len__(self):
         if self.filename == '.hgtags':
             # The length of .hgtags is used to fast path tag checking.
             # remotefilelog doesn't support .hgtags since the entire .hgtags
             # history is needed.  Use the excludepattern setting to make
             # .hgtags a normal filelog.
             return 0
 
         raise RuntimeError("len not supported")
 
     def empty(self):
         return False
 
     def flags(self, node):
         if isinstance(node, int):
             raise error.ProgrammingError(
                 'remotefilelog does not accept integer rev for flags')
         store = self.repo.contentstore
         return store.getmeta(self.filename, node).get(constants.METAKEYFLAG, 0)
 
     def parents(self, node):
         if node == nullid:
             return nullid, nullid
 
         ancestormap = self.repo.metadatastore.getancestors(self.filename, node)
         p1, p2, linknode, copyfrom = ancestormap[node]
         if copyfrom:
             p1 = nullid
 
         return p1, p2
 
     def parentrevs(self, rev):
         # TODO(augie): this is a node and should be a rev, but for now
         # nothing in core seems to actually break.
         return self.parents(rev)
 
     def linknode(self, node):
         ancestormap = self.repo.metadatastore.getancestors(self.filename, node)
         p1, p2, linknode, copyfrom = ancestormap[node]
         return linknode
 
     def linkrev(self, node):
         return self.repo.unfiltered().changelog.rev(self.linknode(node))
 
     def emitrevisions(self, nodes, nodesorder=None, revisiondata=False,
                       assumehaveparentrevisions=False, deltaprevious=False,
                       deltamode=None):
         # we don't use any of these parameters here
         del nodesorder, revisiondata, assumehaveparentrevisions, deltaprevious
         del deltamode
         prevnode = None
         for node in nodes:
             p1, p2 = self.parents(node)
             if prevnode is None:
                 basenode = prevnode = p1
             if basenode == node:
                 basenode = nullid
             if basenode != nullid:
                 revision = None
                 delta = self.revdiff(basenode, node)
             else:
-                revision = self.revision(node, raw=True)
+                revision = self.rawdata(node)
                 delta = None
             yield revlog.revlogrevisiondelta(
                 node=node,
                 p1node=p1,
                 p2node=p2,
                 linknode=self.linknode(node),
                 basenode=basenode,
                 flags=self.flags(node),
                 baserevisionsize=None,
                 revision=revision,
                 delta=delta,
                 )
 
     def revdiff(self, node1, node2):
-        return mdiff.textdiff(self.revision(node1, raw=True),
-                              self.revision(node2, raw=True))
+        return mdiff.textdiff(self.rawdata(node1),
+                              self.rawdata(node2))
 
     def lookup(self, node):
         if len(node) == 40:
             node = bin(node)
         if len(node) != 20:
             raise error.LookupError(node, self.filename,
                                     _('invalid lookup input'))
 
         return node
 
     def rev(self, node):
         # This is a hack to make TortoiseHG work.
         return node
 
     def node(self, rev):
         # This is a hack.
         if isinstance(rev, int):
             raise error.ProgrammingError(
                 'remotefilelog does not convert integer rev to node')
         return rev
 
     def revision(self, node, raw=False):
         """returns the revlog contents at this node.
         this includes the meta data traditionally included in file revlogs.
         this is generally only used for bundling and communicating with vanilla
         hg clients.
         """
         if node == nullid:
             return ""
         if len(node) != 20:
             raise error.LookupError(node, self.filename,
                                     _('invalid revision input'))
         if node == wdirid or node in wdirfilenodeids:
             raise error.WdirUnsupported
 
         store = self.repo.contentstore
         rawtext = store.get(self.filename, node)
         if raw:
             return rawtext
         flags = store.getmeta(self.filename, node).get(constants.METAKEYFLAG, 0)
         if flags == 0:
             return rawtext
         text, verifyhash = self._processflags(rawtext, flags, 'read')
         return text
 
     def rawdata(self, node):
         return self.revision(node, raw=False)
 
     def _processflags(self, text, flags, operation, raw=False):
         # mostly copied from hg/mercurial/revlog.py
         validatehash = True
         orderedflags = revlog.REVIDX_FLAGS_ORDER
         if operation == 'write':
             orderedflags = reversed(orderedflags)
         for flag in orderedflags:
             if flag & flags:
                 vhash = True
                 if flag not in revlog._flagprocessors:
                     message = _("missing processor for flag '%#x'") % (flag)
                     raise revlog.RevlogError(message)
                 readfunc, writefunc, rawfunc = revlog._flagprocessors[flag]
                 if raw:
                     vhash = rawfunc(self, text)
                 elif operation == 'read':
                     text, vhash = readfunc(self, text)
                 elif operation == 'write':
                     text, vhash = writefunc(self, text)
                 validatehash = validatehash and vhash
         return text, validatehash
 
     def _read(self, id):
         """reads the raw file blob from disk, cache, or server"""
         fileservice = self.repo.fileservice
         localcache = fileservice.localcache
         cachekey = fileserverclient.getcachekey(self.repo.name, self.filename,
                                                 id)
         try:
             return localcache.read(cachekey)
         except KeyError:
             pass
 
         localkey = fileserverclient.getlocalkey(self.filename, id)
         localpath = os.path.join(self.localpath, localkey)
         try:
             return shallowutil.readfile(localpath)
         except IOError:
             pass
 
         fileservice.prefetch([(self.filename, id)])
         try:
             return localcache.read(cachekey)
         except KeyError:
             pass
 
         raise error.LookupError(id, self.filename, _('no node'))
 
     def ancestormap(self, node):
         return self.repo.metadatastore.getancestors(self.filename, node)
 
     def ancestor(self, a, b):
         if a == nullid or b == nullid:
             return nullid
 
         revmap, parentfunc = self._buildrevgraph(a, b)
         nodemap = dict(((v, k) for (k, v) in revmap.iteritems()))
 
         ancs = ancestor.ancestors(parentfunc, revmap[a], revmap[b])
         if ancs:
             # choose a consistent winner when there's a tie
             return min(map(nodemap.__getitem__, ancs))
         return nullid
 
     def commonancestorsheads(self, a, b):
         """calculate all the heads of the common ancestors of nodes a and b"""
 
         if a == nullid or b == nullid:
             return nullid
 
         revmap, parentfunc = self._buildrevgraph(a, b)
         nodemap = dict(((v, k) for (k, v) in revmap.iteritems()))
 
         ancs = ancestor.commonancestorsheads(parentfunc, revmap[a], revmap[b])
         return map(nodemap.__getitem__, ancs)
 
     def _buildrevgraph(self, a, b):
         """Builds a numeric revision graph for the given two nodes.
         Returns a node->rev map and a rev->[revs] parent function.
         """
         amap = self.ancestormap(a)
         bmap = self.ancestormap(b)
 
         # Union the two maps
         parentsmap = collections.defaultdict(list)
         allparents = set()
         for mapping in (amap, bmap):
             for node, pdata in mapping.iteritems():
                 parents = parentsmap[node]
                 p1, p2, linknode, copyfrom = pdata
                 # Don't follow renames (copyfrom).
                 # remotefilectx.ancestor does that.
                 if p1 != nullid and not copyfrom:
                     parents.append(p1)
                     allparents.add(p1)
                 if p2 != nullid:
                     parents.append(p2)
                     allparents.add(p2)
 
         # Breadth first traversal to build linkrev graph
         parentrevs = collections.defaultdict(list)
         revmap = {}
         queue = collections.deque(((None, n) for n in parentsmap
                  if n not in allparents))
         while queue:
             prevrev, current = queue.pop()
             if current in revmap:
                 if prevrev:
                     parentrevs[prevrev].append(revmap[current])
                 continue
 
             # Assign linkrevs in reverse order, so start at
             # len(parentsmap) and work backwards.
             currentrev = len(parentsmap) - len(revmap) - 1
             revmap[current] = currentrev
 
             if prevrev:
                 parentrevs[prevrev].append(currentrev)
 
             for parent in parentsmap.get(current):
                 queue.appendleft((currentrev, parent))
 
         return revmap, parentrevs.__getitem__
 
     def strip(self, minlink, transaction):
         pass
 
     # misc unused things
     def files(self):
         return []
 
     def checksize(self):
         return 0, 0

hgext/remotefilelog/remotefilelogserver.py

@@ -1,404 +1,404 b''
 # remotefilelogserver.py - server logic for a remotefilelog server
 #
 # Copyright 2013 Facebook, Inc.
 #
 # 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 errno
 import os
 import stat
 import time
 import zlib
 
 from mercurial.i18n import _
 from mercurial.node import bin, hex, nullid
 from mercurial import (
     changegroup,
     changelog,
     context,
     error,
     extensions,
     match,
     store,
     streamclone,
     util,
     wireprotoserver,
     wireprototypes,
     wireprotov1server,
 )
 from .  import (
     constants,
     shallowutil,
 )
 
 _sshv1server = wireprotoserver.sshv1protocolhandler
 
 def setupserver(ui, repo):
     """Sets up a normal Mercurial repo so it can serve files to shallow repos.
     """
     onetimesetup(ui)
 
     # don't send files to shallow clients during pulls
     def generatefiles(orig, self, changedfiles, linknodes, commonrevs, source,
                       *args, **kwargs):
         caps = self._bundlecaps or []
         if constants.BUNDLE2_CAPABLITY in caps:
             # only send files that don't match the specified patterns
             includepattern = None
             excludepattern = None
             for cap in (self._bundlecaps or []):
                 if cap.startswith("includepattern="):
                     includepattern = cap[len("includepattern="):].split('\0')
                 elif cap.startswith("excludepattern="):
                     excludepattern = cap[len("excludepattern="):].split('\0')
 
             m = match.always()
             if includepattern or excludepattern:
                 m = match.match(repo.root, '', None,
                     includepattern, excludepattern)
 
             changedfiles = list([f for f in changedfiles if not m(f)])
         return orig(self, changedfiles, linknodes, commonrevs, source,
                     *args, **kwargs)
 
     extensions.wrapfunction(
         changegroup.cgpacker, 'generatefiles', generatefiles)
 
 onetime = False
 def onetimesetup(ui):
     """Configures the wireprotocol for both clients and servers.
     """
     global onetime
     if onetime:
         return
     onetime = True
 
     # support file content requests
     wireprotov1server.wireprotocommand(
         'x_rfl_getflogheads', 'path', permission='pull')(getflogheads)
     wireprotov1server.wireprotocommand(
         'x_rfl_getfiles', '', permission='pull')(getfiles)
     wireprotov1server.wireprotocommand(
         'x_rfl_getfile', 'file node', permission='pull')(getfile)
 
     class streamstate(object):
         match = None
         shallowremote = False
         noflatmf = False
     state = streamstate()
 
     def stream_out_shallow(repo, proto, other):
         includepattern = None
         excludepattern = None
         raw = other.get('includepattern')
         if raw:
             includepattern = raw.split('\0')
         raw = other.get('excludepattern')
         if raw:
             excludepattern = raw.split('\0')
 
         oldshallow = state.shallowremote
         oldmatch = state.match
         oldnoflatmf = state.noflatmf
         try:
             state.shallowremote = True
             state.match = match.always()
             state.noflatmf = other.get('noflatmanifest') == 'True'
             if includepattern or excludepattern:
                 state.match = match.match(repo.root, '', None,
                     includepattern, excludepattern)
             streamres = wireprotov1server.stream(repo, proto)
 
             # Force the first value to execute, so the file list is computed
             # within the try/finally scope
             first = next(streamres.gen)
             second = next(streamres.gen)
             def gen():
                 yield first
                 yield second
                 for value in streamres.gen:
                     yield value
             return wireprototypes.streamres(gen())
         finally:
             state.shallowremote = oldshallow
             state.match = oldmatch
             state.noflatmf = oldnoflatmf
 
     wireprotov1server.commands['stream_out_shallow'] = (stream_out_shallow, '*')
 
     # don't clone filelogs to shallow clients
     def _walkstreamfiles(orig, repo, matcher=None):
         if state.shallowremote:
             # if we are shallow ourselves, stream our local commits
             if shallowutil.isenabled(repo):
                 striplen = len(repo.store.path) + 1
                 readdir = repo.store.rawvfs.readdir
                 visit = [os.path.join(repo.store.path, 'data')]
                 while visit:
                     p = visit.pop()
                     for f, kind, st in readdir(p, stat=True):
                         fp = p + '/' + f
                         if kind == stat.S_IFREG:
                             if not fp.endswith('.i') and not fp.endswith('.d'):
                                 n = util.pconvert(fp[striplen:])
                                 yield (store.decodedir(n), n, st.st_size)
                         if kind == stat.S_IFDIR:
                             visit.append(fp)
 
             if 'treemanifest' in repo.requirements:
                 for (u, e, s) in repo.store.datafiles():
                     if (u.startswith('meta/') and
                         (u.endswith('.i') or u.endswith('.d'))):
                         yield (u, e, s)
 
             # Return .d and .i files that do not match the shallow pattern
             match = state.match
             if match and not match.always():
                 for (u, e, s) in repo.store.datafiles():
                     f = u[5:-2]  # trim data/...  and .i/.d
                     if not state.match(f):
                         yield (u, e, s)
 
             for x in repo.store.topfiles():
                 if state.noflatmf and x[0][:11] == '00manifest.':
                     continue
                 yield x
 
         elif shallowutil.isenabled(repo):
             # don't allow cloning from a shallow repo to a full repo
             # since it would require fetching every version of every
             # file in order to create the revlogs.
             raise error.Abort(_("Cannot clone from a shallow repo "
                                 "to a full repo."))
         else:
             for x in orig(repo, matcher):
                 yield x
 
     extensions.wrapfunction(streamclone, '_walkstreamfiles', _walkstreamfiles)
 
     # expose remotefilelog capabilities
     def _capabilities(orig, repo, proto):
         caps = orig(repo, proto)
         if (shallowutil.isenabled(repo) or ui.configbool('remotefilelog',
                                                          'server')):
             if isinstance(proto, _sshv1server):
                 # legacy getfiles method which only works over ssh
                 caps.append(constants.NETWORK_CAP_LEGACY_SSH_GETFILES)
             caps.append('x_rfl_getflogheads')
             caps.append('x_rfl_getfile')
         return caps
     extensions.wrapfunction(wireprotov1server, '_capabilities', _capabilities)
 
     def _adjustlinkrev(orig, self, *args, **kwargs):
         # When generating file blobs, taking the real path is too slow on large
         # repos, so force it to just return the linkrev directly.
         repo = self._repo
         if util.safehasattr(repo, 'forcelinkrev') and repo.forcelinkrev:
             return self._filelog.linkrev(self._filelog.rev(self._filenode))
         return orig(self, *args, **kwargs)
 
     extensions.wrapfunction(
         context.basefilectx, '_adjustlinkrev', _adjustlinkrev)
 
     def _iscmd(orig, cmd):
         if cmd == 'x_rfl_getfiles':
             return False
         return orig(cmd)
 
     extensions.wrapfunction(wireprotoserver, 'iscmd', _iscmd)
 
 def _loadfileblob(repo, cachepath, path, node):
     filecachepath = os.path.join(cachepath, path, hex(node))
     if not os.path.exists(filecachepath) or os.path.getsize(filecachepath) == 0:
         filectx = repo.filectx(path, fileid=node)
         if filectx.node() == nullid:
             repo.changelog = changelog.changelog(repo.svfs)
             filectx = repo.filectx(path, fileid=node)
 
         text = createfileblob(filectx)
         # TODO configurable compression engines
         text = zlib.compress(text)
 
         # everything should be user & group read/writable
         oldumask = os.umask(0o002)
         try:
             dirname = os.path.dirname(filecachepath)
             if not os.path.exists(dirname):
                 try:
                     os.makedirs(dirname)
                 except OSError as ex:
                     if ex.errno != errno.EEXIST:
                         raise
 
             f = None
             try:
                 f = util.atomictempfile(filecachepath, "wb")
                 f.write(text)
             except (IOError, OSError):
                 # Don't abort if the user only has permission to read,
                 # and not write.
                 pass
             finally:
                 if f:
                     f.close()
         finally:
             os.umask(oldumask)
     else:
         with open(filecachepath, "rb") as f:
             text = f.read()
     return text
 
 def getflogheads(repo, proto, path):
     """A server api for requesting a filelog's heads
     """
     flog = repo.file(path)
     heads = flog.heads()
     return '\n'.join((hex(head) for head in heads if head != nullid))
 
 def getfile(repo, proto, file, node):
     """A server api for requesting a particular version of a file. Can be used
     in batches to request many files at once. The return protocol is:
     <errorcode>\0<data/errormsg> where <errorcode> is 0 for success or
     non-zero for an error.
 
     data is a compressed blob with revlog flag and ancestors information. See
     createfileblob for its content.
     """
     if shallowutil.isenabled(repo):
         return '1\0' + _('cannot fetch remote files from shallow repo')
     cachepath = repo.ui.config("remotefilelog", "servercachepath")
     if not cachepath:
         cachepath = os.path.join(repo.path, "remotefilelogcache")
     node = bin(node.strip())
     if node == nullid:
         return '0\0'
     return '0\0' + _loadfileblob(repo, cachepath, file, node)
 
 def getfiles(repo, proto):
     """A server api for requesting particular versions of particular files.
     """
     if shallowutil.isenabled(repo):
         raise error.Abort(_('cannot fetch remote files from shallow repo'))
     if not isinstance(proto, _sshv1server):
         raise error.Abort(_('cannot fetch remote files over non-ssh protocol'))
 
     def streamer():
         fin = proto._fin
 
         cachepath = repo.ui.config("remotefilelog", "servercachepath")
         if not cachepath:
             cachepath = os.path.join(repo.path, "remotefilelogcache")
 
         while True:
             request = fin.readline()[:-1]
             if not request:
                 break
 
             node = bin(request[:40])
             if node == nullid:
                 yield '0\n'
                 continue
 
             path = request[40:]
 
             text = _loadfileblob(repo, cachepath, path, node)
 
             yield '%d\n%s' % (len(text), text)
 
             # it would be better to only flush after processing a whole batch
             # but currently we don't know if there are more requests coming
             proto._fout.flush()
     return wireprototypes.streamres(streamer())
 
 def createfileblob(filectx):
     """
     format:
         v0:
             str(len(rawtext)) + '\0' + rawtext + ancestortext
         v1:
             'v1' + '\n' + metalist + '\0' + rawtext + ancestortext
             metalist := metalist + '\n' + meta | meta
             meta := sizemeta | flagmeta
             sizemeta := METAKEYSIZE + str(len(rawtext))
             flagmeta := METAKEYFLAG + str(flag)
 
             note: sizemeta must exist. METAKEYFLAG and METAKEYSIZE must have a
             length of 1.
     """
     flog = filectx.filelog()
     frev = filectx.filerev()
     revlogflags = flog._revlog.flags(frev)
     if revlogflags == 0:
         # normal files
         text = filectx.data()
     else:
         # lfs, read raw revision data
-        text = flog.revision(frev, raw=True)
+        text = flog.rawdata(frev)
 
     repo = filectx._repo
 
     ancestors = [filectx]
 
     try:
         repo.forcelinkrev = True
         ancestors.extend([f for f in filectx.ancestors()])
 
         ancestortext = ""
         for ancestorctx in ancestors:
             parents = ancestorctx.parents()
             p1 = nullid
             p2 = nullid
             if len(parents) > 0:
                 p1 = parents[0].filenode()
             if len(parents) > 1:
                 p2 = parents[1].filenode()
 
             copyname = ""
             rename = ancestorctx.renamed()
             if rename:
                 copyname = rename[0]
             linknode = ancestorctx.node()
             ancestortext += "%s%s%s%s%s\0" % (
                 ancestorctx.filenode(), p1, p2, linknode,
                 copyname)
     finally:
         repo.forcelinkrev = False
 
     header = shallowutil.buildfileblobheader(len(text), revlogflags)
 
     return "%s\0%s%s" % (header, text, ancestortext)
 
 def gcserver(ui, repo):
     if not repo.ui.configbool("remotefilelog", "server"):
         return
 
     neededfiles = set()
     heads = repo.revs("heads(tip~25000:) - null")
 
     cachepath = repo.vfs.join("remotefilelogcache")
     for head in heads:
         mf = repo[head].manifest()
         for filename, filenode in mf.iteritems():
             filecachepath = os.path.join(cachepath, filename, hex(filenode))
             neededfiles.add(filecachepath)
 
     # delete unneeded older files
     days = repo.ui.configint("remotefilelog", "serverexpiration")
     expiration = time.time() - (days * 24 * 60 * 60)
 
     progress = ui.makeprogress(_("removing old server cache"), unit="files")
     progress.update(0)
     for root, dirs, files in os.walk(cachepath):
         for file in files:
             filepath = os.path.join(root, file)
             progress.increment()
             if filepath in neededfiles:
                 continue
 
             stat = os.stat(filepath)
             if stat.st_mtime < expiration:
                 os.remove(filepath)
 
     progress.complete()