← Back to branch summary

~bzr-pqm/bzr/bzr.dev

« back to all changes in this revision

Viewing changes to bzrlib/chk_map.py

  • Committer: Tarmac
  • Author(s): Vincent Ladeuil
  • Date: 2017-01-30 14:42:05 UTC
  • mfrom: (6620.1.1 trunk)
  • Revision ID: tarmac-20170130144205-r8fh2xpmiuxyozpv
Merge  2.7 into trunk including fix for bug #1657238 [r=vila]

Show diffs side-by-side

added added

removed removed

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