streamclone.py
941 lines
| 30.6 KiB
| text/x-python
|
PythonLexer
/ mercurial / streamclone.py
|
r26441 | # streamclone.py - producing and consuming streaming repository data | ||
# | ||||
# Copyright 2015 Gregory Szorc <gregory.szorc@gmail.com> | ||||
# | ||||
# 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 | ||||
|
r35783 | import contextlib | ||
r48240 | import errno | |||
|
r35783 | import os | ||
|
r26755 | import struct | ||
|
r26443 | |||
|
r26442 | from .i18n import _ | ||
|
r43355 | from .pycompat import open | ||
|
r43346 | from .interfaces import repository | ||
|
r26441 | from . import ( | ||
r48240 | bookmarks, | |||
|
r35785 | cacheutil, | ||
|
r26442 | error, | ||
|
r40375 | narrowspec, | ||
|
r32744 | phases, | ||
|
r38182 | pycompat, | ||
|
r47371 | requirements as requirementsmod, | ||
|
r45666 | scmutil, | ||
|
r26443 | store, | ||
|
r26442 | util, | ||
|
r26441 | ) | ||
r48240 | from .utils import ( | |||
stringutil, | ||||
) | ||||
|
r26441 | |||
|
r43346 | |||
r49442 | def new_stream_clone_requirements( | |||
supported_formats, default_requirements, streamed_requirements | ||||
): | ||||
"""determine the final set of requirement for a new stream clone | ||||
this method combine the "default" requirements that a new repository would | ||||
use with the constaint we get from the stream clone content. We keep local | ||||
configuration choice when possible. | ||||
""" | ||||
requirements = set(default_requirements) | ||||
requirements -= supported_formats | ||||
requirements.update(streamed_requirements) | ||||
return requirements | ||||
r49443 | def streamed_requirements(repo): | |||
"""the set of requirement the new clone will have to support | ||||
This is used for advertising the stream options and to generate the actual | ||||
stream content.""" | ||||
requiredformats = repo.requirements & repo.supportedformats | ||||
return requiredformats | ||||
|
r35775 | def canperformstreamclone(pullop, bundle2=False): | ||
|
r26446 | """Whether it is possible to perform a streaming clone as part of pull. | ||
|
r26445 | |||
|
r35775 | ``bundle2`` will cause the function to consider stream clone through | ||
bundle2 and only through bundle2. | ||||
|
r26467 | |||
|
r26446 | Returns a tuple of (supported, requirements). ``supported`` is True if | ||
streaming clone is supported and False otherwise. ``requirements`` is | ||||
a set of repo requirements from the remote, or ``None`` if stream clone | ||||
isn't supported. | ||||
""" | ||||
|
r26466 | repo = pullop.repo | ||
remote = pullop.remote | ||||
|
r26467 | bundle2supported = False | ||
if pullop.canusebundle2: | ||||
|
r43347 | if b'v2' in pullop.remotebundle2caps.get(b'stream', []): | ||
|
r26467 | bundle2supported = True | ||
# else | ||||
|
r43346 | # Server doesn't support bundle2 stream clone or doesn't support | ||
# the versions we support. Fall back and possibly allow legacy. | ||||
|
r26467 | |||
# Ensures legacy code path uses available bundle2. | ||||
|
r35775 | if bundle2supported and not bundle2: | ||
|
r26467 | return False, None | ||
# Ensures bundle2 doesn't try to do a stream clone if it isn't supported. | ||||
|
r35775 | elif bundle2 and not bundle2supported: | ||
return False, None | ||||
|
r26467 | |||
|
r26447 | # Streaming clone only works on empty repositories. | ||
if len(repo): | ||||
return False, None | ||||
|
r26446 | # Streaming clone only works if all data is being requested. | ||
|
r26466 | if pullop.heads: | ||
|
r26446 | return False, None | ||
|
r26445 | |||
|
r26466 | streamrequested = pullop.streamclonerequested | ||
|
r26446 | # If we don't have a preference, let the server decide for us. This | ||
# likely only comes into play in LANs. | ||||
if streamrequested is None: | ||||
# The server can advertise whether to prefer streaming clone. | ||||
|
r43347 | streamrequested = remote.capable(b'stream-preferred') | ||
|
r26446 | |||
if not streamrequested: | ||||
return False, None | ||||
|
r26445 | |||
|
r26446 | # In order for stream clone to work, the client has to support all the | ||
# requirements advertised by the server. | ||||
# | ||||
# The server advertises its requirements via the "stream" and "streamreqs" | ||||
# capability. "stream" (a value-less capability) is advertised if and only | ||||
# if the only requirement is "revlogv1." Else, the "streamreqs" capability | ||||
# is advertised and contains a comma-delimited list of requirements. | ||||
requirements = set() | ||||
|
r43347 | if remote.capable(b'stream'): | ||
|
r47371 | requirements.add(requirementsmod.REVLOGV1_REQUIREMENT) | ||
|
r26446 | else: | ||
|
r43347 | streamreqs = remote.capable(b'streamreqs') | ||
|
r26446 | # This is weird and shouldn't happen with modern servers. | ||
if not streamreqs: | ||||
|
r43346 | pullop.repo.ui.warn( | ||
_( | ||||
|
r43347 | b'warning: stream clone requested but server has them ' | ||
b'disabled\n' | ||||
|
r43346 | ) | ||
) | ||||
|
r26446 | return False, None | ||
|
r43347 | streamreqs = set(streamreqs.split(b',')) | ||
|
r26446 | # Server requires something we don't support. Bail. | ||
r49446 | missingreqs = streamreqs - repo.supported | |||
|
r32259 | if missingreqs: | ||
pullop.repo.ui.warn( | ||||
|
r43346 | _( | ||
|
r43347 | b'warning: stream clone requested but client is missing ' | ||
b'requirements: %s\n' | ||||
|
r43346 | ) | ||
|
r43347 | % b', '.join(sorted(missingreqs)) | ||
|
r43346 | ) | ||
pullop.repo.ui.warn( | ||||
_( | ||||
|
r43347 | b'(see https://www.mercurial-scm.org/wiki/MissingRequirement ' | ||
b'for more information)\n' | ||||
|
r43346 | ) | ||
) | ||||
|
r26446 | return False, None | ||
requirements = streamreqs | ||||
return True, requirements | ||||
|
r43346 | |||
|
r26462 | def maybeperformlegacystreamclone(pullop): | ||
"""Possibly perform a legacy stream clone operation. | ||||
Legacy stream clones are performed as part of pull but before all other | ||||
operations. | ||||
A legacy stream clone will not be performed if a bundle2 stream clone is | ||||
supported. | ||||
""" | ||||
|
r39736 | from . import localrepo | ||
|
r26466 | supported, requirements = canperformstreamclone(pullop) | ||
|
r26458 | |||
|
r26446 | if not supported: | ||
return | ||||
|
r26466 | repo = pullop.repo | ||
remote = pullop.remote | ||||
|
r26459 | # Save remote branchmap. We will use it later to speed up branchcache | ||
# creation. | ||||
rbranchmap = None | ||||
|
r43347 | if remote.capable(b'branchmap'): | ||
|
r37656 | with remote.commandexecutor() as e: | ||
|
r43347 | rbranchmap = e.callcommand(b'branchmap', {}).result() | ||
|
r26459 | |||
|
r43347 | repo.ui.status(_(b'streaming all changes\n')) | ||
|
r26470 | |||
|
r37656 | with remote.commandexecutor() as e: | ||
|
r43347 | fp = e.callcommand(b'stream_out', {}).result() | ||
|
r37656 | |||
# TODO strictly speaking, this code should all be inside the context | ||||
# manager because the context manager is supposed to ensure all wire state | ||||
# is flushed when exiting. But the legacy peers don't do this, so it | ||||
# doesn't matter. | ||||
|
r26459 | l = fp.readline() | ||
try: | ||||
resp = int(l) | ||||
except ValueError: | ||||
raise error.ResponseError( | ||||
|
r43347 | _(b'unexpected response from remote server:'), l | ||
|
r43346 | ) | ||
|
r26459 | if resp == 1: | ||
|
r43347 | raise error.Abort(_(b'operation forbidden by server')) | ||
|
r26459 | elif resp == 2: | ||
|
r43347 | raise error.Abort(_(b'locking the remote repository failed')) | ||
|
r26459 | elif resp != 0: | ||
|
r43347 | raise error.Abort(_(b'the server sent an unknown error code')) | ||
|
r26459 | |||
|
r26468 | l = fp.readline() | ||
try: | ||||
|
r43347 | filecount, bytecount = map(int, l.split(b' ', 1)) | ||
|
r26468 | except (ValueError, TypeError): | ||
raise error.ResponseError( | ||||
|
r43347 | _(b'unexpected response from remote server:'), l | ||
|
r43346 | ) | ||
|
r26468 | |||
|
r27850 | with repo.lock(): | ||
|
r26468 | consumev1(repo, fp, filecount, bytecount) | ||
r49442 | repo.requirements = new_stream_clone_requirements( | |||
repo.supportedformats, | ||||
repo.requirements, | ||||
requirements, | ||||
|
r43346 | ) | ||
|
r39736 | repo.svfs.options = localrepo.resolvestorevfsoptions( | ||
|
r43346 | repo.ui, repo.requirements, repo.features | ||
) | ||||
|
r45666 | scmutil.writereporequirements(repo) | ||
|
r26461 | |||
if rbranchmap: | ||||
|
r41764 | repo._branchcaches.replace(repo, rbranchmap) | ||
|
r26461 | |||
repo.invalidate() | ||||
|
r26445 | |||
|
r43346 | |||
|
r32744 | def allowservergeneration(repo): | ||
|
r26444 | """Whether streaming clones are allowed from the server.""" | ||
|
r40064 | if repository.REPO_FEATURE_STREAM_CLONE not in repo.features: | ||
return False | ||||
|
r43347 | if not repo.ui.configbool(b'server', b'uncompressed', untrusted=True): | ||
|
r32744 | return False | ||
# The way stream clone works makes it impossible to hide secret changesets. | ||||
# So don't allow this by default. | ||||
secret = phases.hassecret(repo) | ||||
if secret: | ||||
|
r43347 | return repo.ui.configbool(b'server', b'uncompressedallowsecret') | ||
|
r32744 | |||
return True | ||||
|
r26444 | |||
|
r43346 | |||
|
r26443 | # This is it's own function so extensions can override it. | ||
|
r40375 | def _walkstreamfiles(repo, matcher=None): | ||
return repo.store.walk(matcher) | ||||
|
r26443 | |||
|
r43346 | |||
|
r26443 | def generatev1(repo): | ||
"""Emit content for version 1 of a streaming clone. | ||||
|
r26469 | This returns a 3-tuple of (file count, byte size, data iterator). | ||
|
r26443 | |||
|
r26469 | The data iterator consists of N entries for each file being transferred. | ||
Each file entry starts as a line with the file name and integer size | ||||
delimited by a null byte. | ||||
|
r26443 | |||
The raw file data follows. Following the raw file data is the next file | ||||
entry, or EOF. | ||||
When used on the wire protocol, an additional line indicating protocol | ||||
success will be prepended to the stream. This function is not responsible | ||||
for adding it. | ||||
This function will obtain a repository lock to ensure a consistent view of | ||||
the store is captured. It therefore may raise LockError. | ||||
""" | ||||
entries = [] | ||||
total_bytes = 0 | ||||
# Get consistent snapshot of repo, lock during scan. | ||||
|
r27845 | with repo.lock(): | ||
|
r43347 | repo.ui.debug(b'scanning\n') | ||
|
r48691 | for file_type, name, size in _walkstreamfiles(repo): | ||
|
r26443 | if size: | ||
entries.append((name, size)) | ||||
total_bytes += size | ||||
r47748 | _test_sync_point_walk_1(repo) | |||
_test_sync_point_walk_2(repo) | ||||
|
r26443 | |||
|
r43346 | repo.ui.debug( | ||
|
r43347 | b'%d files, %d bytes to transfer\n' % (len(entries), total_bytes) | ||
|
r43346 | ) | ||
|
r26443 | |||
svfs = repo.svfs | ||||
debugflag = repo.ui.debugflag | ||||
|
r26469 | def emitrevlogdata(): | ||
r33256 | for name, size in entries: | |||
if debugflag: | ||||
|
r43347 | repo.ui.debug(b'sending %s (%d bytes)\n' % (name, size)) | ||
r33256 | # partially encode name over the wire for backwards compat | |||
|
r43347 | yield b'%s\0%d\n' % (store.encodedir(name), size) | ||
|
r33411 | # auditing at this stage is both pointless (paths are already | ||
# trusted by the local repo) and expensive | ||||
|
r43347 | with svfs(name, b'rb', auditpath=False) as fp: | ||
|
r33410 | if size <= 65536: | ||
r33256 | yield fp.read(size) | |||
|
r33410 | else: | ||
for chunk in util.filechunkiter(fp, limit=size): | ||||
yield chunk | ||||
|
r26469 | |||
return len(entries), total_bytes, emitrevlogdata() | ||||
|
r43346 | |||
|
r26469 | def generatev1wireproto(repo): | ||
"""Emit content for version 1 of streaming clone suitable for the wire. | ||||
|
r35507 | This is the data output from ``generatev1()`` with 2 header lines. The | ||
first line indicates overall success. The 2nd contains the file count and | ||||
byte size of payload. | ||||
The success line contains "0" for success, "1" for stream generation not | ||||
allowed, and "2" for error locking the repository (possibly indicating | ||||
a permissions error for the server process). | ||||
|
r26469 | """ | ||
|
r35507 | if not allowservergeneration(repo): | ||
|
r43347 | yield b'1\n' | ||
|
r35507 | return | ||
try: | ||||
filecount, bytecount, it = generatev1(repo) | ||||
except error.LockError: | ||||
|
r43347 | yield b'2\n' | ||
|
r35507 | return | ||
# Indicates successful response. | ||||
|
r43347 | yield b'0\n' | ||
yield b'%d %d\n' % (filecount, bytecount) | ||||
|
r26469 | for chunk in it: | ||
yield chunk | ||||
|
r26443 | |||
|
r43346 | |||
|
r43347 | def generatebundlev1(repo, compression=b'UN'): | ||
|
r26755 | """Emit content for version 1 of a stream clone bundle. | ||
The first 4 bytes of the output ("HGS1") denote this as stream clone | ||||
bundle version 1. | ||||
The next 2 bytes indicate the compression type. Only "UN" is currently | ||||
supported. | ||||
The next 16 bytes are two 64-bit big endian unsigned integers indicating | ||||
file count and byte count, respectively. | ||||
The next 2 bytes is a 16-bit big endian unsigned short declaring the length | ||||
of the requirements string, including a trailing \0. The following N bytes | ||||
are the requirements string, which is ASCII containing a comma-delimited | ||||
list of repo requirements that are needed to support the data. | ||||
The remaining content is the output of ``generatev1()`` (which may be | ||||
compressed in the future). | ||||
Returns a tuple of (requirements, data generator). | ||||
""" | ||||
|
r43347 | if compression != b'UN': | ||
raise ValueError(b'we do not support the compression argument yet') | ||||
|
r26755 | |||
r49443 | requirements = streamed_requirements(repo) | |||
|
r43347 | requires = b','.join(sorted(requirements)) | ||
|
r26755 | |||
def gen(): | ||||
|
r43347 | yield b'HGS1' | ||
|
r26755 | yield compression | ||
filecount, bytecount, it = generatev1(repo) | ||||
|
r43346 | repo.ui.status( | ||
|
r43347 | _(b'writing %d bytes for %d files\n') % (bytecount, filecount) | ||
|
r43346 | ) | ||
|
r26755 | |||
|
r43347 | yield struct.pack(b'>QQ', filecount, bytecount) | ||
yield struct.pack(b'>H', len(requires) + 1) | ||||
yield requires + b'\0' | ||||
|
r26755 | |||
# This is where we'll add compression in the future. | ||||
|
r43347 | assert compression == b'UN' | ||
|
r26755 | |||
|
r43346 | progress = repo.ui.makeprogress( | ||
|
r43347 | _(b'bundle'), total=bytecount, unit=_(b'bytes') | ||
|
r43346 | ) | ||
|
r38368 | progress.update(0) | ||
|
r26755 | |||
for chunk in it: | ||||
|
r38368 | progress.increment(step=len(chunk)) | ||
|
r26755 | yield chunk | ||
|
r38392 | progress.complete() | ||
|
r26755 | |||
return requirements, gen() | ||||
|
r43346 | |||
|
r26468 | def consumev1(repo, fp, filecount, bytecount): | ||
|
r26443 | """Apply the contents from version 1 of a streaming clone file handle. | ||
|
r30332 | This takes the output from "stream_out" and applies it to the specified | ||
|
r26443 | repository. | ||
|
r30332 | Like "stream_out," the status line added by the wire protocol is not | ||
handled by this function. | ||||
|
r26443 | """ | ||
|
r27859 | with repo.lock(): | ||
|
r43346 | repo.ui.status( | ||
|
r43347 | _(b'%d files to transfer, %s of data\n') | ||
|
r43346 | % (filecount, util.bytecount(bytecount)) | ||
) | ||||
progress = repo.ui.makeprogress( | ||||
|
r43347 | _(b'clone'), total=bytecount, unit=_(b'bytes') | ||
|
r43346 | ) | ||
|
r38368 | progress.update(0) | ||
|
r30975 | start = util.timer() | ||
|
r26443 | |||
|
r29919 | # TODO: get rid of (potential) inconsistency | ||
# | ||||
# If transaction is started and any @filecache property is | ||||
# changed at this point, it causes inconsistency between | ||||
# in-memory cached property and streamclone-ed file on the | ||||
# disk. Nested transaction prevents transaction scope "clone" | ||||
# below from writing in-memory changes out at the end of it, | ||||
# even though in-memory changes are discarded at the end of it | ||||
# regardless of transaction nesting. | ||||
# | ||||
# But transaction nesting can't be simply prohibited, because | ||||
# nesting occurs also in ordinary case (e.g. enabling | ||||
# clonebundles). | ||||
|
r43347 | with repo.transaction(b'clone'): | ||
|
r27897 | with repo.svfs.backgroundclosing(repo.ui, expectedcount=filecount): | ||
|
r38806 | for i in pycompat.xrange(filecount): | ||
|
r27896 | # XXX doesn't support '\n' or '\r' in filenames | ||
l = fp.readline() | ||||
try: | ||||
|
r43347 | name, size = l.split(b'\0', 1) | ||
|
r27896 | size = int(size) | ||
except (ValueError, TypeError): | ||||
raise error.ResponseError( | ||||
|
r43347 | _(b'unexpected response from remote server:'), l | ||
|
r43346 | ) | ||
|
r27896 | if repo.ui.debugflag: | ||
|
r43346 | repo.ui.debug( | ||
|
r43347 | b'adding %s (%s)\n' % (name, util.bytecount(size)) | ||
|
r43346 | ) | ||
|
r27896 | # for backwards compat, name was partially encoded | ||
|
r27897 | path = store.decodedir(name) | ||
|
r43347 | with repo.svfs(path, b'w', backgroundclose=True) as ofp: | ||
|
r27896 | for chunk in util.filechunkiter(fp, limit=size): | ||
|
r38368 | progress.increment(step=len(chunk)) | ||
|
r27896 | ofp.write(chunk) | ||
|
r26443 | |||
|
r29919 | # force @filecache properties to be reloaded from | ||
# streamclone-ed file at next access | ||||
repo.invalidate(clearfilecache=True) | ||||
|
r26443 | |||
|
r30975 | elapsed = util.timer() - start | ||
|
r26443 | if elapsed <= 0: | ||
elapsed = 0.001 | ||||
|
r38392 | progress.complete() | ||
|
r43346 | repo.ui.status( | ||
|
r43347 | _(b'transferred %s in %.1f seconds (%s/sec)\n') | ||
|
r43346 | % ( | ||
util.bytecount(bytecount), | ||||
elapsed, | ||||
util.bytecount(bytecount / elapsed), | ||||
) | ||||
) | ||||
|
r26755 | |||
|
r27882 | def readbundle1header(fp): | ||
|
r26755 | compression = fp.read(2) | ||
|
r43347 | if compression != b'UN': | ||
|
r43346 | raise error.Abort( | ||
|
r43347 | _( | ||
b'only uncompressed stream clone bundles are ' | ||||
b'supported; got %s' | ||||
) | ||||
|
r43346 | % compression | ||
) | ||||
|
r26755 | |||
|
r43347 | filecount, bytecount = struct.unpack(b'>QQ', fp.read(16)) | ||
requireslen = struct.unpack(b'>H', fp.read(2))[0] | ||||
|
r26755 | requires = fp.read(requireslen) | ||
|
r43347 | if not requires.endswith(b'\0'): | ||
|
r43346 | raise error.Abort( | ||
_( | ||||
|
r43347 | b'malformed stream clone bundle: ' | ||
b'requirements not properly encoded' | ||||
|
r43346 | ) | ||
) | ||||
|
r26755 | |||
|
r43347 | requirements = set(requires.rstrip(b'\0').split(b',')) | ||
|
r27882 | |||
return filecount, bytecount, requirements | ||||
|
r43346 | |||
|
r27882 | def applybundlev1(repo, fp): | ||
"""Apply the content from a stream clone bundle version 1. | ||||
We assume the 4 byte header has been read and validated and the file handle | ||||
is at the 2 byte compression identifier. | ||||
""" | ||||
if len(repo): | ||||
|
r43346 | raise error.Abort( | ||
|
r43387 | _(b'cannot apply stream clone bundle on non-empty repo') | ||
|
r43346 | ) | ||
|
r27882 | |||
filecount, bytecount, requirements = readbundle1header(fp) | ||||
r49446 | missingreqs = requirements - repo.supported | |||
|
r26755 | if missingreqs: | ||
|
r43346 | raise error.Abort( | ||
|
r43387 | _(b'unable to apply stream clone: unsupported format: %s') | ||
|
r43347 | % b', '.join(sorted(missingreqs)) | ||
|
r43346 | ) | ||
|
r26755 | |||
consumev1(repo, fp, filecount, bytecount) | ||||
|
r43346 | |||
|
r26755 | class streamcloneapplier(object): | ||
"""Class to manage applying streaming clone bundles. | ||||
We need to wrap ``applybundlev1()`` in a dedicated type to enable bundle | ||||
readers to perform bundle type-specific functionality. | ||||
""" | ||||
|
r43346 | |||
|
r26755 | def __init__(self, fh): | ||
self._fh = fh | ||||
def apply(self, repo): | ||||
return applybundlev1(repo, self._fh) | ||||
|
r35774 | |||
|
r43346 | |||
|
r35783 | # type of file to stream | ||
|
r43346 | _fileappend = 0 # append only file | ||
_filefull = 1 # full snapshot file | ||||
|
r35783 | |||
|
r35785 | # Source of the file | ||
|
r43347 | _srcstore = b's' # store (svfs) | ||
_srccache = b'c' # cache (cache) | ||||
|
r35785 | |||
|
r35783 | # This is it's own function so extensions can override it. | ||
def _walkstreamfullstorefiles(repo): | ||||
"""list snapshot file from the store""" | ||||
fnames = [] | ||||
if not repo.publishing(): | ||||
|
r43347 | fnames.append(b'phaseroots') | ||
|
r35783 | return fnames | ||
|
r43346 | |||
|
r35785 | def _filterfull(entry, copy, vfsmap): | ||
|
r35783 | """actually copy the snapshot files""" | ||
|
r35785 | src, name, ftype, data = entry | ||
|
r35783 | if ftype != _filefull: | ||
return entry | ||||
|
r35785 | return (src, name, ftype, copy(vfsmap[src].join(name))) | ||
|
r35783 | |||
|
r43346 | |||
|
r35783 | @contextlib.contextmanager | ||
def maketempcopies(): | ||||
"""return a function to temporary copy file""" | ||||
files = [] | ||||
try: | ||||
|
r43346 | |||
|
r35783 | def copy(src): | ||
|
r38182 | fd, dst = pycompat.mkstemp() | ||
|
r35783 | os.close(fd) | ||
files.append(dst) | ||||
util.copyfiles(src, dst, hardlink=True) | ||||
return dst | ||||
|
r43346 | |||
|
r35783 | yield copy | ||
finally: | ||||
for tmp in files: | ||||
util.tryunlink(tmp) | ||||
|
r43346 | |||
|
r35785 | def _makemap(repo): | ||
"""make a (src -> vfs) map for the repo""" | ||||
vfsmap = { | ||||
_srcstore: repo.svfs, | ||||
_srccache: repo.cachevfs, | ||||
} | ||||
# we keep repo.vfs out of the on purpose, ther are too many danger there | ||||
# (eg: .hg/hgrc) | ||||
assert repo.vfs not in vfsmap.values() | ||||
return vfsmap | ||||
|
r43346 | |||
|
r35820 | def _emit2(repo, entries, totalfilesize): | ||
|
r35774 | """actually emit the stream bundle""" | ||
|
r35785 | vfsmap = _makemap(repo) | ||
r48353 | # we keep repo.vfs out of the on purpose, ther are too many danger there | |||
# (eg: .hg/hgrc), | ||||
# | ||||
# this assert is duplicated (from _makemap) as author might think this is | ||||
# fine, while this is really not fine. | ||||
if repo.vfs in vfsmap.values(): | ||||
raise error.ProgrammingError( | ||||
b'repo.vfs must not be added to vfsmap for security reasons' | ||||
) | ||||
|
r43346 | progress = repo.ui.makeprogress( | ||
|
r43347 | _(b'bundle'), total=totalfilesize, unit=_(b'bytes') | ||
|
r43346 | ) | ||
|
r38368 | progress.update(0) | ||
|
r38393 | with maketempcopies() as copy, progress: | ||
# copy is delayed until we are in the try | ||||
entries = [_filterfull(e, copy, vfsmap) for e in entries] | ||||
|
r43346 | yield None # this release the lock on the repository | ||
|
r48592 | totalbytecount = 0 | ||
|
r35783 | |||
|
r38393 | for src, name, ftype, data in entries: | ||
vfs = vfsmap[src] | ||||
yield src | ||||
yield util.uvarintencode(len(name)) | ||||
if ftype == _fileappend: | ||||
fp = vfs(name) | ||||
size = data | ||||
elif ftype == _filefull: | ||||
|
r43347 | fp = open(data, b'rb') | ||
|
r38393 | size = util.fstat(fp).st_size | ||
|
r48592 | bytecount = 0 | ||
|
r38393 | try: | ||
yield util.uvarintencode(size) | ||||
yield name | ||||
if size <= 65536: | ||||
chunks = (fp.read(size),) | ||||
else: | ||||
chunks = util.filechunkiter(fp, limit=size) | ||||
for chunk in chunks: | ||||
|
r48592 | bytecount += len(chunk) | ||
totalbytecount += len(chunk) | ||||
progress.update(totalbytecount) | ||||
|
r38393 | yield chunk | ||
|
r48592 | if bytecount != size: | ||
# Would most likely be caused by a race due to `hg strip` or | ||||
# a revlog split | ||||
raise error.Abort( | ||||
_( | ||||
b'clone could only read %d bytes from %s, but ' | ||||
b'expected %d bytes' | ||||
) | ||||
% (bytecount, name, size) | ||||
) | ||||
|
r38393 | finally: | ||
fp.close() | ||||
|
r35774 | |||
|
r43346 | |||
r47748 | def _test_sync_point_walk_1(repo): | |||
"""a function for synchronisation during tests""" | ||||
def _test_sync_point_walk_2(repo): | ||||
"""a function for synchronisation during tests""" | ||||
r48237 | def _v2_walk(repo, includes, excludes, includeobsmarkers): | |||
"""emit a seris of files information useful to clone a repo | ||||
return (entries, totalfilesize) | ||||
entries is a list of tuple (vfs-key, file-path, file-type, size) | ||||
- `vfs-key`: is a key to the right vfs to write the file (see _makemap) | ||||
- `name`: file path of the file to copy (to be feed to the vfss) | ||||
- `file-type`: do this file need to be copied with the source lock ? | ||||
- `size`: the size of the file (or None) | ||||
""" | ||||
assert repo._currentlock(repo._lockref) is not None | ||||
entries = [] | ||||
totalfilesize = 0 | ||||
matcher = None | ||||
if includes or excludes: | ||||
matcher = narrowspec.match(repo.root, includes, excludes) | ||||
|
r48691 | for rl_type, name, size in _walkstreamfiles(repo, matcher): | ||
r48237 | if size: | |||
ft = _fileappend | ||||
if rl_type & store.FILEFLAGS_VOLATILE: | ||||
ft = _filefull | ||||
entries.append((_srcstore, name, ft, size)) | ||||
totalfilesize += size | ||||
for name in _walkstreamfullstorefiles(repo): | ||||
if repo.svfs.exists(name): | ||||
totalfilesize += repo.svfs.lstat(name).st_size | ||||
entries.append((_srcstore, name, _filefull, None)) | ||||
if includeobsmarkers and repo.svfs.exists(b'obsstore'): | ||||
totalfilesize += repo.svfs.lstat(b'obsstore').st_size | ||||
entries.append((_srcstore, b'obsstore', _filefull, None)) | ||||
for name in cacheutil.cachetocopy(repo): | ||||
if repo.cachevfs.exists(name): | ||||
totalfilesize += repo.cachevfs.lstat(name).st_size | ||||
entries.append((_srccache, name, _filefull, None)) | ||||
return entries, totalfilesize | ||||
r40434 | def generatev2(repo, includes, excludes, includeobsmarkers): | |||
|
r35774 | """Emit content for version 2 of a streaming clone. | ||
the data stream consists the following entries: | ||||
|
r35785 | 1) A char representing the file destination (eg: store or cache) | ||
2) A varint containing the length of the filename | ||||
3) A varint containing the length of file data | ||||
4) N bytes containing the filename (the internal, store-agnostic form) | ||||
5) N bytes containing the file data | ||||
|
r35774 | |||
Returns a 3-tuple of (file count, file size, data iterator). | ||||
""" | ||||
with repo.lock(): | ||||
r48237 | repo.ui.debug(b'scanning\n') | |||
|
r40375 | |||
r48237 | entries, totalfilesize = _v2_walk( | |||
repo, | ||||
includes=includes, | ||||
excludes=excludes, | ||||
includeobsmarkers=includeobsmarkers, | ||||
) | ||||
|
r35774 | |||
|
r35820 | chunks = _emit2(repo, entries, totalfilesize) | ||
|
r35783 | first = next(chunks) | ||
assert first is None | ||||
r47748 | _test_sync_point_walk_1(repo) | |||
_test_sync_point_walk_2(repo) | ||||
|
r35774 | |||
return len(entries), totalfilesize, chunks | ||||
|
r43346 | |||
|
r35785 | @contextlib.contextmanager | ||
def nested(*ctxs): | ||||
|
r39793 | this = ctxs[0] | ||
rest = ctxs[1:] | ||||
with this: | ||||
if rest: | ||||
with nested(*rest): | ||||
yield | ||||
else: | ||||
|
r35785 | yield | ||
|
r43346 | |||
|
r35774 | def consumev2(repo, fp, filecount, filesize): | ||
"""Apply the contents from a version 2 streaming clone. | ||||
Data is read from an object that only needs to provide a ``read(size)`` | ||||
method. | ||||
""" | ||||
with repo.lock(): | ||||
|
r43346 | repo.ui.status( | ||
|
r43347 | _(b'%d files to transfer, %s of data\n') | ||
|
r43346 | % (filecount, util.bytecount(filesize)) | ||
) | ||||
|
r35774 | |||
start = util.timer() | ||||
|
r43346 | progress = repo.ui.makeprogress( | ||
|
r43347 | _(b'clone'), total=filesize, unit=_(b'bytes') | ||
|
r43346 | ) | ||
|
r38368 | progress.update(0) | ||
|
r35774 | |||
|
r35785 | vfsmap = _makemap(repo) | ||
r48353 | # we keep repo.vfs out of the on purpose, ther are too many danger | |||
# there (eg: .hg/hgrc), | ||||
# | ||||
# this assert is duplicated (from _makemap) as author might think this | ||||
# is fine, while this is really not fine. | ||||
if repo.vfs in vfsmap.values(): | ||||
raise error.ProgrammingError( | ||||
b'repo.vfs must not be added to vfsmap for security reasons' | ||||
) | ||||
|
r35774 | |||
|
r43347 | with repo.transaction(b'clone'): | ||
|
r43346 | ctxs = (vfs.backgroundclosing(repo.ui) for vfs in vfsmap.values()) | ||
|
r35785 | with nested(*ctxs): | ||
|
r35774 | for i in range(filecount): | ||
|
r35821 | src = util.readexactly(fp, 1) | ||
|
r35785 | vfs = vfsmap[src] | ||
|
r35774 | namelen = util.uvarintdecodestream(fp) | ||
datalen = util.uvarintdecodestream(fp) | ||||
|
r35821 | name = util.readexactly(fp, namelen) | ||
|
r35774 | |||
if repo.ui.debugflag: | ||||
|
r43346 | repo.ui.debug( | ||
|
r43347 | b'adding [%s] %s (%s)\n' | ||
|
r43346 | % (src, name, util.bytecount(datalen)) | ||
) | ||||
|
r35774 | |||
|
r43347 | with vfs(name, b'w') as ofp: | ||
|
r35774 | for chunk in util.filechunkiter(fp, limit=datalen): | ||
|
r38368 | progress.increment(step=len(chunk)) | ||
|
r35774 | ofp.write(chunk) | ||
# force @filecache properties to be reloaded from | ||||
# streamclone-ed file at next access | ||||
repo.invalidate(clearfilecache=True) | ||||
elapsed = util.timer() - start | ||||
if elapsed <= 0: | ||||
elapsed = 0.001 | ||||
|
r43346 | repo.ui.status( | ||
|
r43347 | _(b'transferred %s in %.1f seconds (%s/sec)\n') | ||
|
r43346 | % ( | ||
util.bytecount(progress.pos), | ||||
elapsed, | ||||
util.bytecount(progress.pos / elapsed), | ||||
) | ||||
) | ||||
|
r38392 | progress.complete() | ||
|
r35774 | |||
|
r43346 | |||
|
r35774 | def applybundlev2(repo, fp, filecount, filesize, requirements): | ||
|
r39736 | from . import localrepo | ||
|
r35774 | missingreqs = [r for r in requirements if r not in repo.supported] | ||
if missingreqs: | ||||
|
r43346 | raise error.Abort( | ||
|
r43387 | _(b'unable to apply stream clone: unsupported format: %s') | ||
|
r43347 | % b', '.join(sorted(missingreqs)) | ||
|
r43346 | ) | ||
|
r35774 | |||
consumev2(repo, fp, filecount, filesize) | ||||
|
r35822 | |||
r49442 | repo.requirements = new_stream_clone_requirements( | |||
repo.supportedformats, | ||||
repo.requirements, | ||||
requirements, | ||||
|
r43346 | ) | ||
|
r39736 | repo.svfs.options = localrepo.resolvestorevfsoptions( | ||
|
r43346 | repo.ui, repo.requirements, repo.features | ||
) | ||||
|
r45666 | scmutil.writereporequirements(repo) | ||
r48240 | ||||
def _copy_files(src_vfs_map, dst_vfs_map, entries, progress): | ||||
hardlink = [True] | ||||
def copy_used(): | ||||
hardlink[0] = False | ||||
progress.topic = _(b'copying') | ||||
for k, path, size in entries: | ||||
src_vfs = src_vfs_map[k] | ||||
dst_vfs = dst_vfs_map[k] | ||||
src_path = src_vfs.join(path) | ||||
dst_path = dst_vfs.join(path) | ||||
r48685 | # We cannot use dirname and makedirs of dst_vfs here because the store | |||
# encoding confuses them. See issue 6581 for details. | ||||
dirname = os.path.dirname(dst_path) | ||||
if not os.path.exists(dirname): | ||||
util.makedirs(dirname) | ||||
r48240 | dst_vfs.register_file(path) | |||
# XXX we could use the #nb_bytes argument. | ||||
util.copyfile( | ||||
src_path, | ||||
dst_path, | ||||
hardlink=hardlink[0], | ||||
no_hardlink_cb=copy_used, | ||||
check_fs_hardlink=False, | ||||
) | ||||
progress.increment() | ||||
return hardlink[0] | ||||
def local_copy(src_repo, dest_repo): | ||||
"""copy all content from one local repository to another | ||||
This is useful for local clone""" | ||||
src_store_requirements = { | ||||
r | ||||
for r in src_repo.requirements | ||||
if r not in requirementsmod.WORKING_DIR_REQUIREMENTS | ||||
} | ||||
dest_store_requirements = { | ||||
r | ||||
for r in dest_repo.requirements | ||||
if r not in requirementsmod.WORKING_DIR_REQUIREMENTS | ||||
} | ||||
assert src_store_requirements == dest_store_requirements | ||||
with dest_repo.lock(): | ||||
with src_repo.lock(): | ||||
r48241 | ||||
# bookmark is not integrated to the streaming as it might use the | ||||
# `repo.vfs` and they are too many sentitive data accessible | ||||
# through `repo.vfs` to expose it to streaming clone. | ||||
src_book_vfs = bookmarks.bookmarksvfs(src_repo) | ||||
srcbookmarks = src_book_vfs.join(b'bookmarks') | ||||
bm_count = 0 | ||||
if os.path.exists(srcbookmarks): | ||||
bm_count = 1 | ||||
r48240 | entries, totalfilesize = _v2_walk( | |||
src_repo, | ||||
includes=None, | ||||
excludes=None, | ||||
includeobsmarkers=True, | ||||
) | ||||
src_vfs_map = _makemap(src_repo) | ||||
dest_vfs_map = _makemap(dest_repo) | ||||
progress = src_repo.ui.makeprogress( | ||||
topic=_(b'linking'), | ||||
r48241 | total=len(entries) + bm_count, | |||
r48240 | unit=_(b'files'), | |||
) | ||||
# copy files | ||||
# | ||||
# We could copy the full file while the source repository is locked | ||||
# and the other one without the lock. However, in the linking case, | ||||
# this would also requires checks that nobody is appending any data | ||||
# to the files while we do the clone, so this is not done yet. We | ||||
# could do this blindly when copying files. | ||||
files = ((k, path, size) for k, path, ftype, size in entries) | ||||
hardlink = _copy_files(src_vfs_map, dest_vfs_map, files, progress) | ||||
# copy bookmarks over | ||||
r48241 | if bm_count: | |||
dst_book_vfs = bookmarks.bookmarksvfs(dest_repo) | ||||
dstbookmarks = dst_book_vfs.join(b'bookmarks') | ||||
r48240 | util.copyfile(srcbookmarks, dstbookmarks) | |||
progress.complete() | ||||
if hardlink: | ||||
msg = b'linked %d files\n' | ||||
else: | ||||
msg = b'copied %d files\n' | ||||
r48241 | src_repo.ui.debug(msg % (len(entries) + bm_count)) | |||
r48240 | ||||
with dest_repo.transaction(b"localclone") as tr: | ||||
dest_repo.store.write(tr) | ||||
# clean up transaction file as they do not make sense | ||||
undo_files = [(dest_repo.svfs, b'undo.backupfiles')] | ||||
undo_files.extend(dest_repo.undofiles()) | ||||
for undovfs, undofile in undo_files: | ||||
try: | ||||
undovfs.unlink(undofile) | ||||
except OSError as e: | ||||
if e.errno != errno.ENOENT: | ||||
msg = _(b'error removing %s: %s\n') | ||||
path = undovfs.join(undofile) | ||||
e_msg = stringutil.forcebytestr(e) | ||||
msg %= (path, e_msg) | ||||
dest_repo.ui.warn(msg) | ||||