This is an archive of the discontinued Mercurial Phabricator instance.

Differential D10852

streamingclone: extract the scanning part from the generation part
ClosedPublic

Authored by marmoute on Jun 9 2021, 10:37 AM.

Details

Reviewers
pulkit
Alphare
Group Reviewers
hg-reviewers
Commits
rHG2f4ca4807033: streamingclone: extract the scanning part from the generation part
Summary

We will reuse the scanning part for local clone, so we need it in a dedicated
function.

Diff Detail

Repository
rHG Mercurial
Branch
default
Lint
No Linters Available
Unit
No Unit Test Coverage

Event Timeline

marmoute created this revision.Jun 9 2021, 10:37 AM
pulkit accepted this revision.Jun 14 2021, 4:43 PM
This revision is now accepted and ready to land.Jun 14 2021, 4:43 PM
baymax updated this revision to Diff 28574.Jun 15 2021, 10:23 AM

✅ refresh by Heptapod after a successful CI run (🐙 💚)

Alphare accepted this revision.Jun 16 2021, 4:12 AM
baymax updated this revision to Diff 28622.Jun 21 2021, 1:09 AM

✅ refresh by Heptapod after a successful CI run (🐙 💚)

Revision Contents
Changeset List

PathPackages
Mmercurial/streamclone.py (73 lines)

Diff 28622

mercurial/streamclone.py

def _test_sync_point_walk_1(repo): def _test_sync_point_walk_1(repo):
"""a function for synchronisation during tests""" """a function for synchronisation during tests"""
def _test_sync_point_walk_2(repo): def _test_sync_point_walk_2(repo):
"""a function for synchronisation during tests""" """a function for synchronisation during tests"""
def generatev2(repo, includes, excludes, includeobsmarkers): def _v2_walk(repo, includes, excludes, includeobsmarkers):
"""Emit content for version 2 of a streaming clone. """emit a seris of files information useful to clone a repo
the data stream consists the following entries: return (entries, totalfilesize)
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
Returns a 3-tuple of (file count, file size, data iterator). entries is a list of tuple (vfs-key, file-path, file-type, size)
"""
with repo.lock():
- `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 = [] entries = []
totalfilesize = 0 totalfilesize = 0
matcher = None matcher = None
if includes or excludes: if includes or excludes:
matcher = narrowspec.match(repo.root, includes, excludes) matcher = narrowspec.match(repo.root, includes, excludes)
repo.ui.debug(b'scanning\n')
for rl_type, name, ename, size in _walkstreamfiles(repo, matcher): for rl_type, name, ename, size in _walkstreamfiles(repo, matcher):
if size: if size:
ft = _fileappend ft = _fileappend
if rl_type & store.FILEFLAGS_VOLATILE: if rl_type & store.FILEFLAGS_VOLATILE:
ft = _filefull ft = _filefull
entries.append((_srcstore, name, ft, size)) entries.append((_srcstore, name, ft, size))
totalfilesize += size totalfilesize += size
for name in _walkstreamfullstorefiles(repo): for name in _walkstreamfullstorefiles(repo):
if repo.svfs.exists(name): if repo.svfs.exists(name):
totalfilesize += repo.svfs.lstat(name).st_size totalfilesize += repo.svfs.lstat(name).st_size
entries.append((_srcstore, name, _filefull, None)) entries.append((_srcstore, name, _filefull, None))
if includeobsmarkers and repo.svfs.exists(b'obsstore'): if includeobsmarkers and repo.svfs.exists(b'obsstore'):
totalfilesize += repo.svfs.lstat(b'obsstore').st_size totalfilesize += repo.svfs.lstat(b'obsstore').st_size
entries.append((_srcstore, b'obsstore', _filefull, None)) entries.append((_srcstore, b'obsstore', _filefull, None))
for name in cacheutil.cachetocopy(repo): for name in cacheutil.cachetocopy(repo):
if repo.cachevfs.exists(name): if repo.cachevfs.exists(name):
totalfilesize += repo.cachevfs.lstat(name).st_size totalfilesize += repo.cachevfs.lstat(name).st_size
entries.append((_srccache, name, _filefull, None)) entries.append((_srccache, name, _filefull, None))
return entries, totalfilesize
def generatev2(repo, includes, excludes, includeobsmarkers):
"""Emit content for version 2 of a streaming clone.
the data stream consists the following entries:
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
Returns a 3-tuple of (file count, file size, data iterator).
"""
with repo.lock():
repo.ui.debug(b'scanning\n')
entries, totalfilesize = _v2_walk(
repo,
includes=includes,
excludes=excludes,
includeobsmarkers=includeobsmarkers,
)
chunks = _emit2(repo, entries, totalfilesize) chunks = _emit2(repo, entries, totalfilesize)
first = next(chunks) first = next(chunks)
assert first is None assert first is None
_test_sync_point_walk_1(repo) _test_sync_point_walk_1(repo)
_test_sync_point_walk_2(repo) _test_sync_point_walk_2(repo)
return len(entries), totalfilesize, chunks return len(entries), totalfilesize, chunks