/brz/remove-bazaar

To get this branch, use:
bzr branch http://gegoxaren.bato24.eu/bzr/brz/remove-bazaar

« back to all changes in this revision

Viewing changes to bzrlib/chk_map.py

  • Committer: John Arbash Meinel
  • Date: 2009-10-20 22:13:23 UTC
  • mto: This revision was merged to the branch mainline in revision 4771.
  • Revision ID: john@arbash-meinel.com-20091020221323-vvukgazqxkicb70n
A bit broken, but getting there.

Start being much stricter about requiring StaticTuples everywhere.
I may go back and loosen this restriction, but getting the code base
StaticTuple pure is probably a good idea. The main reason to be 'looser'
is so that things don't fail 'in the wild' just because someone
calls an api with a tuple rather than a StaticTuple.
However, I'd like the internals to be 'pure' if possible.
We'll see.

Show diffs side-by-side

added added

removed removed

Lines of Context:
1
 
# Copyright (C) 2008, 2009, 2010 Canonical Ltd
 
1
# Copyright (C) 2008, 2009 Canonical Ltd
2
2
#
3
3
# This program is free software; you can redistribute it and/or modify
4
4
# it under the terms of the GNU General Public License as published by
38
38
"""
39
39
 
40
40
import heapq
41
 
import threading
42
41
 
43
42
from bzrlib import lazy_import
44
43
lazy_import.lazy_import(globals(), """
51
50
    lru_cache,
52
51
    osutils,
53
52
    registry,
54
 
    static_tuple,
55
53
    trace,
56
54
    )
57
55
from bzrlib.static_tuple import StaticTuple
60
58
# If each line is 50 bytes, and you have 255 internal pages, with 255-way fan
61
59
# out, it takes 3.1MB to cache the layer.
62
60
_PAGE_CACHE_SIZE = 4*1024*1024
63
 
# Per thread caches for 2 reasons:
64
 
# - in the server we may be serving very different content, so we get less
65
 
#   cache thrashing.
66
 
# - we avoid locking on every cache lookup.
67
 
_thread_caches = threading.local()
68
 
# The page cache.
69
 
_thread_caches.page_cache = None
70
 
 
71
 
def _get_cache():
72
 
    """Get the per-thread page cache.
73
 
 
74
 
    We need a function to do this because in a new thread the _thread_caches
75
 
    threading.local object does not have the cache initialized yet.
76
 
    """
77
 
    page_cache = getattr(_thread_caches, 'page_cache', None)
78
 
    if page_cache is None:
79
 
        # We are caching bytes so len(value) is perfectly accurate
80
 
        page_cache = lru_cache.LRUSizeCache(_PAGE_CACHE_SIZE)
81
 
        _thread_caches.page_cache = page_cache
82
 
    return page_cache
83
 
 
 
61
# We are caching bytes so len(value) is perfectly accurate
 
62
_page_cache = lru_cache.LRUSizeCache(_PAGE_CACHE_SIZE)
84
63
 
85
64
def clear_cache():
86
 
    _get_cache().clear()
87
 
 
 
65
    _page_cache.clear()
88
66
 
89
67
# If a ChildNode falls below this many bytes, we check for a remap
90
68
_INTERESTING_NEW_SIZE = 50
106
84
class CHKMap(object):
107
85
    """A persistent map from string to string backed by a CHK store."""
108
86
 
109
 
    __slots__ = ('_store', '_root_node', '_search_key_func')
110
 
 
111
87
    def __init__(self, store, root_key, search_key_func=None):
112
88
        """Create a CHKMap object.
113
89
 
125
101
        if root_key is None:
126
102
            self._root_node = LeafNode(search_key_func=search_key_func)
127
103
        else:
 
104
            _check_key(root_key)
128
105
            self._root_node = self._node_key(root_key)
129
106
 
130
107
    def apply_delta(self, delta):
137
114
        """
138
115
        delete_count = 0
139
116
        # Check preconditions first.
140
 
        as_st = StaticTuple.from_sequence
141
 
        new_items = set([as_st(key) for (old, key, value) in delta
142
 
                         if key is not None and old is None])
 
117
        new_items = set([key for (old, key, value) in delta if key is not None
 
118
            and old is None])
143
119
        existing_new = list(self.iteritems(key_filter=new_items))
144
120
        if existing_new:
145
121
            raise errors.InconsistentDeltaDelta(delta,
182
158
 
183
159
    def _read_bytes(self, key):
184
160
        try:
185
 
            return _get_cache()[key]
 
161
            return _page_cache[key]
186
162
        except KeyError:
187
163
            stream = self._store.get_record_stream([key], 'unordered', True)
188
164
            bytes = stream.next().get_bytes_as('fulltext')
189
 
            _get_cache()[key] = bytes
 
165
            _page_cache[key] = bytes
190
166
            return bytes
191
167
 
192
168
    def _dump_tree(self, include_keys=False):
244
220
        root_key = klass._create_directly(store, initial_value,
245
221
            maximum_size=maximum_size, key_width=key_width,
246
222
            search_key_func=search_key_func)
247
 
        if type(root_key) is not StaticTuple:
248
 
            raise AssertionError('we got a %s instead of a StaticTuple'
249
 
                                 % (type(root_key),))
 
223
        assert type(root_key) is StaticTuple
250
224
        return root_key
251
225
 
252
226
    @classmethod
267
241
        node = LeafNode(search_key_func=search_key_func)
268
242
        node.set_maximum_size(maximum_size)
269
243
        node._key_width = key_width
270
 
        as_st = StaticTuple.from_sequence
271
 
        node._items = dict([(as_st(key), val) for key, val
272
 
                                               in initial_value.iteritems()])
 
244
        node._items = dict(initial_value)
273
245
        node._raw_size = sum([node._key_value_len(key, value)
274
 
                              for key,value in node._items.iteritems()])
 
246
                              for key,value in initial_value.iteritems()])
275
247
        node._len = len(node._items)
276
248
        node._compute_search_prefix()
277
249
        node._compute_serialised_prefix()
287
259
            for split, subnode in node_details:
288
260
                node.add_node(split, subnode)
289
261
        keys = list(node.serialise(store))
290
 
        return keys[-1]
 
262
        root_node = keys[-1]
 
263
        assert (type(root_node) is StaticTuple
 
264
                and len(root_node) == 1 and
 
265
                type(root_node[0]) is str)
 
266
        return root_node
291
267
 
292
268
    def iter_changes(self, basis):
293
269
        """Iterate over the changes between basis and self.
513
489
    def iteritems(self, key_filter=None):
514
490
        """Iterate over the entire CHKMap's contents."""
515
491
        self._ensure_root()
516
 
        if key_filter is not None:
517
 
            as_st = StaticTuple.from_sequence
518
 
            key_filter = [as_st(key) for key in key_filter]
519
492
        return self._root_node.iteritems(self._store, key_filter=key_filter)
520
493
 
521
494
    def key(self):
522
495
        """Return the key for this map."""
523
496
        if type(self._root_node) is StaticTuple:
 
497
            _check_key(self._root_node)
524
498
            return self._root_node
525
499
        else:
526
500
            return self._root_node._key
535
509
        :param key: A key to map.
536
510
        :param value: The value to assign to key.
537
511
        """
538
 
        key = StaticTuple.from_sequence(key)
539
512
        # Need a root object.
540
513
        self._ensure_root()
541
514
        prefix, node_details = self._root_node.map(self._store, key, value)
551
524
 
552
525
    def _node_key(self, node):
553
526
        """Get the key for a node whether it's a tuple or node."""
554
 
        if type(node) is tuple:
555
 
            node = StaticTuple.from_sequence(node)
556
527
        if type(node) is StaticTuple:
 
528
            _check_key(node)
557
529
            return node
 
530
        elif type(node) is tuple:
 
531
            raise TypeError('node %r should be a StaticTuple not tuple'
 
532
                            % (node,))
558
533
        else:
559
534
            return node._key
560
535
 
561
536
    def unmap(self, key, check_remap=True):
562
537
        """remove key from the map."""
563
 
        key = StaticTuple.from_sequence(key)
564
538
        self._ensure_root()
565
539
        if type(self._root_node) is InternalNode:
566
540
            unmapped = self._root_node.unmap(self._store, key,
594
568
        adding the header bytes, and without prefix compression.
595
569
    """
596
570
 
597
 
    __slots__ = ('_key', '_len', '_maximum_size', '_key_width',
598
 
                 '_raw_size', '_items', '_search_prefix', '_search_key_func'
599
 
                )
600
 
 
601
571
    def __init__(self, key_width=1):
602
572
        """Create a node.
603
573
 
692
662
        the key/value pairs.
693
663
    """
694
664
 
695
 
    __slots__ = ('_common_serialised_prefix',)
696
 
 
697
665
    def __init__(self, search_key_func=None):
698
666
        Node.__init__(self)
699
667
        # All of the keys in this leaf node share this common prefix
700
668
        self._common_serialised_prefix = None
 
669
        self._serialise_key = '\x00'.join
701
670
        if search_key_func is None:
702
671
            self._search_key_func = _search_key_plain
703
672
        else:
741
710
        :param bytes: The bytes of the node.
742
711
        :param key: The key that the serialised node has.
743
712
        """
744
 
        key = static_tuple.expect_static_tuple(key)
745
713
        return _deserialise_leaf_node(bytes, key,
746
714
                                      search_key_func=search_key_func)
747
715
 
887
855
                raise AssertionError('%r must be known' % self._search_prefix)
888
856
            return self._search_prefix, [("", self)]
889
857
 
890
 
    _serialise_key = '\x00'.join
891
 
 
892
858
    def serialise(self, store):
893
859
        """Serialise the LeafNode to store.
894
860
 
923
889
        bytes = ''.join(lines)
924
890
        if len(bytes) != self._current_size():
925
891
            raise AssertionError('Invalid _current_size')
926
 
        _get_cache().add(self._key, bytes)
 
892
        _page_cache.add(self._key, bytes)
927
893
        return [self._key]
928
894
 
929
895
    def refs(self):
993
959
        LeafNode or InternalNode.
994
960
    """
995
961
 
996
 
    __slots__ = ('_node_width',)
997
 
 
998
962
    def __init__(self, prefix='', search_key_func=None):
999
963
        Node.__init__(self)
1000
964
        # The size of an internalnode with default values and no children.
1042
1006
        :param key: The key that the serialised node has.
1043
1007
        :return: An InternalNode instance.
1044
1008
        """
1045
 
        key = static_tuple.expect_static_tuple(key)
 
1009
        if type(key) is not StaticTuple:
 
1010
            import pdb; pdb.set_trace()
 
1011
        key = StaticTuple.from_sequence(key).intern()
1046
1012
        return _deserialise_internal_node(bytes, key,
1047
1013
                                          search_key_func=search_key_func)
1048
1014
 
1165
1131
            found_keys = set()
1166
1132
            for key in keys:
1167
1133
                try:
1168
 
                    bytes = _get_cache()[key]
 
1134
                    bytes = _page_cache[key]
1169
1135
                except KeyError:
1170
1136
                    continue
1171
1137
                else:
1196
1162
                    prefix, node_key_filter = keys[record.key]
1197
1163
                    node_and_filters.append((node, node_key_filter))
1198
1164
                    self._items[prefix] = node
1199
 
                    _get_cache().add(record.key, bytes)
 
1165
                    _page_cache.add(record.key, bytes)
1200
1166
                for info in node_and_filters:
1201
1167
                    yield info
1202
1168
 
1322
1288
            lines.append(serialised[prefix_len:])
1323
1289
        sha1, _, _ = store.add_lines((None,), (), lines)
1324
1290
        self._key = StaticTuple("sha1:" + sha1,).intern()
1325
 
        _get_cache().add(self._key, ''.join(lines))
 
1291
        _page_cache.add(self._key, ''.join(lines))
1326
1292
        yield self._key
1327
1293
 
1328
1294
    def _search_key(self, key):
1476
1442
 
1477
1443
    def __init__(self, store, new_root_keys, old_root_keys,
1478
1444
                 search_key_func, pb=None):
1479
 
        # TODO: Should we add a StaticTuple barrier here? It would be nice to
1480
 
        #       force callers to use StaticTuple, because there will often be
1481
 
        #       lots of keys passed in here. And even if we cast it locally,
1482
 
        #       that just meanst that we will have *both* a StaticTuple and a
1483
 
        #       tuple() in memory, referring to the same object. (so a net
1484
 
        #       increase in memory, not a decrease.)
1485
1445
        self._store = store
1486
1446
        self._new_root_keys = new_root_keys
1487
1447
        self._old_root_keys = old_root_keys
1489
1449
        # All uninteresting chks that we have seen. By the time they are added
1490
1450
        # here, they should be either fully ignored, or queued up for
1491
1451
        # processing
1492
 
        # TODO: This might grow to a large size if there are lots of merge
1493
 
        #       parents, etc. However, it probably doesn't scale to O(history)
1494
 
        #       like _processed_new_refs does.
1495
1452
        self._all_old_chks = set(self._old_root_keys)
1496
1453
        # All items that we have seen from the old_root_keys
1497
1454
        self._all_old_items = set()
1498
1455
        # These are interesting items which were either read, or already in the
1499
1456
        # interesting queue (so we don't need to walk them again)
1500
 
        # TODO: processed_new_refs becomes O(all_chks), consider switching to
1501
 
        #       SimpleSet here.
1502
1457
        self._processed_new_refs = set()
1503
1458
        self._search_key_func = search_key_func
1504
1459
 
1511
1466
        self._state = None
1512
1467
 
1513
1468
    def _read_nodes_from_store(self, keys):
1514
 
        # We chose not to use _get_cache(), because we think in
1515
 
        # terms of records to be yielded. Also, we expect to touch each page
1516
 
        # only 1 time during this code. (We may want to evaluate saving the
1517
 
        # raw bytes into the page cache, which would allow a working tree
1518
 
        # update after the fetch to not have to read the bytes again.)
1519
 
        as_st = StaticTuple.from_sequence
 
1469
        # We chose not to use _page_cache, because we think in terms of records
 
1470
        # to be yielded. Also, we expect to touch each page only 1 time during
 
1471
        # this code. (We may want to evaluate saving the raw bytes into the
 
1472
        # page cache, which would allow a working tree update after the fetch
 
1473
        # to not have to read the bytes again.)
1520
1474
        stream = self._store.get_record_stream(keys, 'unordered', True)
1521
1475
        for record in stream:
1522
1476
            if self._pb is not None:
1529
1483
            if type(node) is InternalNode:
1530
1484
                # Note we don't have to do node.refs() because we know that
1531
1485
                # there are no children that have been pushed into this node
1532
 
                # Note: Using as_st() here seemed to save 1.2MB, which would
1533
 
                #       indicate that we keep 100k prefix_refs around while
1534
 
                #       processing. They *should* be shorter lived than that...
1535
 
                #       It does cost us ~10s of processing time
1536
 
                #prefix_refs = [as_st(item) for item in node._items.iteritems()]
1537
1486
                prefix_refs = node._items.items()
1538
1487
                items = []
1539
1488
            else:
1540
1489
                prefix_refs = []
1541
 
                # Note: We don't use a StaticTuple here. Profiling showed a
1542
 
                #       minor memory improvement (0.8MB out of 335MB peak 0.2%)
1543
 
                #       But a significant slowdown (15s / 145s, or 10%)
1544
1490
                items = node._items.items()
1545
1491
            yield record, node, prefix_refs, items
1546
1492
 
1554
1500
                                if p_r[1] not in all_old_chks]
1555
1501
            new_refs = [p_r[1] for p_r in prefix_refs]
1556
1502
            all_old_chks.update(new_refs)
1557
 
            # TODO: This might be a good time to turn items into StaticTuple
1558
 
            #       instances and possibly intern them. However, this does not
1559
 
            #       impact 'initial branch' performance, so I'm not worrying
1560
 
            #       about this yet
1561
1503
            self._all_old_items.update(items)
1562
1504
            # Queue up the uninteresting references
1563
1505
            # Don't actually put them in the 'to-read' queue until we have
1616
1558
            #       current design allows for this, as callers will do the work
1617
1559
            #       to make the results unique. We might profile whether we
1618
1560
            #       gain anything by ensuring unique return values for items
1619
 
            # TODO: This might be a good time to cast to StaticTuple, as
1620
 
            #       self._new_item_queue will hold the contents of multiple
1621
 
            #       records for an extended lifetime
1622
1561
            new_items = [item for item in items
1623
1562
                               if item not in self._all_old_items]
1624
1563
            self._new_item_queue.extend(new_items)
1649
1588
        if new_items:
1650
1589
            yield None, new_items
1651
1590
        refs = refs.difference(all_old_chks)
1652
 
        processed_new_refs.update(refs)
1653
1591
        while refs:
1654
 
            # TODO: Using a SimpleSet for self._processed_new_refs and
1655
 
            #       saved as much as 10MB of peak memory. However, it requires
1656
 
            #       implementing a non-pyrex version.
1657
1592
            next_refs = set()
1658
1593
            next_refs_update = next_refs.update
1659
1594
            # Inlining _read_nodes_from_store improves 'bzr branch bzr.dev'
1660
1595
            # from 1m54s to 1m51s. Consider it.
1661
1596
            for record, _, p_refs, items in self._read_nodes_from_store(refs):
1662
 
                if all_old_items:
1663
 
                    # using the 'if' check saves about 145s => 141s, when
1664
 
                    # streaming initial branch of Launchpad data.
1665
 
                    items = [item for item in items
1666
 
                             if item not in all_old_items]
 
1597
                items = [item for item in items
 
1598
                         if item not in all_old_items]
1667
1599
                yield record, items
1668
1600
                next_refs_update([p_r[1] for p_r in p_refs])
1669
 
                del p_refs
1670
 
            # set1.difference(set/dict) walks all of set1, and checks if it
1671
 
            # exists in 'other'.
1672
 
            # set1.difference(iterable) walks all of iterable, and does a
1673
 
            # 'difference_update' on a clone of set1. Pick wisely based on the
1674
 
            # expected sizes of objects.
1675
 
            # in our case it is expected that 'new_refs' will always be quite
1676
 
            # small.
1677
1601
            next_refs = next_refs.difference(all_old_chks)
1678
1602
            next_refs = next_refs.difference(processed_new_refs)
1679
1603
            processed_new_refs.update(next_refs)
1686
1610
        self._old_queue = []
1687
1611
        all_old_chks = self._all_old_chks
1688
1612
        for record, _, prefix_refs, items in self._read_nodes_from_store(refs):
1689
 
            # TODO: Use StaticTuple here?
1690
1613
            self._all_old_items.update(items)
1691
1614
            refs = [r for _,r in prefix_refs if r not in all_old_chks]
1692
1615
            self._old_queue.extend(refs)
1743
1666
search_key_registry.register('hash-16-way', _search_key_16)
1744
1667
search_key_registry.register('hash-255-way', _search_key_255)
1745
1668
 
1746
 
 
1747
1669
def _check_key(key):
1748
 
    """Helper function to assert that a key is properly formatted.
1749
 
 
1750
 
    This generally shouldn't be used in production code, but it can be helpful
1751
 
    to debug problems.
1752
 
    """
1753
1670
    if type(key) is not StaticTuple:
1754
1671
        raise TypeError('key %r is not StaticTuple but %s' % (key, type(key)))
1755
1672
    if len(key) != 1: