This is an archive of the discontinued Mercurial Phabricator instance.

Differential D8587

metadata: move computation related to files touched in a dedicated module
ClosedPublic

Authored by marmoute on May 27 2020, 9:16 AM.

Details

Reviewers
durin42
pulkit
Group Reviewers
hg-reviewers
Commits
rHG4c1d39215034: metadata: move computation related to files touched in a dedicated module
Summary

This was suggested by Yuya Nishihara a while back. Since I am about to add more
metadata related computation, lets create a new repositories.

Diff Detail

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

Event Timeline

marmoute created this revision.May 27 2020, 9:16 AM
durin42 requested changes to this revision.May 28 2020, 11:37 AM
durin42 added a subscriber: durin42.
durin42 added inline comments.
mercurial/metadata.py
3

This should read "Google, Inc." instead of Martin's name.

This revision now requires changes to proceed.May 28 2020, 11:37 AM
marmoute updated this revision to Diff 21545.Jun 2 2020, 6:48 AM
pulkit added a subscriber: pulkit.Jun 9 2020, 5:12 AM
pulkit added inline comments.
mercurial/metadata.py
3

And I guess no email.

pulkit accepted this revision.Jun 9 2020, 5:44 AM
pulkit added inline comments.
mercurial/metadata.py
3

Will fix in flight to # Copyright 2019 Google Inc.

pulkit requested changes to this revision.Jun 9 2020, 5:54 AM

Fails to apply on current tip of default branch, kindly rebase and resend.

This revision now requires changes to proceed.Jun 9 2020, 5:54 AM
marmoute updated this revision to Diff 21581.Jun 9 2020, 6:16 AM
marmoute added a comment.Jun 9 2020, 6:41 AM
In D8587#128674, @pulkit wrote:

Fails to apply on current tip of default branch, kindly rebase and resend.

rebased and resent.

marmoute added a commit: rHG4c1d39215034: metadata: move computation related to files touched in a dedicated module.Jun 9 2020, 7:33 AM
This revision was not accepted when it landed; it landed in state Needs Review.
Closed by commit rHG4c1d39215034: metadata: move computation related to files touched in a dedicated module (authored by marmoute). · Explain Why
This revision was automatically updated to reflect the committed changes.
martinvonz added a subscriber: martinvonz.Jun 10 2020, 10:58 AM

Worth mentioning in relnotes? (This has been queued, so that would be a followup.)

marmoute added a comment.Jun 10 2020, 10:59 AM
In D8587#128841, @martinvonz wrote:

Worth mentioning in relnotes? (This has been queued, so that would be a followup.)

I don't think so, what would you mention ?

martinvonz added a comment.Jun 10 2020, 11:05 AM
In D8587#128842, @marmoute wrote:
In D8587#128841, @martinvonz wrote:

Worth mentioning in relnotes? (This has been queued, so that would be a followup.)

I don't think so, what would you mention ?

We have a section for API changes. This patch affects many functions. (I suppose you might argue that we don't need to mention changes in private APIs (name prefixed by _), but these are not all private.)

marmoute added a comment.Jun 10 2020, 11:07 AM
In D8587#128843, @martinvonz wrote:
In D8587#128842, @marmoute wrote:
In D8587#128841, @martinvonz wrote:

Worth mentioning in relnotes? (This has been queued, so that would be a followup.)

I don't think so, what would you mention ?

We have a section for API changes. This patch affects many functions. (I suppose you might argue that we don't need to mention changes in private APIs (name prefixed by _), but these are not all private.)

These are private API used by experimental feature. I really do not expect that to affect anyone.

martinvonz added a comment.Jun 10 2020, 11:11 AM
In D8587#128844, @marmoute wrote:
In D8587#128843, @martinvonz wrote:
In D8587#128842, @marmoute wrote:
In D8587#128841, @martinvonz wrote:

Worth mentioning in relnotes? (This has been queued, so that would be a followup.)

I don't think so, what would you mention ?

We have a section for API changes. This patch affects many functions. (I suppose you might argue that we don't need to mention changes in private APIs (name prefixed by _), but these are not all private.)

These are private API used by experimental feature. I really do not expect that to affect anyone.

I was thinking mostly about computechangesetfiles{added,removed}(), which I think are used even when not using experimental features. We happen to override those, but we also never read the release notes, so it's fine not mention it :)

Revision Contents
Changeset List

PathPackages
Mmercurial/changelog.py (18 lines)
Mmercurial/context.py (8 lines)
Mmercurial/copies.py (250 lines)
AMmercurial/metadata.py (268 lines)
Mmercurial/upgrade.py (6 lines)

Diff 21487

mercurial/changelog.py

from .node import ( from .node import (
bin, bin,
hex, hex,
nullid, nullid,
) )
from .thirdparty import attr from .thirdparty import attr
from . import ( from . import (
copies,
encoding, encoding,
error, error,
metadata,
pycompat, pycompat,
revlog, revlog,
) )
from .utils import ( from .utils import (
dateutil, dateutil,
stringutil, stringutil,
) )
if self._cpsd: if self._cpsd:
rawindices = self._sidedata.get(sidedatamod.SD_FILESADDED) rawindices = self._sidedata.get(sidedatamod.SD_FILESADDED)
if not rawindices: if not rawindices:
return [] return []
else: else:
rawindices = self.extra.get(b'filesadded') rawindices = self.extra.get(b'filesadded')
if rawindices is None: if rawindices is None:
return None return None
return copies.decodefileindices(self.files, rawindices) return metadata.decodefileindices(self.files, rawindices)
@property @property
def filesremoved(self): def filesremoved(self):
if self._cpsd: if self._cpsd:
rawindices = self._sidedata.get(sidedatamod.SD_FILESREMOVED) rawindices = self._sidedata.get(sidedatamod.SD_FILESREMOVED)
if not rawindices: if not rawindices:
return [] return []
else: else:
rawindices = self.extra.get(b'filesremoved') rawindices = self.extra.get(b'filesremoved')
if rawindices is None: if rawindices is None:
return None return None
return copies.decodefileindices(self.files, rawindices) return metadata.decodefileindices(self.files, rawindices)
@property @property
def p1copies(self): def p1copies(self):
if self._cpsd: if self._cpsd:
rawcopies = self._sidedata.get(sidedatamod.SD_P1COPIES) rawcopies = self._sidedata.get(sidedatamod.SD_P1COPIES)
if not rawcopies: if not rawcopies:
return {} return {}
else: else:
rawcopies = self.extra.get(b'p1copies') rawcopies = self.extra.get(b'p1copies')
if rawcopies is None: if rawcopies is None:
return None return None
return copies.decodecopies(self.files, rawcopies) return metadata.decodecopies(self.files, rawcopies)
@property @property
def p2copies(self): def p2copies(self):
if self._cpsd: if self._cpsd:
rawcopies = self._sidedata.get(sidedatamod.SD_P2COPIES) rawcopies = self._sidedata.get(sidedatamod.SD_P2COPIES)
if not rawcopies: if not rawcopies:
return {} return {}
else: else:
rawcopies = self.extra.get(b'p2copies') rawcopies = self.extra.get(b'p2copies')
if rawcopies is None: if rawcopies is None:
return None return None
return copies.decodecopies(self.files, rawcopies) return metadata.decodecopies(self.files, rawcopies)
@property @property
def description(self): def description(self):
return encoding.tolocal(self._text[self._offsets[3] + 2 :]) return encoding.tolocal(self._text[self._offsets[3] + 2 :])
class changelog(revlog.revlog): class changelog(revlog.revlog):
def __init__(self, opener, trypending=False): def __init__(self, opener, trypending=False):
for name in ( for name in (
b'p1copies', b'p1copies',
b'p2copies', b'p2copies',
b'filesadded', b'filesadded',
b'filesremoved', b'filesremoved',
): ):
extra.pop(name, None) extra.pop(name, None)
if p1copies is not None: if p1copies is not None:
p1copies = copies.encodecopies(sortedfiles, p1copies) p1copies = metadata.encodecopies(sortedfiles, p1copies)
if p2copies is not None: if p2copies is not None:
p2copies = copies.encodecopies(sortedfiles, p2copies) p2copies = metadata.encodecopies(sortedfiles, p2copies)
if filesadded is not None: if filesadded is not None:
filesadded = copies.encodefileindices(sortedfiles, filesadded) filesadded = metadata.encodefileindices(sortedfiles, filesadded)
if filesremoved is not None: if filesremoved is not None:
filesremoved = copies.encodefileindices(sortedfiles, filesremoved) filesremoved = metadata.encodefileindices(sortedfiles, filesremoved)
if self._copiesstorage == b'extra': if self._copiesstorage == b'extra':
extrasentries = p1copies, p2copies, filesadded, filesremoved extrasentries = p1copies, p2copies, filesadded, filesremoved
if extra is None and any(x is not None for x in extrasentries): if extra is None and any(x is not None for x in extrasentries):
extra = {} extra = {}
if p1copies is not None: if p1copies is not None:
extra[b'p1copies'] = p1copies extra[b'p1copies'] = p1copies
if p2copies is not None: if p2copies is not None:
extra[b'p2copies'] = p2copies extra[b'p2copies'] = p2copies

mercurial/context.py

wdirfilenodeids, wdirfilenodeids,
wdirhex, wdirhex,
) )
from .pycompat import ( from .pycompat import (
getattr, getattr,
open, open,
) )
from . import ( from . import (
copies,
dagop, dagop,
encoding, encoding,
error, error,
fileset, fileset,
match as matchmod, match as matchmod,
metadata,
obsolete as obsmod, obsolete as obsmod,
patch, patch,
pathutil, pathutil,
phases, phases,
pycompat, pycompat,
repoview, repoview,
scmutil, scmutil,
sparse, sparse,
def flags(self, path): def flags(self, path):
try: try:
return self._fileinfo(path)[1] return self._fileinfo(path)[1]
except error.LookupError: except error.LookupError:
return b'' return b''
@propertycache @propertycache
def _copies(self): def _copies(self):
return copies.computechangesetcopies(self) return metadata.computechangesetcopies(self)
def p1copies(self): def p1copies(self):
return self._copies[0] return self._copies[0]
def p2copies(self): def p2copies(self):
return self._copies[1] return self._copies[1]
def sub(self, path, allowcreate=True): def sub(self, path, allowcreate=True):
source = self._repo.ui.config(b'experimental', b'copies.read-from') source = self._repo.ui.config(b'experimental', b'copies.read-from')
if source == b'changeset-only': if source == b'changeset-only':
compute_on_none = False compute_on_none = False
elif source != b'compatibility': elif source != b'compatibility':
# filelog mode, ignore any changelog content # filelog mode, ignore any changelog content
filesadded = None filesadded = None
if filesadded is None: if filesadded is None:
if compute_on_none: if compute_on_none:
filesadded = copies.computechangesetfilesadded(self) filesadded = metadata.computechangesetfilesadded(self)
else: else:
filesadded = [] filesadded = []
return filesadded return filesadded
def filesremoved(self): def filesremoved(self):
filesremoved = self._changeset.filesremoved filesremoved = self._changeset.filesremoved
compute_on_none = True compute_on_none = True
if self._repo.filecopiesmode == b'changeset-sidedata': if self._repo.filecopiesmode == b'changeset-sidedata':
compute_on_none = False compute_on_none = False
else: else:
source = self._repo.ui.config(b'experimental', b'copies.read-from') source = self._repo.ui.config(b'experimental', b'copies.read-from')
if source == b'changeset-only': if source == b'changeset-only':
compute_on_none = False compute_on_none = False
elif source != b'compatibility': elif source != b'compatibility':
# filelog mode, ignore any changelog content # filelog mode, ignore any changelog content
filesremoved = None filesremoved = None
if filesremoved is None: if filesremoved is None:
if compute_on_none: if compute_on_none:
filesremoved = copies.computechangesetfilesremoved(self) filesremoved = metadata.computechangesetfilesremoved(self)
else: else:
filesremoved = [] filesremoved = []
return filesremoved return filesremoved
@propertycache @propertycache
def _copies(self): def _copies(self):
p1copies = self._changeset.p1copies p1copies = self._changeset.p1copies
p2copies = self._changeset.p2copies p2copies = self._changeset.p2copies

mercurial/copies.py

# copies.py - copy detection for Mercurial # copies.py - copy detection for Mercurial
# #
# Copyright 2008 Matt Mackall <mpm@selenic.com> # Copyright 2008 Matt Mackall <mpm@selenic.com>
# #
# This software may be used and distributed according to the terms of the # This software may be used and distributed according to the terms of the
# GNU General Public License version 2 or any later version. # GNU General Public License version 2 or any later version.
from __future__ import absolute_import from __future__ import absolute_import
import collections import collections
import multiprocessing
import os import os
from .i18n import _ from .i18n import _
from .revlogutils.flagutil import REVIDX_SIDEDATA from .revlogutils.flagutil import REVIDX_SIDEDATA
from . import ( from . import (
error,
match as matchmod, match as matchmod,
node, node,
pathutil, pathutil,
pycompat, pycompat,
util, util,
) )
from .revlogutils import sidedata as sidedatamod
from .utils import stringutil from .utils import stringutil
def _filter(src, dst, t): def _filter(src, dst, t):
"""filters out invalid copies after chaining""" """filters out invalid copies after chaining"""
# When _chain()'ing copies in 'a' (from 'src' via some other commit 'mid') # When _chain()'ing copies in 'a' (from 'src' via some other commit 'mid')
merge._related()). It will also not mark copies if the file wasn't modified merge._related()). It will also not mark copies if the file wasn't modified
on the local side. This function adds the copies that were "missed" on the local side. This function adds the copies that were "missed"
by merge.update(). by merge.update().
""" """
new_copies = pathcopies(base, ctx) new_copies = pathcopies(base, ctx)
_filter(wctx.p1(), wctx, new_copies) _filter(wctx.p1(), wctx, new_copies)
for dst, src in pycompat.iteritems(new_copies): for dst, src in pycompat.iteritems(new_copies):
wctx[dst].markcopied(src) wctx[dst].markcopied(src)
def computechangesetfilesadded(ctx):
"""return the list of files added in a changeset
"""
added = []
for f in ctx.files():
if not any(f in p for p in ctx.parents()):
added.append(f)
return added
def computechangesetfilesremoved(ctx):
"""return the list of files removed in a changeset
"""
removed = []
for f in ctx.files():
if f not in ctx:
removed.append(f)
return removed
def computechangesetcopies(ctx):
"""return the copies data for a changeset
The copies data are returned as a pair of dictionnary (p1copies, p2copies).
Each dictionnary are in the form: `{newname: oldname}`
"""
p1copies = {}
p2copies = {}
p1 = ctx.p1()
p2 = ctx.p2()
narrowmatch = ctx._repo.narrowmatch()
for dst in ctx.files():
if not narrowmatch(dst) or dst not in ctx:
continue
copied = ctx[dst].renamed()
if not copied:
continue
src, srcnode = copied
if src in p1 and p1[src].filenode() == srcnode:
p1copies[dst] = src
elif src in p2 and p2[src].filenode() == srcnode:
p2copies[dst] = src
return p1copies, p2copies
def encodecopies(files, copies):
items = []
for i, dst in enumerate(files):
if dst in copies:
items.append(b'%d\0%s' % (i, copies[dst]))
if len(items) != len(copies):
raise error.ProgrammingError(
b'some copy targets missing from file list'
)
return b"\n".join(items)
def decodecopies(files, data):
try:
copies = {}
if not data:
return copies
for l in data.split(b'\n'):
strindex, src = l.split(b'\0')
i = int(strindex)
dst = files[i]
copies[dst] = src
return copies
except (ValueError, IndexError):
# Perhaps someone had chosen the same key name (e.g. "p1copies") and
# used different syntax for the value.
return None
def encodefileindices(files, subset):
subset = set(subset)
indices = []
for i, f in enumerate(files):
if f in subset:
indices.append(b'%d' % i)
return b'\n'.join(indices)
def decodefileindices(files, data):
try:
subset = []
if not data:
return subset
for strindex in data.split(b'\n'):
i = int(strindex)
if i < 0 or i >= len(files):
return None
subset.append(files[i])
return subset
except (ValueError, IndexError):
# Perhaps someone had chosen the same key name (e.g. "added") and
# used different syntax for the value.
return None
def _getsidedata(srcrepo, rev):
ctx = srcrepo[rev]
filescopies = computechangesetcopies(ctx)
filesadded = computechangesetfilesadded(ctx)
filesremoved = computechangesetfilesremoved(ctx)
sidedata = {}
if any([filescopies, filesadded, filesremoved]):
sortedfiles = sorted(ctx.files())
p1copies, p2copies = filescopies
p1copies = encodecopies(sortedfiles, p1copies)
p2copies = encodecopies(sortedfiles, p2copies)
filesadded = encodefileindices(sortedfiles, filesadded)
filesremoved = encodefileindices(sortedfiles, filesremoved)
if p1copies:
sidedata[sidedatamod.SD_P1COPIES] = p1copies
if p2copies:
sidedata[sidedatamod.SD_P2COPIES] = p2copies
if filesadded:
sidedata[sidedatamod.SD_FILESADDED] = filesadded
if filesremoved:
sidedata[sidedatamod.SD_FILESREMOVED] = filesremoved
return sidedata
def getsidedataadder(srcrepo, destrepo):
use_w = srcrepo.ui.configbool(b'experimental', b'worker.repository-upgrade')
if pycompat.iswindows or not use_w:
return _get_simple_sidedata_adder(srcrepo, destrepo)
else:
return _get_worker_sidedata_adder(srcrepo, destrepo)
def _sidedata_worker(srcrepo, revs_queue, sidedata_queue, tokens):
"""The function used by worker precomputing sidedata
It read an input queue containing revision numbers
It write in an output queue containing (rev, <sidedata-map>)
The `None` input value is used as a stop signal.
The `tokens` semaphore is user to avoid having too many unprocessed
entries. The workers needs to acquire one token before fetching a task.
They will be released by the consumer of the produced data.
"""
tokens.acquire()
rev = revs_queue.get()
while rev is not None:
data = _getsidedata(srcrepo, rev)
sidedata_queue.put((rev, data))
tokens.acquire()
rev = revs_queue.get()
# processing of `None` is completed, release the token.
tokens.release()
BUFF_PER_WORKER = 50
def _get_worker_sidedata_adder(srcrepo, destrepo):
"""The parallel version of the sidedata computation
This code spawn a pool of worker that precompute a buffer of sidedata
before we actually need them"""
# avoid circular import copies -> scmutil -> worker -> copies
from . import worker
nbworkers = worker._numworkers(srcrepo.ui)
tokens = multiprocessing.BoundedSemaphore(nbworkers * BUFF_PER_WORKER)
revsq = multiprocessing.Queue()
sidedataq = multiprocessing.Queue()
assert srcrepo.filtername is None
# queue all tasks beforehand, revision numbers are small and it make
# synchronisation simpler
#
# Since the computation for each node can be quite expensive, the overhead
# of using a single queue is not revelant. In practice, most computation
# are fast but some are very expensive and dominate all the other smaller
# cost.
for r in srcrepo.changelog.revs():
revsq.put(r)
# queue the "no more tasks" markers
for i in range(nbworkers):
revsq.put(None)
allworkers = []
for i in range(nbworkers):
args = (srcrepo, revsq, sidedataq, tokens)
w = multiprocessing.Process(target=_sidedata_worker, args=args)
allworkers.append(w)
w.start()
# dictionnary to store results for revision higher than we one we are
# looking for. For example, if we need the sidedatamap for 42, and 43 is
# received, when shelve 43 for later use.
staging = {}
def sidedata_companion(revlog, rev):
sidedata = {}
if util.safehasattr(revlog, b'filteredrevs'): # this is a changelog
# Is the data previously shelved ?
sidedata = staging.pop(rev, None)
if sidedata is None:
# look at the queued result until we find the one we are lookig
# for (shelve the other ones)
r, sidedata = sidedataq.get()
while r != rev:
staging[r] = sidedata
r, sidedata = sidedataq.get()
tokens.release()
return False, (), sidedata
return sidedata_companion
def _get_simple_sidedata_adder(srcrepo, destrepo):
"""The simple version of the sidedata computation
It just compute it in the same thread on request"""
def sidedatacompanion(revlog, rev):
sidedata = {}
if util.safehasattr(revlog, 'filteredrevs'): # this is a changelog
sidedata = _getsidedata(srcrepo, rev)
return False, (), sidedata
return sidedatacompanion
def getsidedataremover(srcrepo, destrepo):
def sidedatacompanion(revlog, rev):
f = ()
if util.safehasattr(revlog, 'filteredrevs'): # this is a changelog
if revlog.flags(rev) & REVIDX_SIDEDATA:
f = (
sidedatamod.SD_P1COPIES,
sidedatamod.SD_P2COPIES,
sidedatamod.SD_FILESADDED,
sidedatamod.SD_FILESREMOVED,
)
return False, f, {}
return sidedatacompanion

mercurial/metadata.py

  • This file was added.
# metadata.py -- code related to various metadata computation and access.
#
# Copyright 2019 Martin von Zweigbergk <martinvonz@google.com>
durin42Unsubmitted
Not Done

This should read "Google, Inc." instead of Martin's name.

durin42: This should read "Google, Inc." instead of Martin's name.
pulkitUnsubmitted
Not Done

And I guess no email.

pulkit: And I guess no email.
pulkitUnsubmitted
Not Done

Will fix in flight to # Copyright 2019 Google Inc.

pulkit: Will fix in flight to `# Copyright 2019 Google Inc.`
# Copyright 2020 Pierre-Yves David <pierre-yves.david@octobus.net>
#
# 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, print_function
import multiprocessing
from . import (
error,
pycompat,
util,
)
from .revlogutils import (
flagutil as sidedataflag,
sidedata as sidedatamod,
)
def computechangesetfilesadded(ctx):
"""return the list of files added in a changeset
"""
added = []
for f in ctx.files():
if not any(f in p for p in ctx.parents()):
added.append(f)
return added
def computechangesetfilesremoved(ctx):
"""return the list of files removed in a changeset
"""
removed = []
for f in ctx.files():
if f not in ctx:
removed.append(f)
return removed
def computechangesetcopies(ctx):
"""return the copies data for a changeset
The copies data are returned as a pair of dictionnary (p1copies, p2copies).
Each dictionnary are in the form: `{newname: oldname}`
"""
p1copies = {}
p2copies = {}
p1 = ctx.p1()
p2 = ctx.p2()
narrowmatch = ctx._repo.narrowmatch()
for dst in ctx.files():
if not narrowmatch(dst) or dst not in ctx:
continue
copied = ctx[dst].renamed()
if not copied:
continue
src, srcnode = copied
if src in p1 and p1[src].filenode() == srcnode:
p1copies[dst] = src
elif src in p2 and p2[src].filenode() == srcnode:
p2copies[dst] = src
return p1copies, p2copies
def encodecopies(files, copies):
items = []
for i, dst in enumerate(files):
if dst in copies:
items.append(b'%d\0%s' % (i, copies[dst]))
if len(items) != len(copies):
raise error.ProgrammingError(
b'some copy targets missing from file list'
)
return b"\n".join(items)
def decodecopies(files, data):
try:
copies = {}
if not data:
return copies
for l in data.split(b'\n'):
strindex, src = l.split(b'\0')
i = int(strindex)
dst = files[i]
copies[dst] = src
return copies
except (ValueError, IndexError):
# Perhaps someone had chosen the same key name (e.g. "p1copies") and
# used different syntax for the value.
return None
def encodefileindices(files, subset):
subset = set(subset)
indices = []
for i, f in enumerate(files):
if f in subset:
indices.append(b'%d' % i)
return b'\n'.join(indices)
def decodefileindices(files, data):
try:
subset = []
if not data:
return subset
for strindex in data.split(b'\n'):
i = int(strindex)
if i < 0 or i >= len(files):
return None
subset.append(files[i])
return subset
except (ValueError, IndexError):
# Perhaps someone had chosen the same key name (e.g. "added") and
# used different syntax for the value.
return None
def _getsidedata(srcrepo, rev):
ctx = srcrepo[rev]
filescopies = computechangesetcopies(ctx)
filesadded = computechangesetfilesadded(ctx)
filesremoved = computechangesetfilesremoved(ctx)
sidedata = {}
if any([filescopies, filesadded, filesremoved]):
sortedfiles = sorted(ctx.files())
p1copies, p2copies = filescopies
p1copies = encodecopies(sortedfiles, p1copies)
p2copies = encodecopies(sortedfiles, p2copies)
filesadded = encodefileindices(sortedfiles, filesadded)
filesremoved = encodefileindices(sortedfiles, filesremoved)
if p1copies:
sidedata[sidedatamod.SD_P1COPIES] = p1copies
if p2copies:
sidedata[sidedatamod.SD_P2COPIES] = p2copies
if filesadded:
sidedata[sidedatamod.SD_FILESADDED] = filesadded
if filesremoved:
sidedata[sidedatamod.SD_FILESREMOVED] = filesremoved
return sidedata
def getsidedataadder(srcrepo, destrepo):
use_w = srcrepo.ui.configbool(b'experimental', b'worker.repository-upgrade')
if pycompat.iswindows or not use_w:
return _get_simple_sidedata_adder(srcrepo, destrepo)
else:
return _get_worker_sidedata_adder(srcrepo, destrepo)
def _sidedata_worker(srcrepo, revs_queue, sidedata_queue, tokens):
"""The function used by worker precomputing sidedata
It read an input queue containing revision numbers
It write in an output queue containing (rev, <sidedata-map>)
The `None` input value is used as a stop signal.
The `tokens` semaphore is user to avoid having too many unprocessed
entries. The workers needs to acquire one token before fetching a task.
They will be released by the consumer of the produced data.
"""
tokens.acquire()
rev = revs_queue.get()
while rev is not None:
data = _getsidedata(srcrepo, rev)
sidedata_queue.put((rev, data))
tokens.acquire()
rev = revs_queue.get()
# processing of `None` is completed, release the token.
tokens.release()
BUFF_PER_WORKER = 50
def _get_worker_sidedata_adder(srcrepo, destrepo):
"""The parallel version of the sidedata computation
This code spawn a pool of worker that precompute a buffer of sidedata
before we actually need them"""
# avoid circular import copies -> scmutil -> worker -> copies
from . import worker
nbworkers = worker._numworkers(srcrepo.ui)
tokens = multiprocessing.BoundedSemaphore(nbworkers * BUFF_PER_WORKER)
revsq = multiprocessing.Queue()
sidedataq = multiprocessing.Queue()
assert srcrepo.filtername is None
# queue all tasks beforehand, revision numbers are small and it make
# synchronisation simpler
#
# Since the computation for each node can be quite expensive, the overhead
# of using a single queue is not revelant. In practice, most computation
# are fast but some are very expensive and dominate all the other smaller
# cost.
for r in srcrepo.changelog.revs():
revsq.put(r)
# queue the "no more tasks" markers
for i in range(nbworkers):
revsq.put(None)
allworkers = []
for i in range(nbworkers):
args = (srcrepo, revsq, sidedataq, tokens)
w = multiprocessing.Process(target=_sidedata_worker, args=args)
allworkers.append(w)
w.start()
# dictionnary to store results for revision higher than we one we are
# looking for. For example, if we need the sidedatamap for 42, and 43 is
# received, when shelve 43 for later use.
staging = {}
def sidedata_companion(revlog, rev):
sidedata = {}
if util.safehasattr(revlog, b'filteredrevs'): # this is a changelog
# Is the data previously shelved ?
sidedata = staging.pop(rev, None)
if sidedata is None:
# look at the queued result until we find the one we are lookig
# for (shelve the other ones)
r, sidedata = sidedataq.get()
while r != rev:
staging[r] = sidedata
r, sidedata = sidedataq.get()
tokens.release()
return False, (), sidedata
return sidedata_companion
def _get_simple_sidedata_adder(srcrepo, destrepo):
"""The simple version of the sidedata computation
It just compute it in the same thread on request"""
def sidedatacompanion(revlog, rev):
sidedata = {}
if util.safehasattr(revlog, 'filteredrevs'): # this is a changelog
sidedata = _getsidedata(srcrepo, rev)
return False, (), sidedata
return sidedatacompanion
def getsidedataremover(srcrepo, destrepo):
def sidedatacompanion(revlog, rev):
f = ()
if util.safehasattr(revlog, 'filteredrevs'): # this is a changelog
if revlog.flags(rev) & sidedataflag.REVIDX_SIDEDATA:
f = (
sidedatamod.SD_P1COPIES,
sidedatamod.SD_P2COPIES,
sidedatamod.SD_FILESADDED,
sidedatamod.SD_FILESREMOVED,
)
return False, f, {}
return sidedatacompanion

mercurial/upgrade.py

# upgrade.py - functions for in place upgrade of Mercurial repository # upgrade.py - functions for in place upgrade of Mercurial repository
# #
# Copyright (c) 2016-present, Gregory Szorc # Copyright (c) 2016-present, Gregory Szorc
# #
# This software may be used and distributed according to the terms of the # This software may be used and distributed according to the terms of the
# GNU General Public License version 2 or any later version. # GNU General Public License version 2 or any later version.
from __future__ import absolute_import from __future__ import absolute_import
import stat import stat
from .i18n import _ from .i18n import _
from .pycompat import getattr from .pycompat import getattr
from . import ( from . import (
changelog, changelog,
copies,
error, error,
filelog, filelog,
hg, hg,
localrepo, localrepo,
manifest, manifest,
metadata,
pycompat, pycompat,
revlog, revlog,
scmutil, scmutil,
util, util,
vfs as vfsmod, vfs as vfsmod,
) )
from .utils import compression from .utils import compression
def sidedatacompanion(rl, rev): def sidedatacompanion(rl, rev):
rl = getattr(rl, '_revlog', rl) rl = getattr(rl, '_revlog', rl)
if rl.flags(rev) & revlog.REVIDX_SIDEDATA: if rl.flags(rev) & revlog.REVIDX_SIDEDATA:
return True, (), {} return True, (), {}
return False, (), {} return False, (), {}
elif localrepo.COPIESSDC_REQUIREMENT in addedreqs: elif localrepo.COPIESSDC_REQUIREMENT in addedreqs:
sidedatacompanion = copies.getsidedataadder(srcrepo, dstrepo) sidedatacompanion = metadata.getsidedataadder(srcrepo, dstrepo)
elif localrepo.COPIESSDC_REQUIREMENT in removedreqs: elif localrepo.COPIESSDC_REQUIREMENT in removedreqs:
sidedatacompanion = copies.getsidedataremover(srcrepo, dstrepo) sidedatacompanion = metadata.getsidedataremover(srcrepo, dstrepo)
return sidedatacompanion return sidedatacompanion
def matchrevlog(revlogfilter, entry): def matchrevlog(revlogfilter, entry):
"""check is a revlog is selected for cloning """check is a revlog is selected for cloning
The store entry is checked against the passed filter""" The store entry is checked against the passed filter"""
if entry.endswith(b'00changelog.i'): if entry.endswith(b'00changelog.i'):