remotefilelogserver.py
437 lines
| 14.1 KiB
| text/x-python
|
PythonLexer
|
r40530 | # 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. | ||||
import os | ||||
import stat | ||||
import time | ||||
|
r40542 | import zlib | ||
|
r40530 | |||
from mercurial.i18n import _ | ||||
|
r47771 | from mercurial.node import bin, hex | ||
|
r43355 | from mercurial.pycompat import open | ||
|
r40530 | from mercurial import ( | ||
changegroup, | ||||
changelog, | ||||
context, | ||||
error, | ||||
extensions, | ||||
match, | ||||
|
r46129 | scmutil, | ||
|
r40530 | store, | ||
streamclone, | ||||
util, | ||||
wireprotoserver, | ||||
wireprototypes, | ||||
wireprotov1server, | ||||
) | ||||
|
r43346 | from . import ( | ||
|
r40543 | constants, | ||
|
r40530 | shallowutil, | ||
) | ||||
_sshv1server = wireprotoserver.sshv1protocolhandler | ||||
|
r43346 | |||
|
r40530 | def setupserver(ui, repo): | ||
|
r46554 | """Sets up a normal Mercurial repo so it can serve files to shallow repos.""" | ||
|
r40530 | onetimesetup(ui) | ||
# don't send files to shallow clients during pulls | ||||
|
r43346 | def generatefiles( | ||
orig, self, changedfiles, linknodes, commonrevs, source, *args, **kwargs | ||||
): | ||||
|
r40530 | caps = self._bundlecaps or [] | ||
|
r40544 | if constants.BUNDLE2_CAPABLITY in caps: | ||
|
r40530 | # only send files that don't match the specified patterns | ||
includepattern = None | ||||
excludepattern = None | ||||
|
r43346 | for cap in self._bundlecaps or []: | ||
|
r43347 | if cap.startswith(b"includepattern="): | ||
includepattern = cap[len(b"includepattern=") :].split(b'\0') | ||||
elif cap.startswith(b"excludepattern="): | ||||
excludepattern = cap[len(b"excludepattern=") :].split(b'\0') | ||||
|
r40530 | |||
|
r41825 | m = match.always() | ||
|
r40530 | if includepattern or excludepattern: | ||
|
r43346 | m = match.match( | ||
|
r43347 | repo.root, b'', None, includepattern, excludepattern | ||
|
r43346 | ) | ||
|
r40530 | |||
changedfiles = list([f for f in changedfiles if not m(f)]) | ||||
|
r43346 | return orig( | ||
self, changedfiles, linknodes, commonrevs, source, *args, **kwargs | ||||
) | ||||
|
r40530 | |||
extensions.wrapfunction( | ||||
|
r43347 | changegroup.cgpacker, b'generatefiles', generatefiles | ||
|
r43346 | ) | ||
|
r40530 | |||
onetime = False | ||||
|
r43346 | |||
|
r40530 | def onetimesetup(ui): | ||
|
r46554 | """Configures the wireprotocol for both clients and servers.""" | ||
|
r40530 | global onetime | ||
if onetime: | ||||
return | ||||
onetime = True | ||||
# support file content requests | ||||
wireprotov1server.wireprotocommand( | ||||
|
r43347 | b'x_rfl_getflogheads', b'path', permission=b'pull' | ||
|
r43346 | )(getflogheads) | ||
|
r40530 | wireprotov1server.wireprotocommand( | ||
|
r43347 | b'x_rfl_getfiles', b'', permission=b'pull' | ||
)(getfiles) | ||||
wireprotov1server.wireprotocommand( | ||||
b'x_rfl_getfile', b'file node', permission=b'pull' | ||||
|
r43346 | )(getfile) | ||
|
r40530 | |||
|
r49801 | class streamstate: | ||
|
r40530 | match = None | ||
shallowremote = False | ||||
noflatmf = False | ||||
|
r43346 | |||
|
r40530 | state = streamstate() | ||
def stream_out_shallow(repo, proto, other): | ||||
includepattern = None | ||||
excludepattern = None | ||||
|
r43347 | raw = other.get(b'includepattern') | ||
|
r40530 | if raw: | ||
|
r43347 | includepattern = raw.split(b'\0') | ||
raw = other.get(b'excludepattern') | ||||
|
r40530 | if raw: | ||
|
r43347 | excludepattern = raw.split(b'\0') | ||
|
r40530 | |||
oldshallow = state.shallowremote | ||||
oldmatch = state.match | ||||
oldnoflatmf = state.noflatmf | ||||
try: | ||||
state.shallowremote = True | ||||
|
r41825 | state.match = match.always() | ||
|
r43347 | state.noflatmf = other.get(b'noflatmanifest') == b'True' | ||
|
r40530 | if includepattern or excludepattern: | ||
|
r43346 | state.match = match.match( | ||
|
r43347 | repo.root, b'', None, includepattern, excludepattern | ||
|
r43346 | ) | ||
|
r40530 | 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) | ||||
|
r43346 | |||
|
r40530 | def gen(): | ||
yield first | ||||
yield second | ||||
for value in streamres.gen: | ||||
yield value | ||||
|
r43346 | |||
|
r40530 | return wireprototypes.streamres(gen()) | ||
finally: | ||||
state.shallowremote = oldshallow | ||||
state.match = oldmatch | ||||
state.noflatmf = oldnoflatmf | ||||
|
r43347 | wireprotov1server.commands[b'stream_out_shallow'] = ( | ||
stream_out_shallow, | ||||
b'*', | ||||
) | ||||
|
r40530 | |||
# don't clone filelogs to shallow clients | ||||
|
r40550 | def _walkstreamfiles(orig, repo, matcher=None): | ||
|
r40530 | if state.shallowremote: | ||
# if we are shallow ourselves, stream our local commits | ||||
|
r40549 | if shallowutil.isenabled(repo): | ||
|
r40530 | striplen = len(repo.store.path) + 1 | ||
readdir = repo.store.rawvfs.readdir | ||||
|
r43347 | visit = [os.path.join(repo.store.path, b'data')] | ||
|
r40530 | while visit: | ||
p = visit.pop() | ||||
for f, kind, st in readdir(p, stat=True): | ||||
|
r43347 | fp = p + b'/' + f | ||
|
r40530 | if kind == stat.S_IFREG: | ||
|
r43347 | if not fp.endswith(b'.i') and not fp.endswith( | ||
b'.d' | ||||
): | ||||
|
r40530 | n = util.pconvert(fp[striplen:]) | ||
r47657 | d = store.decodedir(n) | |||
t = store.FILETYPE_OTHER | ||||
|
r48691 | yield (t, d, st.st_size) | ||
|
r40530 | if kind == stat.S_IFDIR: | ||
visit.append(fp) | ||||
|
r46129 | if scmutil.istreemanifest(repo): | ||
|
r48691 | for (t, u, s) in repo.store.datafiles(): | ||
|
r43347 | if u.startswith(b'meta/') and ( | ||
u.endswith(b'.i') or u.endswith(b'.d') | ||||
|
r43346 | ): | ||
|
r48691 | yield (t, u, s) | ||
|
r40530 | |||
# Return .d and .i files that do not match the shallow pattern | ||||
match = state.match | ||||
if match and not match.always(): | ||||
|
r48691 | for (t, u, s) in repo.store.datafiles(): | ||
|
r40530 | f = u[5:-2] # trim data/... and .i/.d | ||
if not state.match(f): | ||||
|
r48691 | yield (t, u, s) | ||
|
r40530 | |||
for x in repo.store.topfiles(): | ||||
|
r48601 | if state.noflatmf and x[1][:11] == b'00manifest.': | ||
|
r40530 | continue | ||
yield x | ||||
|
r40549 | elif shallowutil.isenabled(repo): | ||
|
r40530 | # 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. | ||||
|
r43346 | raise error.Abort( | ||
|
r43387 | _(b"Cannot clone from a shallow repo to a full repo.") | ||
|
r43346 | ) | ||
|
r40530 | else: | ||
|
r40550 | for x in orig(repo, matcher): | ||
|
r40530 | yield x | ||
|
r43347 | extensions.wrapfunction(streamclone, b'_walkstreamfiles', _walkstreamfiles) | ||
|
r40530 | |||
# expose remotefilelog capabilities | ||||
def _capabilities(orig, repo, proto): | ||||
caps = orig(repo, proto) | ||||
|
r43346 | if shallowutil.isenabled(repo) or ui.configbool( | ||
|
r43347 | b'remotefilelog', b'server' | ||
|
r43346 | ): | ||
|
r40530 | if isinstance(proto, _sshv1server): | ||
# legacy getfiles method which only works over ssh | ||||
|
r40543 | caps.append(constants.NETWORK_CAP_LEGACY_SSH_GETFILES) | ||
|
r43347 | caps.append(b'x_rfl_getflogheads') | ||
caps.append(b'x_rfl_getfile') | ||||
|
r40530 | return caps | ||
|
r43346 | |||
|
r43347 | extensions.wrapfunction(wireprotov1server, b'_capabilities', _capabilities) | ||
|
r40530 | |||
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 | ||||
|
r43347 | if util.safehasattr(repo, b'forcelinkrev') and repo.forcelinkrev: | ||
|
r40530 | return self._filelog.linkrev(self._filelog.rev(self._filenode)) | ||
return orig(self, *args, **kwargs) | ||||
extensions.wrapfunction( | ||||
|
r43347 | context.basefilectx, b'_adjustlinkrev', _adjustlinkrev | ||
|
r43346 | ) | ||
|
r40530 | |||
def _iscmd(orig, cmd): | ||||
|
r43347 | if cmd == b'x_rfl_getfiles': | ||
|
r40530 | return False | ||
return orig(cmd) | ||||
|
r43347 | extensions.wrapfunction(wireprotoserver, b'iscmd', _iscmd) | ||
|
r40530 | |||
|
r43346 | |||
|
r40530 | 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) | ||||
|
r47771 | if filectx.node() == repo.nullid: | ||
|
r40530 | repo.changelog = changelog.changelog(repo.svfs) | ||
filectx = repo.filectx(path, fileid=node) | ||||
text = createfileblob(filectx) | ||||
|
r40542 | # TODO configurable compression engines | ||
text = zlib.compress(text) | ||||
|
r40530 | |||
# 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) | ||||
|
r50200 | except FileExistsError: | ||
pass | ||||
|
r40530 | |||
f = None | ||||
try: | ||||
|
r43347 | f = util.atomictempfile(filecachepath, b"wb") | ||
|
r40530 | 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: | ||||
|
r43347 | with open(filecachepath, b"rb") as f: | ||
|
r40530 | text = f.read() | ||
return text | ||||
|
r43346 | |||
|
r40530 | def getflogheads(repo, proto, path): | ||
|
r46554 | """A server api for requesting a filelog's heads""" | ||
|
r40530 | flog = repo.file(path) | ||
heads = flog.heads() | ||||
|
r47771 | return b'\n'.join((hex(head) for head in heads if head != repo.nullid)) | ||
|
r40530 | |||
|
r43346 | |||
|
r40530 | 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. | ||||
""" | ||||
|
r40549 | if shallowutil.isenabled(repo): | ||
|
r43347 | return b'1\0' + _(b'cannot fetch remote files from shallow repo') | ||
cachepath = repo.ui.config(b"remotefilelog", b"servercachepath") | ||||
|
r40530 | if not cachepath: | ||
|
r43347 | cachepath = os.path.join(repo.path, b"remotefilelogcache") | ||
|
r40530 | node = bin(node.strip()) | ||
|
r47771 | if node == repo.nullid: | ||
|
r43347 | return b'0\0' | ||
return b'0\0' + _loadfileblob(repo, cachepath, file, node) | ||||
|
r40530 | |||
|
r43346 | |||
|
r40530 | def getfiles(repo, proto): | ||
|
r46554 | """A server api for requesting particular versions of particular files.""" | ||
|
r40549 | if shallowutil.isenabled(repo): | ||
|
r43347 | raise error.Abort(_(b'cannot fetch remote files from shallow repo')) | ||
|
r40530 | if not isinstance(proto, _sshv1server): | ||
|
r43347 | raise error.Abort(_(b'cannot fetch remote files over non-ssh protocol')) | ||
|
r40530 | |||
def streamer(): | ||||
fin = proto._fin | ||||
|
r43347 | cachepath = repo.ui.config(b"remotefilelog", b"servercachepath") | ||
|
r40530 | if not cachepath: | ||
|
r43347 | cachepath = os.path.join(repo.path, b"remotefilelogcache") | ||
|
r40530 | |||
while True: | ||||
request = fin.readline()[:-1] | ||||
if not request: | ||||
break | ||||
node = bin(request[:40]) | ||||
|
r47771 | if node == repo.nullid: | ||
|
r43347 | yield b'0\n' | ||
|
r40530 | continue | ||
path = request[40:] | ||||
text = _loadfileblob(repo, cachepath, path, node) | ||||
|
r43347 | yield b'%d\n%s' % (len(text), text) | ||
|
r40530 | |||
# 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() | ||||
|
r43346 | |||
|
r40530 | return wireprototypes.streamres(streamer()) | ||
|
r43346 | |||
|
r40530 | 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 | ||||
r43039 | text = flog.rawdata(frev) | |||
|
r40530 | |||
repo = filectx._repo | ||||
ancestors = [filectx] | ||||
try: | ||||
repo.forcelinkrev = True | ||||
ancestors.extend([f for f in filectx.ancestors()]) | ||||
|
r43347 | ancestortext = b"" | ||
|
r40530 | for ancestorctx in ancestors: | ||
parents = ancestorctx.parents() | ||||
|
r47771 | p1 = repo.nullid | ||
p2 = repo.nullid | ||||
|
r40530 | if len(parents) > 0: | ||
p1 = parents[0].filenode() | ||||
if len(parents) > 1: | ||||
p2 = parents[1].filenode() | ||||
|
r43347 | copyname = b"" | ||
|
r40530 | rename = ancestorctx.renamed() | ||
if rename: | ||||
copyname = rename[0] | ||||
linknode = ancestorctx.node() | ||||
|
r43347 | ancestortext += b"%s%s%s%s%s\0" % ( | ||
|
r43346 | ancestorctx.filenode(), | ||
p1, | ||||
p2, | ||||
linknode, | ||||
copyname, | ||||
) | ||||
|
r40530 | finally: | ||
repo.forcelinkrev = False | ||||
header = shallowutil.buildfileblobheader(len(text), revlogflags) | ||||
|
r43347 | return b"%s\0%s%s" % (header, text, ancestortext) | ||
|
r40530 | |||
|
r43346 | |||
|
r40530 | def gcserver(ui, repo): | ||
|
r43347 | if not repo.ui.configbool(b"remotefilelog", b"server"): | ||
|
r40530 | return | ||
neededfiles = set() | ||||
|
r43347 | heads = repo.revs(b"heads(tip~25000:) - null") | ||
|
r40530 | |||
|
r43347 | cachepath = repo.vfs.join(b"remotefilelogcache") | ||
|
r40530 | for head in heads: | ||
mf = repo[head].manifest() | ||||
|
r49768 | for filename, filenode in mf.items(): | ||
|
r40530 | filecachepath = os.path.join(cachepath, filename, hex(filenode)) | ||
neededfiles.add(filecachepath) | ||||
# delete unneeded older files | ||||
|
r43347 | days = repo.ui.configint(b"remotefilelog", b"serverexpiration") | ||
|
r40530 | expiration = time.time() - (days * 24 * 60 * 60) | ||
|
r43347 | progress = ui.makeprogress(_(b"removing old server cache"), unit=b"files") | ||
|
r40877 | progress.update(0) | ||
|
r40530 | for root, dirs, files in os.walk(cachepath): | ||
for file in files: | ||||
filepath = os.path.join(root, file) | ||||
|
r40877 | progress.increment() | ||
|
r40530 | if filepath in neededfiles: | ||
continue | ||||
stat = os.stat(filepath) | ||||
if stat.st_mtime < expiration: | ||||
os.remove(filepath) | ||||
|
r40877 | progress.complete() | ||