This patch fixes 2 things:
And restores the test to use as many pack files as it was before c66406 to
exercise the extreme case.
2 was fixed by replacing mmap.mmap to a thin mmap implementation that does
not keep a fd internally. The new thin mmap implementation only supports
mmapobj[a:b], so unpack_from was changed to unpack.
ulimit -n 50 and run ./scripts/unit.py
| Lint Skipped |
| Unit Tests Skipped |
Does this actually fix the test? I say that because the limit is somehow 256 and when I check the number of open files for this test, it is >400 irrespective of whether we use the datapack or the fastdatapack.
In D1430#23627, @singhsrb wrote:The test doesn't pass for me after ulimit -n 256
Should be fixed now - the old code was not using the cdatapack code path.
It even worked with ulimit -n 128 now. This should fix the test. Thanks for taking care of this!
You will probably need to change the title and description of this commit somewhat to reflect what the code is actually doing now.
Let's just fix the test to not create 100+ pack files. No point in reducing our test coverage for something easily fixable.
I'm not sure we should use a completely custom mmap implementation. I don't know if the python mmap implementation is handling subtle edge cases or anything for us. I'd rather not move off the time tested implementation just for this edge case that only appears in a test or horribly broken repositories. If you do want to land this, let's put it behind a config flag and turn it on for our team for a bit first.
| Path | Packages | |||
|---|---|---|---|---|
| A | M | hgext3rd/extmmap.pyx (70 lines) | ||
| M | remotefilelog/basepack.py (19 lines) | |||
| M | remotefilelog/datapack.py (2 lines) | |||
| M | setup.py (9 lines) | |||
| M | tests/test-remotefilelog-datapack.py (11 lines) |
| # extmmap.pyx - Simple mmap implementation that does not keep a fd open | |||||
| # | |||||
| # Copyright 2017 Facebook, Inc. | |||||
| # | |||||
| # This software may be used and distributed according to the terms of the | |||||
| # GNU General Public License version 2 or any later version. | |||||
| from cpython.bytes cimport PyBytes_FromStringAndSize | |||||
| from libc.errno cimport errno | |||||
| from posix cimport mman, stat | |||||
| import mmap as pymmap | |||||
| import os | |||||
| cdef extern from "sys/mman.h": | |||||
| cdef char *MAP_FAILED | |||||
| cdef _raise_oserror(message=''): | |||||
| if message: | |||||
| message += ': ' | |||||
| message += os.strerror(errno) | |||||
| return OSError(errno, message) | |||||
| cdef class mmap: | |||||
| cdef char *ptr | |||||
| cdef size_t len | |||||
| def __cinit__(self, int fd, size_t length, int access=pymmap.ACCESS_READ): | |||||
| cdef stat.struct_stat st | |||||
| # Only support a subset of options for now | |||||
| if access != pymmap.ACCESS_READ: | |||||
| raise RuntimeError('access %s is unsupported' % access) | |||||
| # If length is 0, read size from file | |||||
| if length == 0: | |||||
| r = stat.fstat(fd, &st) | |||||
| if r != 0: | |||||
| _raise_oserror('fstat failed') | |||||
| length = st.st_size | |||||
| self.ptr = <char*>mman.mmap(NULL, length, mman.PROT_READ, | |||||
| mman.MAP_SHARED, fd, 0) | |||||
| if self.ptr == MAP_FAILED: | |||||
| _raise_oserror('mmap failed') | |||||
| self.len = length | |||||
| # This method should be called "unmap". But it's to be compatible with | |||||
| # the Python API. | |||||
| cpdef close(self): | |||||
| if self.ptr == NULL: | |||||
| return | |||||
| r = mman.munmap(self.ptr, self.len) | |||||
| if r != 0: | |||||
| _raise_oserror('munmap failed') | |||||
| self.ptr = NULL | |||||
| def __getslice__(self, Py_ssize_t i, Py_ssize_t j): | |||||
| if self.ptr == NULL: | |||||
| raise RuntimeError('mmap closed') | |||||
| if j > self.len: | |||||
| j = self.len | |||||
| return PyBytes_FromStringAndSize(self.ptr + i, j - i) | |||||
| def __len__(self): | |||||
| return self.len | |||||
| def __dealloc__(self): | |||||
| self.close() | |||||
| from __future__ import absolute_import | from __future__ import absolute_import | ||||
| import errno, hashlib, mmap, os, struct, time | import errno, hashlib, mmap, os, struct, time | ||||
| from collections import defaultdict | from collections import defaultdict | ||||
| from mercurial import policy, pycompat, util | from mercurial import policy, pycompat, util | ||||
| from mercurial.i18n import _ | from mercurial.i18n import _ | ||||
| from mercurial import vfs as vfsmod | from mercurial import vfs as vfsmod | ||||
| from . import shallowutil | from . import shallowutil | ||||
| mmapfunc = mmap.mmap | |||||
| try: | |||||
| from hgext3rd import extmmap | |||||
| except ImportError: | |||||
| pass | |||||
| else: | |||||
| mmapfunc = extmmap.mmap | |||||
| osutil = policy.importmod(r'osutil') | osutil = policy.importmod(r'osutil') | ||||
| # The pack version supported by this implementation. This will need to be | # The pack version supported by this implementation. This will need to be | ||||
| # rev'd whenever the byte format changes. Ex: changing the fanout prefix, | # rev'd whenever the byte format changes. Ex: changing the fanout prefix, | ||||
| # changing any of the int sizes, changing the delta algorithm, etc. | # changing any of the int sizes, changing the delta algorithm, etc. | ||||
| PACKVERSIONSIZE = 1 | PACKVERSIONSIZE = 1 | ||||
| INDEXVERSIONSIZE = 2 | INDEXVERSIONSIZE = 2 | ||||
| fanouttable.append(fanoutentry) | fanouttable.append(fanoutentry) | ||||
| return fanouttable | return fanouttable | ||||
| @util.propertycache | @util.propertycache | ||||
| def _indexend(self): | def _indexend(self): | ||||
| if self.VERSION == 0: | if self.VERSION == 0: | ||||
| return self.indexsize | return self.indexsize | ||||
| else: | else: | ||||
| nodecount = struct.unpack_from('!Q', self._index, | offset = self.params.indexstart - 8 | ||||
| self.params.indexstart - 8)[0] | nodecount = struct.unpack_from('!Q', | ||||
| self._index[offset:offset + 8])[0] | |||||
| return self.params.indexstart + nodecount * self.INDEXENTRYLENGTH | return self.params.indexstart + nodecount * self.INDEXENTRYLENGTH | ||||
| def freememory(self): | def freememory(self): | ||||
| """Unmap and remap the memory to free it up after known expensive | """Unmap and remap the memory to free it up after known expensive | ||||
| operations. Return True if self._data and self._index were reloaded. | operations. Return True if self._data and self._index were reloaded. | ||||
| """ | """ | ||||
| if self._index: | if self._index: | ||||
| if self._pagedin < self.MAXPAGEDIN: | if self._pagedin < self.MAXPAGEDIN: | ||||
| return False | return False | ||||
| self._index.close() | self._index.close() | ||||
| self._data.close() | self._data.close() | ||||
| # TODO: use an opener/vfs to access these paths | # TODO: use an opener/vfs to access these paths | ||||
| with open(self.indexpath, PACKOPENMODE) as indexfp: | with open(self.indexpath, PACKOPENMODE) as indexfp: | ||||
| # memory-map the file, size 0 means whole file | # memory-map the file, size 0 means whole file | ||||
| self._index = mmap.mmap(indexfp.fileno(), 0, | self._index = mmapfunc(indexfp.fileno(), 0, | ||||
| access=mmap.ACCESS_READ) | access=mmap.ACCESS_READ) | ||||
| with open(self.packpath, PACKOPENMODE) as datafp: | with open(self.packpath, PACKOPENMODE) as datafp: | ||||
| self._data = mmap.mmap(datafp.fileno(), 0, access=mmap.ACCESS_READ) | self._data = mmapfunc(datafp.fileno(), 0, access=mmap.ACCESS_READ) | ||||
| self._pagedin = 0 | self._pagedin = 0 | ||||
| return True | return True | ||||
| def getmissing(self, keys): | def getmissing(self, keys): | ||||
| raise NotImplemented() | raise NotImplemented() | ||||
| def markledger(self, ledger, options=None): | def markledger(self, ledger, options=None): | ||||
| # python-lz4 stores the length of the uncompressed field as a | # python-lz4 stores the length of the uncompressed field as a | ||||
| # little-endian 32-bit integer at the start of the data. | # little-endian 32-bit integer at the start of the data. | ||||
| uncompressedlen = struct.unpack('<I', data[offset:offset + 4])[0] | uncompressedlen = struct.unpack('<I', data[offset:offset + 4])[0] | ||||
| offset += deltalen | offset += deltalen | ||||
| if self.VERSION == 1: | if self.VERSION == 1: | ||||
| # <4 byte len> + <metadata-list> | # <4 byte len> + <metadata-list> | ||||
| metalen = struct.unpack_from('!I', data, offset)[0] | metalen = struct.unpack('!I', data[offset:offset + 4])[0] | ||||
| offset += 4 + metalen | offset += 4 + metalen | ||||
| yield (filename, node, deltabase, uncompressedlen) | yield (filename, node, deltabase, uncompressedlen) | ||||
| # If we've read a lot of data from the mmap, free some memory. | # If we've read a lot of data from the mmap, free some memory. | ||||
| self._pagedin += offset - oldoffset | self._pagedin += offset - oldoffset | ||||
| if self.freememory(): | if self.freememory(): | ||||
| data = self._data | data = self._data | ||||
| libraries=[SHA1_LIBRARY], | libraries=[SHA1_LIBRARY], | ||||
| extra_compile_args=filter(None, [ | extra_compile_args=filter(None, [ | ||||
| STDC99, | STDC99, | ||||
| WALL, | WALL, | ||||
| WSTRICTPROTOTYPES, | WSTRICTPROTOTYPES, | ||||
| ] + cflags), | ] + cflags), | ||||
| ), | ), | ||||
| ], | ], | ||||
| 'extmmap': [ | |||||
| Extension('hgext3rd.extmmap', | |||||
| sources=['hgext3rd/extmmap.pyx'], | |||||
| extra_compile_args=filter(None, [ | |||||
| STDC99, WALL, WEXTRA, WCONVERSION, PEDANTIC, | |||||
| ]), | |||||
| ), | |||||
| ], | |||||
| 'linelog' : [ | 'linelog' : [ | ||||
| Extension('linelog', | Extension('linelog', | ||||
| sources=['linelog/pyext/linelog.pyx'], | sources=['linelog/pyext/linelog.pyx'], | ||||
| extra_compile_args=filter(None, [ | extra_compile_args=filter(None, [ | ||||
| STDC99, WALL, WEXTRA, WCONVERSION, PEDANTIC, | STDC99, WALL, WEXTRA, WCONVERSION, PEDANTIC, | ||||
| ]), | ]), | ||||
| ), | ), | ||||
| ], | ], | ||||
| components.append(dep) | components.append(dep) | ||||
| processdep = True | processdep = True | ||||
| if iswindows: | if iswindows: | ||||
| # The modules that successfully compile on Windows | # The modules that successfully compile on Windows | ||||
| cythonmodules = ['linelog'] | cythonmodules = ['linelog'] | ||||
| else: | else: | ||||
| cythonmodules = [ | cythonmodules = [ | ||||
| 'extmmap', | |||||
| 'linelog', | 'linelog', | ||||
| 'patchrmdir', | 'patchrmdir', | ||||
| 'traceprof', | 'traceprof', | ||||
| ] | ] | ||||
| if USECYTHON: | if USECYTHON: | ||||
| # see http://cython.readthedocs.io/en/latest/src/reference/compilation.html | # see http://cython.readthedocs.io/en/latest/src/reference/compilation.html | ||||
| compileroptions = { | compileroptions = { | ||||
| from remotefilelog import constants | from remotefilelog import constants | ||||
| from mercurial.node import nullid | from mercurial.node import nullid | ||||
| import mercurial.ui | import mercurial.ui | ||||
| class datapacktestsbase(object): | class datapacktestsbase(object): | ||||
| def __init__(self, datapackreader, paramsavailable): | def __init__(self, datapackreader, paramsavailable): | ||||
| self.datapackreader = datapackreader | self.datapackreader = datapackreader | ||||
| self.iscdatapack = not paramsavailable | |||||
| self.paramsavailable = paramsavailable | self.paramsavailable = paramsavailable | ||||
| def setUp(self): | def setUp(self): | ||||
| self.tempdirs = [] | self.tempdirs = [] | ||||
| def tearDown(self): | def tearDown(self): | ||||
| for d in self.tempdirs: | for d in self.tempdirs: | ||||
| shutil.rmtree(d) | shutil.rmtree(d) | ||||
| for (filename, node), content in blobs.iteritems(): | for (filename, node), content in blobs.iteritems(): | ||||
| actualcontent = pack.getdeltachain(filename, node)[0][4] | actualcontent = pack.getdeltachain(filename, node)[0][4] | ||||
| self.assertEquals(actualcontent, content) | self.assertEquals(actualcontent, content) | ||||
| def testPacksCache(self): | def testPacksCache(self): | ||||
| """Test that we remember the most recent packs while fetching the delta | """Test that we remember the most recent packs while fetching the delta | ||||
| chain.""" | chain.""" | ||||
| packdir = self.makeTempDir() | packdir = self.makeTempDir() | ||||
| deltachains = [] | deltachains = [] | ||||
| numpacks = 10 | # Ensures that we are not keeping everything in the cache. | ||||
| numpacks = datapackstore.DEFAULTCACHESIZE * 2 | |||||
| revisionsperpack = 100 | revisionsperpack = 100 | ||||
| for i in range(numpacks): | for i in range(numpacks): | ||||
| chain = [] | chain = [] | ||||
| revision = (str(i), self.getFakeHash(), nullid, "content") | revision = (str(i), self.getFakeHash(), nullid, "content") | ||||
| for _ in range(revisionsperpack): | for _ in range(revisionsperpack): | ||||
| chain.append(revision) | chain.append(revision) | ||||
| revision = ( | revision = ( | ||||
| str(i), | str(i), | ||||
| self.getFakeHash(), | self.getFakeHash(), | ||||
| revision[1], | revision[1], | ||||
| self.getFakeHash() | self.getFakeHash() | ||||
| ) | ) | ||||
| self.createPack(chain, packdir) | self.createPack(chain, packdir) | ||||
| deltachains.append(chain) | deltachains.append(chain) | ||||
| class testdatapackstore(datapackstore): | store = datapackstore(mercurial.ui.ui(), packdir, self.iscdatapack) | ||||
| # Ensures that we are not keeping everything in the cache. | |||||
| DEFAULTCACHESIZE = numpacks / 2 | |||||
| store = testdatapackstore(mercurial.ui.ui(), packdir) | |||||
| random.shuffle(deltachains) | random.shuffle(deltachains) | ||||
| for randomchain in deltachains: | for randomchain in deltachains: | ||||
| revision = random.choice(randomchain) | revision = random.choice(randomchain) | ||||
| chain = store.getdeltachain(revision[0], revision[1]) | chain = store.getdeltachain(revision[0], revision[1]) | ||||
| mostrecentpack = next(iter(store.packs), None) | mostrecentpack = next(iter(store.packs), None) | ||||
| self.assertEquals( | self.assertEquals( | ||||