← Back to branch summary

~bzr-pqm/bzr/bzr.dev

« back to all changes in this revision

Viewing changes to bzrlib/chk_map.py

  • Committer: John Arbash Meinel
  • Date: 2009-06-19 17:53:37 UTC
  • mto: This revision was merged to the branch mainline in revision 4466.
  • Revision ID: john@arbash-meinel.com-20090619175337-uozt3bntdd48lh4z
Update time_graph to use X:1 ratios rather than 0.xxx ratios.
It is just easier to track now that the new code is much faster.

Show diffs side-by-side

added added

removed removed

Lines of Context:
bzrlib/chk_map.py
1
 
# Copyright (C) 2008, 2009, 2010 Canonical Ltd
 
1
# Copyright (C) 2008 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
26
26
 
27
27
Updates to a CHKMap are done preferentially via the apply_delta method, to
28
28
allow optimisation of the update operation; but individual map/unmap calls are
29
 
possible and supported. Individual changes via map/unmap are buffered in memory
30
 
until the _save method is called to force serialisation of the tree.
31
 
apply_delta records its changes immediately by performing an implicit _save.
 
29
possible and supported. All changes via map/unmap are buffered in memory until
 
30
the _save method is called to force serialisation of the tree. apply_delta
 
31
performs a _save implicitly.
32
32
 
33
33
TODO:
34
34
-----
38
38
"""
39
39
 
40
40
import heapq
41
 
import threading
 
41
import time
42
42
 
43
43
from bzrlib import lazy_import
44
44
lazy_import.lazy_import(globals(), """
 
45
from bzrlib import versionedfile
 
46
""")
45
47
from bzrlib import (
46
48
    errors,
47
 
    versionedfile,
48
 
    )
49
 
""")
50
 
from bzrlib import (
51
49
    lru_cache,
52
50
    osutils,
53
51
    registry,
54
 
    static_tuple,
55
52
    trace,
56
53
    )
57
 
from bzrlib.static_tuple import StaticTuple
58
54
 
59
55
# approx 4MB
60
56
# If each line is 50 bytes, and you have 255 internal pages, with 255-way fan
61
57
# out, it takes 3.1MB to cache the layer.
62
58
_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
 
 
84
 
 
85
 
def clear_cache():
86
 
    _get_cache().clear()
87
 
 
 
59
# We are caching bytes so len(value) is perfectly accurate
 
60
_page_cache = lru_cache.LRUSizeCache(_PAGE_CACHE_SIZE)
88
61
 
89
62
# If a ChildNode falls below this many bytes, we check for a remap
90
63
_INTERESTING_NEW_SIZE = 50
91
64
# If a ChildNode shrinks by more than this amount, we check for a remap
92
65
_INTERESTING_SHRINKAGE_LIMIT = 20
 
66
# If we delete more than this many nodes applying a delta, we check for a remap
 
67
_INTERESTING_DELETES_LIMIT = 5
93
68
 
94
69
 
95
70
def _search_key_plain(key):
104
79
class CHKMap(object):
105
80
    """A persistent map from string to string backed by a CHK store."""
106
81
 
107
 
    __slots__ = ('_store', '_root_node', '_search_key_func')
108
 
 
109
82
    def __init__(self, store, root_key, search_key_func=None):
110
83
        """Create a CHKMap object.
111
84
 
133
106
            into the map; if old_key is not None, then the old mapping
134
107
            of old_key is removed.
135
108
        """
136
 
        has_deletes = False
137
 
        # Check preconditions first.
138
 
        as_st = StaticTuple.from_sequence
139
 
        new_items = set([as_st(key) for (old, key, value) in delta
140
 
                         if key is not None and old is None])
141
 
        existing_new = list(self.iteritems(key_filter=new_items))
142
 
        if existing_new:
143
 
            raise errors.InconsistentDeltaDelta(delta,
144
 
                "New items are already in the map %r." % existing_new)
145
 
        # Now apply changes.
 
109
        delete_count = 0
146
110
        for old, new, value in delta:
147
111
            if old is not None and old != new:
148
112
                self.unmap(old, check_remap=False)
149
 
                has_deletes = True
 
113
                delete_count += 1
150
114
        for old, new, value in delta:
151
115
            if new is not None:
152
116
                self.map(new, value)
153
 
        if has_deletes:
 
117
        if delete_count > _INTERESTING_DELETES_LIMIT:
 
118
            trace.mutter("checking remap as %d deletions", delete_count)
154
119
            self._check_remap()
155
120
        return self._save()
156
121
 
157
122
    def _ensure_root(self):
158
123
        """Ensure that the root node is an object not a key."""
159
 
        if type(self._root_node) is StaticTuple:
 
124
        if type(self._root_node) is tuple:
160
125
            # Demand-load the root
161
126
            self._root_node = self._get_node(self._root_node)
162
127
 
170
135
        :param node: A tuple key or node object.
171
136
        :return: A node object.
172
137
        """
173
 
        if type(node) is StaticTuple:
 
138
        if type(node) is tuple:
174
139
            bytes = self._read_bytes(node)
175
140
            return _deserialise(bytes, node,
176
141
                search_key_func=self._search_key_func)
179
144
 
180
145
    def _read_bytes(self, key):
181
146
        try:
182
 
            return _get_cache()[key]
 
147
            return _page_cache[key]
183
148
        except KeyError:
184
149
            stream = self._store.get_record_stream([key], 'unordered', True)
185
150
            bytes = stream.next().get_bytes_as('fulltext')
186
 
            _get_cache()[key] = bytes
 
151
            _page_cache[key] = bytes
187
152
            return bytes
188
153
 
189
154
    def _dump_tree(self, include_keys=False):
217
182
            for key, value in sorted(node._items.iteritems()):
218
183
                # Don't use prefix nor indent here to line up when used in
219
184
                # tests in conjunction with assertEqualDiff
220
 
                result.append('      %r %r' % (tuple(key), value))
 
185
                result.append('      %r %r' % (key, value))
221
186
        return result
222
187
 
223
188
    @classmethod
238
203
            multiple pages.
239
204
        :return: The root chk of the resulting CHKMap.
240
205
        """
241
 
        root_key = klass._create_directly(store, initial_value,
242
 
            maximum_size=maximum_size, key_width=key_width,
243
 
            search_key_func=search_key_func)
244
 
        if type(root_key) is not StaticTuple:
245
 
            raise AssertionError('we got a %s instead of a StaticTuple'
246
 
                                 % (type(root_key),))
247
 
        return root_key
248
 
 
249
 
    @classmethod
250
 
    def _create_via_map(klass, store, initial_value, maximum_size=0,
251
 
                        key_width=1, search_key_func=None):
252
 
        result = klass(store, None, search_key_func=search_key_func)
 
206
        result = CHKMap(store, None, search_key_func=search_key_func)
253
207
        result._root_node.set_maximum_size(maximum_size)
254
208
        result._root_node._key_width = key_width
255
209
        delta = []
256
210
        for key, value in initial_value.items():
257
211
            delta.append((None, key, value))
258
 
        root_key = result.apply_delta(delta)
259
 
        return root_key
260
 
 
261
 
    @classmethod
262
 
    def _create_directly(klass, store, initial_value, maximum_size=0,
263
 
                         key_width=1, search_key_func=None):
264
 
        node = LeafNode(search_key_func=search_key_func)
265
 
        node.set_maximum_size(maximum_size)
266
 
        node._key_width = key_width
267
 
        as_st = StaticTuple.from_sequence
268
 
        node._items = dict([(as_st(key), val) for key, val
269
 
                                               in initial_value.iteritems()])
270
 
        node._raw_size = sum([node._key_value_len(key, value)
271
 
                              for key,value in node._items.iteritems()])
272
 
        node._len = len(node._items)
273
 
        node._compute_search_prefix()
274
 
        node._compute_serialised_prefix()
275
 
        if (node._len > 1
276
 
            and maximum_size
277
 
            and node._current_size() > maximum_size):
278
 
            prefix, node_details = node._split(store)
279
 
            if len(node_details) == 1:
280
 
                raise AssertionError('Failed to split using node._split')
281
 
            node = InternalNode(prefix, search_key_func=search_key_func)
282
 
            node.set_maximum_size(maximum_size)
283
 
            node._key_width = key_width
284
 
            for split, subnode in node_details:
285
 
                node.add_node(split, subnode)
286
 
        keys = list(node.serialise(store))
287
 
        return keys[-1]
 
212
        return result.apply_delta(delta)
288
213
 
289
214
    def iter_changes(self, basis):
290
215
        """Iterate over the changes between basis and self.
510
435
    def iteritems(self, key_filter=None):
511
436
        """Iterate over the entire CHKMap's contents."""
512
437
        self._ensure_root()
513
 
        if key_filter is not None:
514
 
            as_st = StaticTuple.from_sequence
515
 
            key_filter = [as_st(key) for key in key_filter]
516
438
        return self._root_node.iteritems(self._store, key_filter=key_filter)
517
439
 
518
440
    def key(self):
519
441
        """Return the key for this map."""
520
 
        if type(self._root_node) is StaticTuple:
 
442
        if type(self._root_node) is tuple:
521
443
            return self._root_node
522
444
        else:
523
445
            return self._root_node._key
527
449
        return len(self._root_node)
528
450
 
529
451
    def map(self, key, value):
530
 
        """Map a key tuple to value.
531
 
        
532
 
        :param key: A key to map.
533
 
        :param value: The value to assign to key.
534
 
        """
535
 
        key = StaticTuple.from_sequence(key)
 
452
        """Map a key tuple to value."""
536
453
        # Need a root object.
537
454
        self._ensure_root()
538
455
        prefix, node_details = self._root_node.map(self._store, key, value)
549
466
    def _node_key(self, node):
550
467
        """Get the key for a node whether it's a tuple or node."""
551
468
        if type(node) is tuple:
552
 
            node = StaticTuple.from_sequence(node)
553
 
        if type(node) is StaticTuple:
554
469
            return node
555
470
        else:
556
471
            return node._key
557
472
 
558
473
    def unmap(self, key, check_remap=True):
559
474
        """remove key from the map."""
560
 
        key = StaticTuple.from_sequence(key)
561
475
        self._ensure_root()
562
476
        if type(self._root_node) is InternalNode:
563
477
            unmapped = self._root_node.unmap(self._store, key,
570
484
        """Check if nodes can be collapsed."""
571
485
        self._ensure_root()
572
486
        if type(self._root_node) is InternalNode:
573
 
            self._root_node = self._root_node._check_remap(self._store)
 
487
            self._root_node._check_remap(self._store)
574
488
 
575
489
    def _save(self):
576
490
        """Save the map completely.
577
491
 
578
492
        :return: The key of the root node.
579
493
        """
580
 
        if type(self._root_node) is StaticTuple:
 
494
        if type(self._root_node) is tuple:
581
495
            # Already saved.
582
496
            return self._root_node
583
497
        keys = list(self._root_node.serialise(self._store))
591
505
        adding the header bytes, and without prefix compression.
592
506
    """
593
507
 
594
 
    __slots__ = ('_key', '_len', '_maximum_size', '_key_width',
595
 
                 '_raw_size', '_items', '_search_prefix', '_search_key_func'
596
 
                )
597
 
 
598
508
    def __init__(self, key_width=1):
599
509
        """Create a node.
600
510
 
689
599
        the key/value pairs.
690
600
    """
691
601
 
692
 
    __slots__ = ('_common_serialised_prefix',)
693
 
 
694
602
    def __init__(self, search_key_func=None):
695
603
        Node.__init__(self)
696
604
        # All of the keys in this leaf node share this common prefix
697
605
        self._common_serialised_prefix = None
 
606
        self._serialise_key = '\x00'.join
698
607
        if search_key_func is None:
699
608
            self._search_key_func = _search_key_plain
700
609
        else:
738
647
        :param bytes: The bytes of the node.
739
648
        :param key: The key that the serialised node has.
740
649
        """
741
 
        key = static_tuple.expect_static_tuple(key)
742
650
        return _deserialise_leaf_node(bytes, key,
743
651
                                      search_key_func=search_key_func)
744
652
 
856
764
                result[prefix] = node
857
765
            else:
858
766
                node = result[prefix]
859
 
            sub_prefix, node_details = node.map(store, key, value)
860
 
            if len(node_details) > 1:
861
 
                if prefix != sub_prefix:
862
 
                    # This node has been split and is now found via a different
863
 
                    # path
864
 
                    result.pop(prefix)
865
 
                new_node = InternalNode(sub_prefix,
866
 
                    search_key_func=self._search_key_func)
867
 
                new_node.set_maximum_size(self._maximum_size)
868
 
                new_node._key_width = self._key_width
869
 
                for split, node in node_details:
870
 
                    new_node.add_node(split, node)
871
 
                result[prefix] = new_node
 
767
            node.map(store, key, value)
872
768
        return common_prefix, result.items()
873
769
 
874
770
    def map(self, store, key, value):
884
780
                raise AssertionError('%r must be known' % self._search_prefix)
885
781
            return self._search_prefix, [("", self)]
886
782
 
887
 
    _serialise_key = '\x00'.join
888
 
 
889
783
    def serialise(self, store):
890
784
        """Serialise the LeafNode to store.
891
785
 
916
810
            lines.append(serialized[prefix_len:])
917
811
            lines.extend(value_lines)
918
812
        sha1, _, _ = store.add_lines((None,), (), lines)
919
 
        self._key = StaticTuple("sha1:" + sha1,).intern()
 
813
        self._key = ("sha1:" + sha1,)
920
814
        bytes = ''.join(lines)
921
815
        if len(bytes) != self._current_size():
922
816
            raise AssertionError('Invalid _current_size')
923
 
        _get_cache().add(self._key, bytes)
 
817
        _page_cache.add(self._key, bytes)
924
818
        return [self._key]
925
819
 
926
820
    def refs(self):
990
884
        LeafNode or InternalNode.
991
885
    """
992
886
 
993
 
    __slots__ = ('_node_width',)
994
 
 
995
887
    def __init__(self, prefix='', search_key_func=None):
996
888
        Node.__init__(self)
997
889
        # The size of an internalnode with default values and no children.
1039
931
        :param key: The key that the serialised node has.
1040
932
        :return: An InternalNode instance.
1041
933
        """
1042
 
        key = static_tuple.expect_static_tuple(key)
1043
934
        return _deserialise_internal_node(bytes, key,
1044
935
                                          search_key_func=search_key_func)
1045
936
 
1070
961
            # for whatever we are missing
1071
962
            shortcut = True
1072
963
            for prefix, node in self._items.iteritems():
1073
 
                if node.__class__ is StaticTuple:
 
964
                if node.__class__ is tuple:
1074
965
                    keys[node] = (prefix, None)
1075
966
                else:
1076
967
                    yield node, None
1105
996
                    # A given key can only match 1 child node, if it isn't
1106
997
                    # there, then we can just return nothing
1107
998
                    return
1108
 
                if node.__class__ is StaticTuple:
 
999
                if node.__class__ is tuple:
1109
1000
                    keys[node] = (search_prefix, [key])
1110
1001
                else:
1111
1002
                    # This is loaded, and the only thing that can match,
1138
1029
                        # We can ignore this one
1139
1030
                        continue
1140
1031
                    node_key_filter = prefix_to_keys[search_prefix]
1141
 
                    if node.__class__ is StaticTuple:
 
1032
                    if node.__class__ is tuple:
1142
1033
                        keys[node] = (search_prefix, node_key_filter)
1143
1034
                    else:
1144
1035
                        yield node, node_key_filter
1153
1044
                        if sub_prefix in length_filter:
1154
1045
                            node_key_filter.extend(prefix_to_keys[sub_prefix])
1155
1046
                    if node_key_filter: # this key matched something, yield it
1156
 
                        if node.__class__ is StaticTuple:
 
1047
                        if node.__class__ is tuple:
1157
1048
                            keys[node] = (prefix, node_key_filter)
1158
1049
                        else:
1159
1050
                            yield node, node_key_filter
1162
1053
            found_keys = set()
1163
1054
            for key in keys:
1164
1055
                try:
1165
 
                    bytes = _get_cache()[key]
 
1056
                    bytes = _page_cache[key]
1166
1057
                except KeyError:
1167
1058
                    continue
1168
1059
                else:
1193
1084
                    prefix, node_key_filter = keys[record.key]
1194
1085
                    node_and_filters.append((node, node_key_filter))
1195
1086
                    self._items[prefix] = node
1196
 
                    _get_cache().add(record.key, bytes)
 
1087
                    _page_cache.add(record.key, bytes)
1197
1088
                for info in node_and_filters:
1198
1089
                    yield info
1199
1090
 
1291
1182
        :return: An iterable of the keys inserted by this operation.
1292
1183
        """
1293
1184
        for node in self._items.itervalues():
1294
 
            if type(node) is StaticTuple:
 
1185
            if type(node) is tuple:
1295
1186
                # Never deserialised.
1296
1187
                continue
1297
1188
            if node._key is not None:
1308
1199
        lines.append('%s\n' % (self._search_prefix,))
1309
1200
        prefix_len = len(self._search_prefix)
1310
1201
        for prefix, node in sorted(self._items.items()):
1311
 
            if type(node) is StaticTuple:
 
1202
            if type(node) is tuple:
1312
1203
                key = node[0]
1313
1204
            else:
1314
1205
                key = node._key[0]
1318
1209
                    % (serialised, self._search_prefix))
1319
1210
            lines.append(serialised[prefix_len:])
1320
1211
        sha1, _, _ = store.add_lines((None,), (), lines)
1321
 
        self._key = StaticTuple("sha1:" + sha1,).intern()
1322
 
        _get_cache().add(self._key, ''.join(lines))
 
1212
        self._key = ("sha1:" + sha1,)
 
1213
        _page_cache.add(self._key, ''.join(lines))
1323
1214
        yield self._key
1324
1215
 
1325
1216
    def _search_key(self, key):
1353
1244
            raise AssertionError("unserialised nodes have no refs.")
1354
1245
        refs = []
1355
1246
        for value in self._items.itervalues():
1356
 
            if type(value) is StaticTuple:
 
1247
            if type(value) is tuple:
1357
1248
                refs.append(value)
1358
1249
            else:
1359
1250
                refs.append(value.key())
1369
1260
        return self._search_prefix
1370
1261
 
1371
1262
    def unmap(self, store, key, check_remap=True):
1372
 
        """Remove key from this node and its children."""
 
1263
        """Remove key from this node and it's children."""
1373
1264
        if not len(self._items):
1374
1265
            raise AssertionError("can't unmap in an empty InternalNode.")
1375
1266
        children = [node for node, _
1460
1351
    return node
1461
1352
 
1462
1353
 
1463
 
class CHKMapDifference(object):
1464
 
    """Iterate the stored pages and key,value pairs for (new - old).
1465
 
 
1466
 
    This class provides a generator over the stored CHK pages and the
1467
 
    (key, value) pairs that are in any of the new maps and not in any of the
1468
 
    old maps.
1469
 
 
1470
 
    Note that it may yield chk pages that are common (especially root nodes),
1471
 
    but it won't yield (key,value) pairs that are common.
1472
 
    """
1473
 
 
1474
 
    def __init__(self, store, new_root_keys, old_root_keys,
1475
 
                 search_key_func, pb=None):
1476
 
        # TODO: Should we add a StaticTuple barrier here? It would be nice to
1477
 
        #       force callers to use StaticTuple, because there will often be
1478
 
        #       lots of keys passed in here. And even if we cast it locally,
1479
 
        #       that just meanst that we will have *both* a StaticTuple and a
1480
 
        #       tuple() in memory, referring to the same object. (so a net
1481
 
        #       increase in memory, not a decrease.)
1482
 
        self._store = store
1483
 
        self._new_root_keys = new_root_keys
1484
 
        self._old_root_keys = old_root_keys
1485
 
        self._pb = pb
1486
 
        # All uninteresting chks that we have seen. By the time they are added
1487
 
        # here, they should be either fully ignored, or queued up for
1488
 
        # processing
1489
 
        # TODO: This might grow to a large size if there are lots of merge
1490
 
        #       parents, etc. However, it probably doesn't scale to O(history)
1491
 
        #       like _processed_new_refs does.
1492
 
        self._all_old_chks = set(self._old_root_keys)
1493
 
        # All items that we have seen from the old_root_keys
1494
 
        self._all_old_items = set()
1495
 
        # These are interesting items which were either read, or already in the
1496
 
        # interesting queue (so we don't need to walk them again)
1497
 
        # TODO: processed_new_refs becomes O(all_chks), consider switching to
1498
 
        #       SimpleSet here.
1499
 
        self._processed_new_refs = set()
1500
 
        self._search_key_func = search_key_func
1501
 
 
1502
 
        # The uninteresting and interesting nodes to be searched
1503
 
        self._old_queue = []
1504
 
        self._new_queue = []
1505
 
        # Holds the (key, value) items found when processing the root nodes,
1506
 
        # waiting for the uninteresting nodes to be walked
1507
 
        self._new_item_queue = []
1508
 
        self._state = None
1509
 
 
1510
 
    def _read_nodes_from_store(self, keys):
1511
 
        # We chose not to use _get_cache(), because we think in
1512
 
        # terms of records to be yielded. Also, we expect to touch each page
1513
 
        # only 1 time during this code. (We may want to evaluate saving the
1514
 
        # raw bytes into the page cache, which would allow a working tree
1515
 
        # update after the fetch to not have to read the bytes again.)
1516
 
        as_st = StaticTuple.from_sequence
1517
 
        stream = self._store.get_record_stream(keys, 'unordered', True)
1518
 
        for record in stream:
1519
 
            if self._pb is not None:
1520
 
                self._pb.tick()
1521
 
            if record.storage_kind == 'absent':
1522
 
                raise errors.NoSuchRevision(self._store, record.key)
 
1354
def _find_children_info(store, interesting_keys, uninteresting_keys, pb):
 
1355
    """Read the associated records, and determine what is interesting."""
 
1356
    uninteresting_keys = set(uninteresting_keys)
 
1357
    chks_to_read = uninteresting_keys.union(interesting_keys)
 
1358
    next_uninteresting = set()
 
1359
    next_interesting = set()
 
1360
    uninteresting_items = set()
 
1361
    interesting_items = set()
 
1362
    interesting_to_yield = []
 
1363
    for record in store.get_record_stream(chks_to_read, 'unordered', True):
 
1364
        # records_read.add(record.key())
 
1365
        if pb is not None:
 
1366
            pb.tick()
 
1367
        bytes = record.get_bytes_as('fulltext')
 
1368
        # We don't care about search_key_func for this code, because we only
 
1369
        # care about external references.
 
1370
        node = _deserialise(bytes, record.key, search_key_func=None)
 
1371
        if record.key in uninteresting_keys:
 
1372
            if type(node) is InternalNode:
 
1373
                next_uninteresting.update(node.refs())
 
1374
            else:
 
1375
                # We know we are at a LeafNode, so we can pass None for the
 
1376
                # store
 
1377
                uninteresting_items.update(node.iteritems(None))
 
1378
        else:
 
1379
            interesting_to_yield.append(record.key)
 
1380
            if type(node) is InternalNode:
 
1381
                next_interesting.update(node.refs())
 
1382
            else:
 
1383
                interesting_items.update(node.iteritems(None))
 
1384
    return (next_uninteresting, uninteresting_items,
 
1385
            next_interesting, interesting_to_yield, interesting_items)
 
1386
 
 
1387
 
 
1388
def _find_all_uninteresting(store, interesting_root_keys,
 
1389
                            uninteresting_root_keys, pb):
 
1390
    """Determine the full set of uninteresting keys."""
 
1391
    # What about duplicates between interesting_root_keys and
 
1392
    # uninteresting_root_keys?
 
1393
    if not uninteresting_root_keys:
 
1394
        # Shortcut case. We know there is nothing uninteresting to filter out
 
1395
        # So we just let the rest of the algorithm do the work
 
1396
        # We know there is nothing uninteresting, and we didn't have to read
 
1397
        # any interesting records yet.
 
1398
        return (set(), set(), set(interesting_root_keys), [], set())
 
1399
    all_uninteresting_chks = set(uninteresting_root_keys)
 
1400
    all_uninteresting_items = set()
 
1401
 
 
1402
    # First step, find the direct children of both the interesting and
 
1403
    # uninteresting set
 
1404
    (uninteresting_keys, uninteresting_items,
 
1405
     interesting_keys, interesting_to_yield,
 
1406
     interesting_items) = _find_children_info(store, interesting_root_keys,
 
1407
                                              uninteresting_root_keys,
 
1408
                                              pb=pb)
 
1409
    all_uninteresting_chks.update(uninteresting_keys)
 
1410
    all_uninteresting_items.update(uninteresting_items)
 
1411
    del uninteresting_items
 
1412
    # Note: Exact matches between interesting and uninteresting do not need
 
1413
    #       to be search further. Non-exact matches need to be searched in case
 
1414
    #       there is a future exact-match
 
1415
    uninteresting_keys.difference_update(interesting_keys)
 
1416
 
 
1417
    # Second, find the full set of uninteresting bits reachable by the
 
1418
    # uninteresting roots
 
1419
    chks_to_read = uninteresting_keys
 
1420
    while chks_to_read:
 
1421
        next_chks = set()
 
1422
        for record in store.get_record_stream(chks_to_read, 'unordered', False):
 
1423
            # TODO: Handle 'absent'
 
1424
            if pb is not None:
 
1425
                pb.tick()
1523
1426
            bytes = record.get_bytes_as('fulltext')
1524
 
            node = _deserialise(bytes, record.key,
1525
 
                                search_key_func=self._search_key_func)
 
1427
            # We don't care about search_key_func for this code, because we
 
1428
            # only care about external references.
 
1429
            node = _deserialise(bytes, record.key, search_key_func=None)
1526
1430
            if type(node) is InternalNode:
1527
 
                # Note we don't have to do node.refs() because we know that
1528
 
                # there are no children that have been pushed into this node
1529
 
                # Note: Using as_st() here seemed to save 1.2MB, which would
1530
 
                #       indicate that we keep 100k prefix_refs around while
1531
 
                #       processing. They *should* be shorter lived than that...
1532
 
                #       It does cost us ~10s of processing time
1533
 
                #prefix_refs = [as_st(item) for item in node._items.iteritems()]
1534
 
                prefix_refs = node._items.items()
1535
 
                items = []
 
1431
                # uninteresting_prefix_chks.update(node._items.iteritems())
 
1432
                chks = node._items.values()
 
1433
                # TODO: We remove the entries that are already in
 
1434
                #       uninteresting_chks ?
 
1435
                next_chks.update(chks)
 
1436
                all_uninteresting_chks.update(chks)
1536
1437
            else:
1537
 
                prefix_refs = []
1538
 
                # Note: We don't use a StaticTuple here. Profiling showed a
1539
 
                #       minor memory improvement (0.8MB out of 335MB peak 0.2%)
1540
 
                #       But a significant slowdown (15s / 145s, or 10%)
1541
 
                items = node._items.items()
1542
 
            yield record, node, prefix_refs, items
1543
 
 
1544
 
    def _read_old_roots(self):
1545
 
        old_chks_to_enqueue = []
1546
 
        all_old_chks = self._all_old_chks
1547
 
        for record, node, prefix_refs, items in \
1548
 
                self._read_nodes_from_store(self._old_root_keys):
1549
 
            # Uninteresting node
1550
 
            prefix_refs = [p_r for p_r in prefix_refs
1551
 
                                if p_r[1] not in all_old_chks]
1552
 
            new_refs = [p_r[1] for p_r in prefix_refs]
1553
 
            all_old_chks.update(new_refs)
1554
 
            # TODO: This might be a good time to turn items into StaticTuple
1555
 
            #       instances and possibly intern them. However, this does not
1556
 
            #       impact 'initial branch' performance, so I'm not worrying
1557
 
            #       about this yet
1558
 
            self._all_old_items.update(items)
1559
 
            # Queue up the uninteresting references
1560
 
            # Don't actually put them in the 'to-read' queue until we have
1561
 
            # finished checking the interesting references
1562
 
            old_chks_to_enqueue.extend(prefix_refs)
1563
 
        return old_chks_to_enqueue
1564
 
 
1565
 
    def _enqueue_old(self, new_prefixes, old_chks_to_enqueue):
1566
 
        # At this point, we have read all the uninteresting and interesting
1567
 
        # items, so we can queue up the uninteresting stuff, knowing that we've
1568
 
        # handled the interesting ones
1569
 
        for prefix, ref in old_chks_to_enqueue:
1570
 
            not_interesting = True
1571
 
            for i in xrange(len(prefix), 0, -1):
1572
 
                if prefix[:i] in new_prefixes:
1573
 
                    not_interesting = False
1574
 
                    break
1575
 
            if not_interesting:
1576
 
                # This prefix is not part of the remaining 'interesting set'
1577
 
                continue
1578
 
            self._old_queue.append(ref)
1579
 
 
1580
 
    def _read_all_roots(self):
1581
 
        """Read the root pages.
1582
 
 
1583
 
        This is structured as a generator, so that the root records can be
1584
 
        yielded up to whoever needs them without any buffering.
1585
 
        """
1586
 
        # This is the bootstrap phase
1587
 
        if not self._old_root_keys:
1588
 
            # With no old_root_keys we can just shortcut and be ready
1589
 
            # for _flush_new_queue
1590
 
            self._new_queue = list(self._new_root_keys)
1591
 
            return
1592
 
        old_chks_to_enqueue = self._read_old_roots()
1593
 
        # filter out any root keys that are already known to be uninteresting
1594
 
        new_keys = set(self._new_root_keys).difference(self._all_old_chks)
1595
 
        # These are prefixes that are present in new_keys that we are
1596
 
        # thinking to yield
1597
 
        new_prefixes = set()
1598
 
        # We are about to yield all of these, so we don't want them getting
1599
 
        # added a second time
1600
 
        processed_new_refs = self._processed_new_refs
1601
 
        processed_new_refs.update(new_keys)
1602
 
        for record, node, prefix_refs, items in \
1603
 
                self._read_nodes_from_store(new_keys):
1604
 
            # At this level, we now know all the uninteresting references
1605
 
            # So we filter and queue up whatever is remaining
1606
 
            prefix_refs = [p_r for p_r in prefix_refs
1607
 
                           if p_r[1] not in self._all_old_chks
1608
 
                              and p_r[1] not in processed_new_refs]
1609
 
            refs = [p_r[1] for p_r in prefix_refs]
1610
 
            new_prefixes.update([p_r[0] for p_r in prefix_refs])
1611
 
            self._new_queue.extend(refs)
1612
 
            # TODO: We can potentially get multiple items here, however the
1613
 
            #       current design allows for this, as callers will do the work
1614
 
            #       to make the results unique. We might profile whether we
1615
 
            #       gain anything by ensuring unique return values for items
1616
 
            # TODO: This might be a good time to cast to StaticTuple, as
1617
 
            #       self._new_item_queue will hold the contents of multiple
1618
 
            #       records for an extended lifetime
1619
 
            new_items = [item for item in items
1620
 
                               if item not in self._all_old_items]
1621
 
            self._new_item_queue.extend(new_items)
1622
 
            new_prefixes.update([self._search_key_func(item[0])
1623
 
                                 for item in new_items])
1624
 
            processed_new_refs.update(refs)
1625
 
            yield record
1626
 
        # For new_prefixes we have the full length prefixes queued up.
1627
 
        # However, we also need possible prefixes. (If we have a known ref to
1628
 
        # 'ab', then we also need to include 'a'.) So expand the
1629
 
        # new_prefixes to include all shorter prefixes
1630
 
        for prefix in list(new_prefixes):
1631
 
            new_prefixes.update([prefix[:i] for i in xrange(1, len(prefix))])
1632
 
        self._enqueue_old(new_prefixes, old_chks_to_enqueue)
1633
 
 
1634
 
    def _flush_new_queue(self):
1635
 
        # No need to maintain the heap invariant anymore, just pull things out
1636
 
        # and process them
1637
 
        refs = set(self._new_queue)
1638
 
        self._new_queue = []
1639
 
        # First pass, flush all interesting items and convert to using direct refs
1640
 
        all_old_chks = self._all_old_chks
1641
 
        processed_new_refs = self._processed_new_refs
1642
 
        all_old_items = self._all_old_items
1643
 
        new_items = [item for item in self._new_item_queue
1644
 
                           if item not in all_old_items]
1645
 
        self._new_item_queue = []
1646
 
        if new_items:
1647
 
            yield None, new_items
1648
 
        refs = refs.difference(all_old_chks)
1649
 
        processed_new_refs.update(refs)
1650
 
        while refs:
1651
 
            # TODO: Using a SimpleSet for self._processed_new_refs and
1652
 
            #       saved as much as 10MB of peak memory. However, it requires
1653
 
            #       implementing a non-pyrex version.
1654
 
            next_refs = set()
1655
 
            next_refs_update = next_refs.update
1656
 
            # Inlining _read_nodes_from_store improves 'bzr branch bzr.dev'
1657
 
            # from 1m54s to 1m51s. Consider it.
1658
 
            for record, _, p_refs, items in self._read_nodes_from_store(refs):
1659
 
                if all_old_items:
1660
 
                    # using the 'if' check saves about 145s => 141s, when
1661
 
                    # streaming initial branch of Launchpad data.
1662
 
                    items = [item for item in items
1663
 
                             if item not in all_old_items]
1664
 
                yield record, items
1665
 
                next_refs_update([p_r[1] for p_r in p_refs])
1666
 
                del p_refs
1667
 
            # set1.difference(set/dict) walks all of set1, and checks if it
1668
 
            # exists in 'other'.
1669
 
            # set1.difference(iterable) walks all of iterable, and does a
1670
 
            # 'difference_update' on a clone of set1. Pick wisely based on the
1671
 
            # expected sizes of objects.
1672
 
            # in our case it is expected that 'new_refs' will always be quite
1673
 
            # small.
1674
 
            next_refs = next_refs.difference(all_old_chks)
1675
 
            next_refs = next_refs.difference(processed_new_refs)
1676
 
            processed_new_refs.update(next_refs)
1677
 
            refs = next_refs
1678
 
 
1679
 
    def _process_next_old(self):
1680
 
        # Since we don't filter uninteresting any further than during
1681
 
        # _read_all_roots, process the whole queue in a single pass.
1682
 
        refs = self._old_queue
1683
 
        self._old_queue = []
1684
 
        all_old_chks = self._all_old_chks
1685
 
        for record, _, prefix_refs, items in self._read_nodes_from_store(refs):
1686
 
            # TODO: Use StaticTuple here?
1687
 
            self._all_old_items.update(items)
1688
 
            refs = [r for _,r in prefix_refs if r not in all_old_chks]
1689
 
            self._old_queue.extend(refs)
1690
 
            all_old_chks.update(refs)
1691
 
 
1692
 
    def _process_queues(self):
1693
 
        while self._old_queue:
1694
 
            self._process_next_old()
1695
 
        return self._flush_new_queue()
1696
 
 
1697
 
    def process(self):
1698
 
        for record in self._read_all_roots():
1699
 
            yield record, []
1700
 
        for record, items in self._process_queues():
1701
 
            yield record, items
 
1438
                all_uninteresting_items.update(node._items.iteritems())
 
1439
        chks_to_read = next_chks
 
1440
    return (all_uninteresting_chks, all_uninteresting_items,
 
1441
            interesting_keys, interesting_to_yield, interesting_items)
1702
1442
 
1703
1443
 
1704
1444
def iter_interesting_nodes(store, interesting_root_keys,
1715
1455
    :return: Yield
1716
1456
        (interesting record, {interesting key:values})
1717
1457
    """
1718
 
    iterator = CHKMapDifference(store, interesting_root_keys,
1719
 
                                uninteresting_root_keys,
1720
 
                                search_key_func=store._search_key_func,
1721
 
                                pb=pb)
1722
 
    return iterator.process()
 
1458
    # TODO: consider that it may be more memory efficient to use the 20-byte
 
1459
    #       sha1 string, rather than tuples of hexidecimal sha1 strings.
 
1460
    # TODO: Try to factor out a lot of the get_record_stream() calls into a
 
1461
    #       helper function similar to _read_bytes. This function should be
 
1462
    #       able to use nodes from the _page_cache as well as actually
 
1463
    #       requesting bytes from the store.
 
1464
 
 
1465
    (all_uninteresting_chks, all_uninteresting_items, interesting_keys,
 
1466
     interesting_to_yield, interesting_items) = _find_all_uninteresting(store,
 
1467
        interesting_root_keys, uninteresting_root_keys, pb)
 
1468
 
 
1469
    # Now that we know everything uninteresting, we can yield information from
 
1470
    # our first request
 
1471
    interesting_items.difference_update(all_uninteresting_items)
 
1472
    interesting_to_yield = set(interesting_to_yield) - all_uninteresting_chks
 
1473
    if interesting_items:
 
1474
        yield None, interesting_items
 
1475
    if interesting_to_yield:
 
1476
        # We request these records again, rather than buffering the root
 
1477
        # records, most likely they are still in the _group_cache anyway.
 
1478
        for record in store.get_record_stream(interesting_to_yield,
 
1479
                                              'unordered', False):
 
1480
            yield record, []
 
1481
    all_uninteresting_chks.update(interesting_to_yield)
 
1482
    interesting_keys.difference_update(all_uninteresting_chks)
 
1483
 
 
1484
    chks_to_read = interesting_keys
 
1485
    counter = 0
 
1486
    while chks_to_read:
 
1487
        next_chks = set()
 
1488
        for record in store.get_record_stream(chks_to_read, 'unordered', False):
 
1489
            counter += 1
 
1490
            if pb is not None:
 
1491
                pb.update('find chk pages', counter)
 
1492
            # TODO: Handle 'absent'?
 
1493
            bytes = record.get_bytes_as('fulltext')
 
1494
            # We don't care about search_key_func for this code, because we
 
1495
            # only care about external references.
 
1496
            node = _deserialise(bytes, record.key, search_key_func=None)
 
1497
            if type(node) is InternalNode:
 
1498
                # all_uninteresting_chks grows large, as it lists all nodes we
 
1499
                # don't want to process (including already seen interesting
 
1500
                # nodes).
 
1501
                # small.difference_update(large) scales O(large), but
 
1502
                # small.difference(large) scales O(small).
 
1503
                # Also, we know we just _deserialised this node, so we can
 
1504
                # access the dict directly.
 
1505
                chks = set(node._items.itervalues()).difference(
 
1506
                            all_uninteresting_chks)
 
1507
                # Is set() and .difference_update better than:
 
1508
                # chks = [chk for chk in node.refs()
 
1509
                #              if chk not in all_uninteresting_chks]
 
1510
                next_chks.update(chks)
 
1511
                # These are now uninteresting everywhere else
 
1512
                all_uninteresting_chks.update(chks)
 
1513
                interesting_items = []
 
1514
            else:
 
1515
                interesting_items = [item for item in node._items.iteritems()
 
1516
                                     if item not in all_uninteresting_items]
 
1517
                # TODO: Do we need to filter out items that we have already
 
1518
                #       seen on other pages? We don't really want to buffer the
 
1519
                #       whole thing, but it does mean that callers need to
 
1520
                #       understand they may get duplicate values.
 
1521
                # all_uninteresting_items.update(interesting_items)
 
1522
            yield record, interesting_items
 
1523
        chks_to_read = next_chks
1723
1524
 
1724
1525
 
1725
1526
try:
1726
1527
    from bzrlib._chk_map_pyx import (
1727
 
        _bytes_to_text_key,
1728
1528
        _search_key_16,
1729
1529
        _search_key_255,
1730
1530
        _deserialise_leaf_node,
1731
1531
        _deserialise_internal_node,
1732
1532
        )
1733
 
except ImportError, e:
1734
 
    osutils.failed_to_load_extension(e)
 
1533
except ImportError:
1735
1534
    from bzrlib._chk_map_py import (
1736
 
        _bytes_to_text_key,
1737
1535
        _search_key_16,
1738
1536
        _search_key_255,
1739
1537
        _deserialise_leaf_node,
1741
1539
        )
1742
1540
search_key_registry.register('hash-16-way', _search_key_16)
1743
1541
search_key_registry.register('hash-255-way', _search_key_255)
1744
 
 
1745
 
 
1746
 
def _check_key(key):
1747
 
    """Helper function to assert that a key is properly formatted.
1748
 
 
1749
 
    This generally shouldn't be used in production code, but it can be helpful
1750
 
    to debug problems.
1751
 
    """
1752
 
    if type(key) is not StaticTuple:
1753
 
        raise TypeError('key %r is not StaticTuple but %s' % (key, type(key)))
1754
 
    if len(key) != 1:
1755
 
        raise ValueError('key %r should have length 1, not %d' % (key, len(key),))
1756
 
    if type(key[0]) is not str:
1757
 
        raise TypeError('key %r should hold a str, not %r'
1758
 
                        % (key, type(key[0])))
1759
 
    if not key[0].startswith('sha1:'):
1760
 
        raise ValueError('key %r should point to a sha1:' % (key,))
1761
 
 
1762