This is an archive of the discontinued Mercurial Phabricator instance.

Differential D1688

datapack: add getdelta api
ClosedPublic

Authored by durham on Dec 14 2017, 2:24 PM.
Tags
None
Subscribers
quark

Details

Reviewers
quark
Group Reviewers
Restricted Project
Commits
rFBHGX38db46cc245c: datapack: add getdelta api
Summary

In order to improve repack performance, we need to be able to fetch a single
delta from a data store. This diff adds the new api and implements it in all our
existing interfaces.

Diff Detail

Repository
rFBHGX Facebook Mercurial Extensions
Lint
Lint Skipped
Unit
Unit Tests Skipped

Event Timeline

durham created this revision.Dec 14 2017, 2:24 PM
Herald added a reviewer: Restricted Project. · View Herald TranscriptDec 14 2017, 2:24 PM
quark requested changes to this revision.Dec 14 2017, 4:46 PM
quark added a subscriber: quark.
quark added inline comments.
cstore/py-cdatapack.h
476

nit: &args[0] is the same as args.

args is not a C array but a Python tuple. Use PyTuple_GET_ITEM(x, 0) instead of [0] here:

PyErr_SetObject(PyExc_KeyError, PyTuple_GET_ITEM(args, 0));

That said, KeyError accepts a tuple so the old code won't crash.

503–506

s/PyString_/PyBytes_/ to decouple from Python 2.

I think it's worthwhile to use:

delta = PyBuffer_FromMemory(link.delta, (Py_ssize_t) link.delta_sz)

(and move`free() below to be inside if (!delta)`).

519–533

PyXDECREF(NULL) is also a no-op. So the block is equivalent to:

free((void *)link.delta);
return tuple;
530

nit: free(NULL) is a no-op. So if is unnecessary.

This revision now requires changes to proceed.Dec 14 2017, 4:46 PM
quark added inline comments.Dec 14 2017, 5:01 PM
cstore/py-cdatapack.h
503–506

I take that back. PyBuffer_FromMemory won't take care of deallocation. So we have to do a copy here.

durham updated this revision to Diff 4439.Dec 14 2017, 5:46 PM
durham marked 5 inline comments as done.Dec 14 2017, 5:47 PM
durham added inline comments.
cstore/py-cdatapack.h
476

This is just a copy of the code in cdatapack_getmeta. I'll update it though.

quark accepted this revision.Dec 14 2017, 10:19 PM
This revision is now accepted and ready to land.Dec 14 2017, 10:19 PM
durham marked an inline comment as done.Dec 15 2017, 2:34 PM
Closed by commit rFBHGX38db46cc245c: datapack: add getdelta api (authored by durham). · Explain Why
This revision was automatically updated to reflect the committed changes.

Revision Contents
Changeset List

Diff 4435

cstore/py-cdatapack.h

err_cleanup: err_cleanup:
PyErr_Format(PyExc_ValueError, "corrupted datapack metadata"); PyErr_Format(PyExc_ValueError, "corrupted datapack metadata");
Py_XDECREF(pymeta); Py_XDECREF(pymeta);
return NULL; return NULL;
} }
/** /**
* Finds a node and returns its delta entry (delta, deltabasenode,
* meta) tuple if found.
*/
static PyObject *cdatapack_getdelta(
py_cdatapack *self,
PyObject *args) {
const char *node;
Py_ssize_t node_sz;
// 1. Parse the args
if (!PyArg_ParseTuple(args, "s#", &node, &node_sz)) {
return NULL;
}
if (node_sz != NODE_SZ) {
PyErr_Format(PyExc_ValueError, "node must be %d bytes long", NODE_SZ);
return NULL;
}
// 2. Read the delta chain
pack_index_entry_t index_entry;
if (find(self->handle, (const uint8_t *) node, &index_entry) == false) {
PyErr_SetObject(PyExc_KeyError, &args[0]);
quarkUnsubmitted
Done

nit: &args[0] is the same as args.

args is not a C array but a Python tuple. Use PyTuple_GET_ITEM(x, 0) instead of [0] here:

PyErr_SetObject(PyExc_KeyError, PyTuple_GET_ITEM(args, 0));

That said, KeyError accepts a tuple so the old code won't crash.

quark: nit: `&args[0]` is the same as `args`. `args` is not a C array but a Python tuple. Use…
durhamAuthorUnsubmitted
Not Done

This is just a copy of the code in cdatapack_getmeta. I'll update it though.

durham: This is just a copy of the code in cdatapack_getmeta. I'll update it though.
return NULL;
}
delta_chain_link_t link;
get_delta_chain_link_result_t next = getdeltachainlink(
self->handle,
((uint8_t *) self->handle->data_mmap) + index_entry.data_offset,
&link);
if (next.code != GET_DELTA_CHAIN_LINK_OK) {
PyErr_SetObject(PyExc_KeyError, &args[0]);
return NULL;
}
// Populate the link.delta pointer
if (!uncompressdeltachainlink(&link)) {
PyErr_Format(PyExc_ValueError, "unable to decompress pack entry");
return NULL;
}
// 3. Convert it into python objects
PyObject *tuple = NULL;
PyObject *delta = NULL, *deltabasenode = NULL, *meta =
NULL;
delta = PyString_FromStringAndSize(
(const char *) link.delta, (Py_ssize_t) link.delta_sz);
deltabasenode = PyString_FromStringAndSize(
(const char *) link.deltabase_node, NODE_SZ);
quarkUnsubmitted
Done

s/PyString_/PyBytes_/ to decouple from Python 2.

I think it's worthwhile to use:

delta = PyBuffer_FromMemory(link.delta, (Py_ssize_t) link.delta_sz)

(and move`free() below to be inside if (!delta)`).

quark: `s/PyString_/PyBytes_/` to decouple from Python 2. I think it's worthwhile to use: delta =…
quarkUnsubmitted
Done

I take that back. PyBuffer_FromMemory won't take care of deallocation. So we have to do a copy here.

quark: I take that back. `PyBuffer_FromMemory` won't take care of deallocation. So we have to do a…
meta = readpymeta(&link);
if (deltabasenode != NULL &&
delta != NULL &&
meta != NULL) {
tuple = PyTuple_Pack(3, delta, deltabasenode, meta);
}
Py_XDECREF(delta);
Py_XDECREF(deltabasenode);
Py_XDECREF(meta);
if (tuple == NULL) {
goto err_cleanup;
}
goto cleanup;
err_cleanup:
Py_XDECREF(tuple);
tuple = NULL;
cleanup:
if (link.delta) {
quarkUnsubmitted
Done

nit: free(NULL) is a no-op. So if is unnecessary.

quark: nit: `free(NULL)` is a no-op. So `if` is unnecessary.
free((void *)link.delta);
}
return tuple;
quarkUnsubmitted
Done

PyXDECREF(NULL) is also a no-op. So the block is equivalent to:

free((void *)link.delta);
return tuple;
quark: `PyXDECREF(NULL)` is also a no-op. So the block is equivalent to: free((void *)link.delta)…
}
/**
* Finds a node and returns a list of (filename, node, filename, delta base * Finds a node and returns a list of (filename, node, filename, delta base
* node, delta) tuples if found. * node, delta) tuples if found.
*/ */
static PyObject *cdatapack_getdeltachain( static PyObject *cdatapack_getdeltachain(
py_cdatapack *self, py_cdatapack *self,
PyObject *args) { PyObject *args) {
const char *node; const char *node;
Py_ssize_t node_sz; Py_ssize_t node_sz;
{"iterentries", (PyCFunction)cdatapack_getiterentries, {"iterentries", (PyCFunction)cdatapack_getiterentries,
METH_NOARGS, METH_NOARGS,
"Iterate over (path, nodeid, deltabasenode, delta) tuples in this " "Iterate over (path, nodeid, deltabasenode, delta) tuples in this "
"datapack."}, "datapack."},
{"_find", (PyCFunction)cdatapack_find, {"_find", (PyCFunction)cdatapack_find,
METH_VARARGS, METH_VARARGS,
"Finds a node and returns a (node, deltabase index offset, " "Finds a node and returns a (node, deltabase index offset, "
"data offset, data size) tuple if found."}, "data offset, data size) tuple if found."},
{"getdelta", (PyCFunction)cdatapack_getdelta,
METH_VARARGS,
"Finds a node and returns its delta entry (delta, deltabasename, "
"deltabasenode, meta) tuple if found."},
{"getdeltachain", (PyCFunction)cdatapack_getdeltachain, {"getdeltachain", (PyCFunction)cdatapack_getdeltachain,
METH_VARARGS, METH_VARARGS,
"Finds a node and returns a list of (filename, node, filename, delta " "Finds a node and returns a list of (filename, node, filename, delta "
"base node, delta) tuples if found."}, "base node, delta) tuples if found."},
{"getmeta", (PyCFunction)cdatapack_getmeta, METH_VARARGS, {"getmeta", (PyCFunction)cdatapack_getmeta, METH_VARARGS,
"Return a metadata dictionary for given node"}, "Return a metadata dictionary for given node"},
{NULL, NULL} {NULL, NULL}
}; };

remotefilelog/contentstore.py

text = fulltext text = fulltext
while chain: while chain:
delta = chain.pop()[ChainIndicies.DATA] delta = chain.pop()[ChainIndicies.DATA]
text = mdiff.patches(text, [delta]) text = mdiff.patches(text, [delta])
return text return text
def getdelta(self, name, node):
"""Return the single delta entry for the given name/node pair.
"""
for store in self.stores:
try:
return store.getdelta(name, node)
except KeyError:
pass
raise KeyError((name, hex(node)))
def getdeltachain(self, name, node): def getdeltachain(self, name, node):
"""Returns the deltachain for the given name/node pair. """Returns the deltachain for the given name/node pair.
Returns an ordered list of: Returns an ordered list of:
[(name, node, deltabasename, deltabasenode, deltacontent),...] [(name, node, deltabasename, deltabasenode, deltacontent),...]
where the chain is terminated by a full text entry with a nullid where the chain is terminated by a full text entry with a nullid
# lfs tracks renames in its own metadata, remove hg copy metadata, # lfs tracks renames in its own metadata, remove hg copy metadata,
# because copy metadata will be re-added by lfs flag processor. # because copy metadata will be re-added by lfs flag processor.
if flags & revlog.REVIDX_EXTSTORED: if flags & revlog.REVIDX_EXTSTORED:
copyrev = copyfrom = None copyrev = copyfrom = None
revision = shallowutil.createrevlogtext(content, copyfrom, copyrev) revision = shallowutil.createrevlogtext(content, copyfrom, copyrev)
return revision return revision
def getdelta(self, name, node):
# Since remotefilelog content stores only contain full texts, just
# return that.
revision = self.get(name, node)
return revision, name, nullid, self.getmeta(name, node)
def getdeltachain(self, name, node): def getdeltachain(self, name, node):
# Since remotefilelog content stores just contain full texts, we return # Since remotefilelog content stores just contain full texts, we return
# a fake delta chain that just consists of a single full text revision. # a fake delta chain that just consists of a single full text revision.
# The nullid in the deltabasenode slot indicates that the revision is a # The nullid in the deltabasenode slot indicates that the revision is a
# fulltext. # fulltext.
revision = self.get(name, node) revision = self.get(name, node)
return [(name, node, None, nullid, revision)] return [(name, node, None, nullid, revision)]
# type(shared) is usually remotefilelogcontentstore # type(shared) is usually remotefilelogcontentstore
self._shared = shared self._shared = shared
def get(self, name, node): def get(self, name, node):
self._fileservice.prefetch([(name, hex(node))], force=True, self._fileservice.prefetch([(name, hex(node))], force=True,
fetchdata=True) fetchdata=True)
return self._shared.get(name, node) return self._shared.get(name, node)
def getdelta(self, name, node):
revision = self.get(name, node)
return revision, name, nullid, self._shared.getmeta(name, node)
def getdeltachain(self, name, node): def getdeltachain(self, name, node):
# Since our remote content stores just contain full texts, we return a # Since our remote content stores just contain full texts, we return a
# fake delta chain that just consists of a single full text revision. # fake delta chain that just consists of a single full text revision.
# The nullid in the deltabasenode slot indicates that the revision is a # The nullid in the deltabasenode slot indicates that the revision is a
# fulltext. # fulltext.
revision = self.get(name, node) revision = self.get(name, node)
return [(name, node, None, nullid, revision)] return [(name, node, None, nullid, revision)]
self._svfs = repo.svfs self._svfs = repo.svfs
self._revlogs = dict() self._revlogs = dict()
self._cl = revlog.revlog(self._svfs, '00changelog.i') self._cl = revlog.revlog(self._svfs, '00changelog.i')
self._repackstartlinkrev = 0 self._repackstartlinkrev = 0
def get(self, name, node): def get(self, name, node):
return self._revlog(name).revision(node, raw=True) return self._revlog(name).revision(node, raw=True)
def getdelta(self, name, node):
revision = self.get(name, node)
return revision, name, nullid, self.getmeta(name, node)
def getdeltachain(self, name, node): def getdeltachain(self, name, node):
revision = self.get(name, node) revision = self.get(name, node)
return [(name, node, None, nullid, revision)] return [(name, node, None, nullid, revision)]
def getmeta(self, name, node): def getmeta(self, name, node):
rl = self._revlog(name) rl = self._revlog(name)
rev = rl.rev(node) rev = rl.rev(node)
return {constants.METAKEYFLAG: rl.flags(rev), return {constants.METAKEYFLAG: rl.flags(rev),

remotefilelog/datapack.py

for pack in self.refresh(): for pack in self.refresh():
try: try:
return pack.getmeta(name, node) return pack.getmeta(name, node)
except KeyError: except KeyError:
pass pass
raise KeyError((name, hex(node))) raise KeyError((name, hex(node)))
def getdelta(self, name, node):
for pack in self.packs:
try:
return pack.getdelta(name, node)
except KeyError:
pass
for pack in self.refresh():
try:
return pack.getdelta(name, node)
except KeyError:
pass
raise KeyError((name, hex(node)))
def getdeltachain(self, name, node): def getdeltachain(self, name, node):
for pack in self.packs: for pack in self.packs:
try: try:
return pack.getdeltachain(name, node) return pack.getdeltachain(name, node)
except KeyError: except KeyError:
pass pass
for pack in self.refresh(): for pack in self.refresh():
metalen = struct.unpack_from('!I', rawentry, offset)[0] metalen = struct.unpack_from('!I', rawentry, offset)[0]
offset += 4 offset += 4
meta = shallowutil.parsepackmeta(rawentry[offset:offset + metalen]) meta = shallowutil.parsepackmeta(rawentry[offset:offset + metalen])
return meta return meta
def getdelta(self, name, node):
value = self._find(node)
if value is None:
raise KeyError((name, hex(node)))
node, deltabaseoffset, offset, size = value
entry = self._readentry(offset, size, getmeta=True)
filename, node, deltabasenode, delta, meta = entry
# If we've read a lot of data from the mmap, free some memory.
self.freememory()
return delta, filename, deltabasenode, meta
def getdeltachain(self, name, node): def getdeltachain(self, name, node):
value = self._find(node) value = self._find(node)
if value is None: if value is None:
raise KeyError((name, hex(node))) raise KeyError((name, hex(node)))
params = self.params params = self.params
# Precompute chains # Precompute chains
def get(self, name, node): def get(self, name, node):
raise RuntimeError("must use getdeltachain with datapack (%s:%s)" raise RuntimeError("must use getdeltachain with datapack (%s:%s)"
% (name, hex(node))) % (name, hex(node)))
def getmeta(self, name, node): def getmeta(self, name, node):
return self.datapack.getmeta(node) return self.datapack.getmeta(node)
def getdelta(self, name, node):
result = self.datapack.getdelta(node)
if result is None:
raise KeyError((name, hex(node)))
delta, deltabasenode, meta = result
return delta, name, deltabasenode, meta
def getdeltachain(self, name, node): def getdeltachain(self, name, node):
result = self.datapack.getdeltachain(node) result = self.datapack.getdeltachain(node)
if result is None: if result is None:
raise KeyError((name, hex(node))) raise KeyError((name, hex(node)))
return result return result
def add(self, name, node, data): def add(self, name, node, data):

tests/test-remotefilelog-datapack.py

"""Test putting multiple unrelated blobs into a pack and reading them """Test putting multiple unrelated blobs into a pack and reading them
out. out.
""" """
revisions = [] revisions = []
for i in range(10): for i in range(10):
filename = "foo%s" % i filename = "foo%s" % i
content = "abcdef%s" % i content = "abcdef%s" % i
node = self.getHash(content) node = self.getHash(content)
revisions.append((filename, node, nullid, content)) revisions.append((filename, node, self.getFakeHash(), content))
pack = self.createPack(revisions) pack = self.createPack(revisions)
for filename, node, base, content in revisions: for filename, node, base, content in revisions:
entry = pack.getdelta(filename, node)
self.assertEquals((content, filename, base, {}), entry)
chain = pack.getdeltachain(filename, node) chain = pack.getdeltachain(filename, node)
self.assertEquals(content, chain[0][4]) self.assertEquals(content, chain[0][4])
def testAddDeltas(self): def testAddDeltas(self):
"""Test putting multiple delta blobs into a pack and read the chain. """Test putting multiple delta blobs into a pack and read the chain.
""" """
revisions = [] revisions = []
filename = "foo" filename = "foo"
lastnode = nullid lastnode = nullid
for i in range(10): for i in range(10):
content = "abcdef%s" % i content = "abcdef%s" % i
node = self.getHash(content) node = self.getHash(content)
revisions.append((filename, node, lastnode, content)) revisions.append((filename, node, lastnode, content))
lastnode = node lastnode = node
pack = self.createPack(revisions) pack = self.createPack(revisions)
entry = pack.getdelta(filename, revisions[0][1])
realvalue = (revisions[0][3], filename, revisions[0][2], {})
self.assertEquals(entry, realvalue)
# Test that the chain for the final entry has all the others # Test that the chain for the final entry has all the others
chain = pack.getdeltachain(filename, node) chain = pack.getdeltachain(filename, node)
for i in range(10): for i in range(10):
content = "abcdef%s" % i content = "abcdef%s" % i
self.assertEquals(content, chain[-i - 1][4]) self.assertEquals(content, chain[-i - 1][4])
def testPackMany(self): def testPackMany(self):
"""Pack many related and unrelated objects. """Pack many related and unrelated objects.