Mercurial > hg
view tests/test-ctxmanager.py @ 27784:432242f41d9f
obsolete: make _computeobsoleteset much faster
This patch makes _computeobsoleteset much faster by looping
over the draft and secrets as opposed to looping over the
successors.
This works because "number of draft and secret" is typically
way smaller(<100) than the number of successor in the repo (~90k in
my checkout of core mercurial as of today). And also because
it is very fast to compute "not public()".
I timed the code with the following setup:
"""
from mercurial import hg, ui, obsolete
ui = ui.ui()
repo = hg.repository(ui, "~/hg")
l = repo.obsstore.successors # This caches the result
"""
With about 90k successors.
k=obsolete._computeobsoleteset(repo) before this patch:
10 loops, best of 3: 33.9 ms per loop
k=obsolete._computeobsoleteset(repo) after this patch:
10000 loops, best of 3: 83.3 µs per loop
author | Laurent Charignon <lc2817@columbia.edu> |
---|---|
date | Wed, 13 Jan 2016 21:52:26 -0800 |
parents | 4e27c0a70574 |
children | ba427b51f1d8 |
line wrap: on
line source
from __future__ import absolute_import import silenttestrunner import unittest from mercurial.util import ctxmanager class contextmanager(object): def __init__(self, name, trace): self.name = name self.entered = False self.exited = False self.trace = trace def __enter__(self): self.entered = True self.trace(('enter', self.name)) return self def __exit__(self, exc_type, exc_val, exc_tb): self.exited = exc_type, exc_val, exc_tb self.trace(('exit', self.name)) def __repr__(self): return '<ctx %r>' % self.name class ctxerror(Exception): pass class raise_on_enter(contextmanager): def __enter__(self): self.trace(('raise', self.name)) raise ctxerror(self.name) class raise_on_exit(contextmanager): def __exit__(self, exc_type, exc_val, exc_tb): self.trace(('raise', self.name)) raise ctxerror(self.name) def ctxmgr(name, trace): return lambda: contextmanager(name, trace) class test_ctxmanager(unittest.TestCase): def test_basics(self): trace = [] addtrace = trace.append with ctxmanager(ctxmgr('a', addtrace), ctxmgr('b', addtrace)) as c: a, b = c() c.atexit(addtrace, ('atexit', 'x')) c.atexit(addtrace, ('atexit', 'y')) self.assertEqual(trace, [('enter', 'a'), ('enter', 'b'), ('atexit', 'y'), ('atexit', 'x'), ('exit', 'b'), ('exit', 'a')]) def test_raise_on_enter(self): trace = [] addtrace = trace.append with self.assertRaises(ctxerror): with ctxmanager(ctxmgr('a', addtrace), lambda: raise_on_enter('b', addtrace)) as c: c() addtrace('unreachable') self.assertEqual(trace, [('enter', 'a'), ('raise', 'b'), ('exit', 'a')]) def test_raise_on_exit(self): trace = [] addtrace = trace.append with self.assertRaises(ctxerror): with ctxmanager(ctxmgr('a', addtrace), lambda: raise_on_exit('b', addtrace)) as c: c() addtrace('running') self.assertEqual(trace, [('enter', 'a'), ('enter', 'b'), 'running', ('raise', 'b'), ('exit', 'a')]) if __name__ == '__main__': silenttestrunner.main(__name__)