This is an archive of the discontinued Mercurial Phabricator instance.

Differential D4724

wireprotov2server: port to emitrevisions()
ClosedPublic

Authored by indygreg on Sep 24 2018, 1:15 PM.

Details

Reviewers
None
Group Reviewers
hg-reviewers
Commits
rHG7b752bf08435: wireprotov2server: port to emitrevisions()
Summary

We now have a proper storage API to request data on multiple
revisions. We can drop it into wire protocol version 2 with
minimal effort.

The new API handles pretty much everything we were doing manually to
build up the delta request. So we were able to delete a lot of code.

As a bonus, wireprotov2 code is no longer accessing some low-level
storage APIs. This includes the assumption that a node has an
associated numeric revision number! This should make it drastically
simpler to implement a server that doesn't have the concept of
revision numbers.

Diff Detail

Repository
rHG Mercurial
Lint
Automatic diff as part of commit; lint not applicable.
Unit
Automatic diff as part of commit; unit tests not applicable.

Event Timeline

indygreg created this revision.Sep 24 2018, 1:15 PM
Closed by commit rHG7b752bf08435: wireprotov2server: port to emitrevisions() (authored by indygreg). · Explain WhySep 26 2018, 1:26 PM
This revision was automatically updated to reflect the committed changes.

Revision Contents
Changeset List

PathPackages
Mmercurial/wireprotov2server.py (177 lines)

Diff 11395

mercurial/wireprotov2server.py

# Copyright 21 May 2005 - (c) 2005 Jake Edge <jake@edge2.net> # Copyright 21 May 2005 - (c) 2005 Jake Edge <jake@edge2.net>
# Copyright 2005-2007 Matt Mackall <mpm@selenic.com> # Copyright 2005-2007 Matt Mackall <mpm@selenic.com>
# #
# This software may be used and distributed according to the terms of the # This software may be used and distributed according to the terms of the
# GNU General Public License version 2 or any later version. # GNU General Public License version 2 or any later version.
from __future__ import absolute_import from __future__ import absolute_import
import contextlib import contextlib
from .i18n import _ from .i18n import _
from .node import ( from .node import (
hex, hex,
nullid, nullid,
nullrev,
) )
from . import ( from . import (
changegroup,
dagop,
discovery, discovery,
encoding, encoding,
error, error,
narrowspec, narrowspec,
pycompat, pycompat,
streamclone, streamclone,
util, util,
wireprotoframing, wireprotoframing,
} }
if streamclone.allowservergeneration(repo): if streamclone.allowservergeneration(repo):
caps['rawrepoformats'] = sorted(repo.requirements & caps['rawrepoformats'] = sorted(repo.requirements &
repo.supportedformats) repo.supportedformats)
return proto.addcapabilities(repo, caps) return proto.addcapabilities(repo, caps)
def builddeltarequests(store, nodes, haveparents):
"""Build a series of revision delta requests against a backend store.
Returns a list of revision numbers in the order they should be sent
and a list of ``irevisiondeltarequest`` instances to be made against
the backend store.
"""
# We sort and send nodes in DAG order because this is optimal for
# storage emission.
# TODO we may want a better storage API here - one where we can throw
# a list of nodes and delta preconditions over a figurative wall and
# have the storage backend figure it out for us.
revs = dagop.linearize({store.rev(n) for n in nodes}, store.parentrevs)
requests = []
seenrevs = set()
for rev in revs:
node = store.node(rev)
parentnodes = store.parents(node)
parentrevs = [store.rev(n) for n in parentnodes]
deltabaserev = store.deltaparent(rev)
deltabasenode = store.node(deltabaserev)
# The choice of whether to send a fulltext revision or a delta and
# what delta to send is governed by a few factors.
#
# To send a delta, we need to ensure the receiver is capable of
# decoding it. And that requires the receiver to have the base
# revision the delta is against.
#
# We can only guarantee the receiver has the base revision if
# a) we've already sent the revision as part of this group
# b) the receiver has indicated they already have the revision.
# And the mechanism for "b" is the client indicating they have
# parent revisions. So this means we can only send the delta if
# it is sent before or it is against a delta and the receiver says
# they have a parent.
# We can send storage delta if it is against a revision we've sent
# in this group.
if deltabaserev != nullrev and deltabaserev in seenrevs:
basenode = deltabasenode
# We can send storage delta if it is against a parent revision and
# the receiver indicates they have the parents.
elif (deltabaserev != nullrev and deltabaserev in parentrevs
and haveparents):
basenode = deltabasenode
# Otherwise the storage delta isn't appropriate. Fall back to
# using another delta, if possible.
# Use p1 if we've emitted it or receiver says they have it.
elif parentrevs[0] != nullrev and (
parentrevs[0] in seenrevs or haveparents):
basenode = parentnodes[0]
# Use p2 if we've emitted it or receiver says they have it.
elif parentrevs[1] != nullrev and (
parentrevs[1] in seenrevs or haveparents):
basenode = parentnodes[1]
# Nothing appropriate to delta against. Send the full revision.
else:
basenode = nullid
requests.append(changegroup.revisiondeltarequest(
node=node,
p1node=parentnodes[0],
p2node=parentnodes[1],
# Receiver deals with linknode resolution.
linknode=nullid,
basenode=basenode,
))
seenrevs.add(rev)
return revs, requests
def wireprotocommand(name, args=None, permission='push'): def wireprotocommand(name, args=None, permission='push'):
"""Decorator to declare a wire protocol command. """Decorator to declare a wire protocol command.
``name`` is the name of the wire protocol command being provided. ``name`` is the name of the wire protocol command being provided.
``args`` is a dict defining arguments accepted by the command. Keys are ``args`` is a dict defining arguments accepted by the command. Keys are
the argument name. Values are dicts with the following keys: the argument name. Values are dicts with the following keys:
# Validate requested nodes. # Validate requested nodes.
for node in nodes: for node in nodes:
try: try:
store.rev(node) store.rev(node)
except error.LookupError: except error.LookupError:
raise error.WireprotoCommandError('unknown file node: %s', raise error.WireprotoCommandError('unknown file node: %s',
(hex(node),)) (hex(node),))
revs, requests = builddeltarequests(store, nodes, haveparents) revisions = store.emitrevisions(nodes,
revisiondata=b'revision' in fields,
assumehaveparentrevisions=haveparents)
yield { yield {
b'totalitems': len(revs), b'totalitems': len(nodes),
} }
if b'revision' in fields: for revision in revisions:
deltas = store.emitrevisiondeltas(requests)
else:
deltas = None
for rev in revs:
node = store.node(rev)
if deltas is not None:
delta = next(deltas)
else:
delta = None
d = { d = {
b'node': node, b'node': revision.node,
} }
if b'parents' in fields: if b'parents' in fields:
d[b'parents'] = store.parents(node) d[b'parents'] = [revision.p1node, revision.p2node]
followingmeta = [] followingmeta = []
followingdata = [] followingdata = []
if b'revision' in fields: if b'revision' in fields:
assert delta is not None if revision.revision is not None:
assert delta.flags == 0 followingmeta.append((b'revision', len(revision.revision)))
assert d[b'node'] == delta.node followingdata.append(revision.revision)
else:
if delta.revision is not None: d[b'deltabasenode'] = revision.basenode
followingmeta.append((b'revision', len(delta.revision))) followingmeta.append((b'delta', len(revision.delta)))
followingdata.append(delta.revision) followingdata.append(revision.delta)
else:
d[b'deltabasenode'] = delta.basenode
followingmeta.append((b'delta', len(delta.delta)))
followingdata.append(delta.delta)
if followingmeta: if followingmeta:
d[b'fieldsfollowing'] = followingmeta d[b'fieldsfollowing'] = followingmeta
yield d yield d
for extra in followingdata: for extra in followingdata:
yield extra yield extra
if deltas is not None:
try:
next(deltas)
raise error.ProgrammingError('should not have more deltas')
except GeneratorExit:
pass
@wireprotocommand( @wireprotocommand(
'heads', 'heads',
args={ args={
'publiconly': { 'publiconly': {
'type': 'bool', 'type': 'bool',
'default': lambda: False, 'default': lambda: False,
'example': False, 'example': False,
}, },
# Validate the node is known and abort on unknown revisions. # Validate the node is known and abort on unknown revisions.
for node in nodes: for node in nodes:
try: try:
store.rev(node) store.rev(node)
except error.LookupError: except error.LookupError:
raise error.WireprotoCommandError( raise error.WireprotoCommandError(
'unknown node: %s', (node,)) 'unknown node: %s', (node,))
revs, requests = builddeltarequests(store, nodes, haveparents) revisions = store.emitrevisions(nodes,
revisiondata=b'revision' in fields,
assumehaveparentrevisions=haveparents)
yield { yield {
b'totalitems': len(revs), b'totalitems': len(nodes),
} }
if b'revision' in fields: for revision in revisions:
deltas = store.emitrevisiondeltas(requests)
else:
deltas = None
for rev in revs:
node = store.node(rev)
if deltas is not None:
delta = next(deltas)
else:
delta = None
d = { d = {
b'node': node, b'node': revision.node,
} }
if b'parents' in fields: if b'parents' in fields:
d[b'parents'] = store.parents(node) d[b'parents'] = [revision.p1node, revision.p2node]
followingmeta = [] followingmeta = []
followingdata = [] followingdata = []
if b'revision' in fields: if b'revision' in fields:
assert delta is not None if revision.revision is not None:
assert delta.flags == 0 followingmeta.append((b'revision', len(revision.revision)))
assert d[b'node'] == delta.node followingdata.append(revision.revision)
else:
if delta.revision is not None: d[b'deltabasenode'] = revision.basenode
followingmeta.append((b'revision', len(delta.revision))) followingmeta.append((b'delta', len(revision.delta)))
followingdata.append(delta.revision) followingdata.append(revision.delta)
else:
d[b'deltabasenode'] = delta.basenode
followingmeta.append((b'delta', len(delta.delta)))
followingdata.append(delta.delta)
if followingmeta: if followingmeta:
d[b'fieldsfollowing'] = followingmeta d[b'fieldsfollowing'] = followingmeta
yield d yield d
for extra in followingdata: for extra in followingdata:
yield extra yield extra
if deltas is not None:
try:
next(deltas)
raise error.ProgrammingError('should not have more deltas')
except GeneratorExit:
pass
@wireprotocommand( @wireprotocommand(
'pushkey', 'pushkey',
args={ args={
'namespace': { 'namespace': {
'type': 'bytes', 'type': 'bytes',
'required': True, 'required': True,
'example': b'ns', 'example': b'ns',
}, },