~bzr-pqm/bzr/bzr.dev

« back to all changes in this revision

Viewing changes to bzrlib/chk_map.py

  • Committer: Jonathan Lange
  • Date: 2009-06-10 08:05:05 UTC
  • mto: This revision was merged to the branch mainline in revision 4433.
  • Revision ID: jml@canonical.com-20090610080505-lmch4q01z1q7deu2
MoreĀ unusedĀ imports.

Show diffs side-by-side

added added

removed removed

Lines of Context:
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) == 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) == 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)
548
465
 
549
466
    def _node_key(self, node):
550
467
        """Get the key for a node whether it's a tuple or node."""
551
 
        if type(node) is tuple:
552
 
            node = StaticTuple.from_sequence(node)
553
 
        if type(node) is StaticTuple:
 
468
        if type(node) == tuple:
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) == 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
 
1064
955
        # prefix is the key in self._items to use, key_filter is the key_filter
1065
956
        # entries that would match this node
1066
957
        keys = {}
1067
 
        shortcut = False
1068
958
        if key_filter is None:
1069
 
            # yielding all nodes, yield whatever we have, and queue up a read
1070
 
            # for whatever we are missing
1071
 
            shortcut = True
1072
959
            for prefix, node in self._items.iteritems():
1073
 
                if node.__class__ is StaticTuple:
 
960
                if type(node) == tuple:
1074
961
                    keys[node] = (prefix, None)
1075
962
                else:
1076
963
                    yield node, None
1077
 
        elif len(key_filter) == 1:
1078
 
            # Technically, this path could also be handled by the first check
1079
 
            # in 'self._node_width' in length_filters. However, we can handle
1080
 
            # this case without spending any time building up the
1081
 
            # prefix_to_keys, etc state.
1082
 
 
1083
 
            # This is a bit ugly, but TIMEIT showed it to be by far the fastest
1084
 
            # 0.626us   list(key_filter)[0]
1085
 
            #       is a func() for list(), 2 mallocs, and a getitem
1086
 
            # 0.489us   [k for k in key_filter][0]
1087
 
            #       still has the mallocs, avoids the func() call
1088
 
            # 0.350us   iter(key_filter).next()
1089
 
            #       has a func() call, and mallocs an iterator
1090
 
            # 0.125us   for key in key_filter: pass
1091
 
            #       no func() overhead, might malloc an iterator
1092
 
            # 0.105us   for key in key_filter: break
1093
 
            #       no func() overhead, might malloc an iterator, probably
1094
 
            #       avoids checking an 'else' clause as part of the for
1095
 
            for key in key_filter:
1096
 
                break
1097
 
            search_prefix = self._search_prefix_filter(key)
1098
 
            if len(search_prefix) == self._node_width:
1099
 
                # This item will match exactly, so just do a dict lookup, and
1100
 
                # see what we can return
1101
 
                shortcut = True
1102
 
                try:
1103
 
                    node = self._items[search_prefix]
1104
 
                except KeyError:
1105
 
                    # A given key can only match 1 child node, if it isn't
1106
 
                    # there, then we can just return nothing
1107
 
                    return
1108
 
                if node.__class__ is StaticTuple:
1109
 
                    keys[node] = (search_prefix, [key])
1110
 
                else:
1111
 
                    # This is loaded, and the only thing that can match,
1112
 
                    # return
1113
 
                    yield node, [key]
1114
 
                    return
1115
 
        if not shortcut:
1116
 
            # First, convert all keys into a list of search prefixes
1117
 
            # Aggregate common prefixes, and track the keys they come from
 
964
        else:
 
965
            # XXX defaultdict ?
1118
966
            prefix_to_keys = {}
1119
967
            length_filters = {}
1120
968
            for key in key_filter:
1121
 
                search_prefix = self._search_prefix_filter(key)
 
969
                search_key = self._search_prefix_filter(key)
1122
970
                length_filter = length_filters.setdefault(
1123
 
                                    len(search_prefix), set())
1124
 
                length_filter.add(search_prefix)
1125
 
                prefix_to_keys.setdefault(search_prefix, []).append(key)
1126
 
 
1127
 
            if (self._node_width in length_filters
1128
 
                and len(length_filters) == 1):
1129
 
                # all of the search prefixes match exactly _node_width. This
1130
 
                # means that everything is an exact match, and we can do a
1131
 
                # lookup into self._items, rather than iterating over the items
1132
 
                # dict.
1133
 
                search_prefixes = length_filters[self._node_width]
1134
 
                for search_prefix in search_prefixes:
1135
 
                    try:
1136
 
                        node = self._items[search_prefix]
1137
 
                    except KeyError:
1138
 
                        # We can ignore this one
1139
 
                        continue
1140
 
                    node_key_filter = prefix_to_keys[search_prefix]
1141
 
                    if node.__class__ is StaticTuple:
1142
 
                        keys[node] = (search_prefix, node_key_filter)
 
971
                                    len(search_key), set())
 
972
                length_filter.add(search_key)
 
973
                prefix_to_keys.setdefault(search_key, []).append(key)
 
974
            length_filters = length_filters.items()
 
975
            for prefix, node in self._items.iteritems():
 
976
                node_key_filter = []
 
977
                for length, length_filter in length_filters:
 
978
                    sub_prefix = prefix[:length]
 
979
                    if sub_prefix in length_filter:
 
980
                        node_key_filter.extend(prefix_to_keys[sub_prefix])
 
981
                if node_key_filter: # this key matched something, yield it
 
982
                    if type(node) == tuple:
 
983
                        keys[node] = (prefix, node_key_filter)
1143
984
                    else:
1144
985
                        yield node, node_key_filter
1145
 
            else:
1146
 
                # The slow way. We walk every item in self._items, and check to
1147
 
                # see if there are any matches
1148
 
                length_filters = length_filters.items()
1149
 
                for prefix, node in self._items.iteritems():
1150
 
                    node_key_filter = []
1151
 
                    for length, length_filter in length_filters:
1152
 
                        sub_prefix = prefix[:length]
1153
 
                        if sub_prefix in length_filter:
1154
 
                            node_key_filter.extend(prefix_to_keys[sub_prefix])
1155
 
                    if node_key_filter: # this key matched something, yield it
1156
 
                        if node.__class__ is StaticTuple:
1157
 
                            keys[node] = (prefix, node_key_filter)
1158
 
                        else:
1159
 
                            yield node, node_key_filter
1160
986
        if keys:
1161
987
            # Look in the page cache for some more bytes
1162
988
            found_keys = set()
1163
989
            for key in keys:
1164
990
                try:
1165
 
                    bytes = _get_cache()[key]
 
991
                    bytes = _page_cache[key]
1166
992
                except KeyError:
1167
993
                    continue
1168
994
                else:
1193
1019
                    prefix, node_key_filter = keys[record.key]
1194
1020
                    node_and_filters.append((node, node_key_filter))
1195
1021
                    self._items[prefix] = node
1196
 
                    _get_cache().add(record.key, bytes)
 
1022
                    _page_cache.add(record.key, bytes)
1197
1023
                for info in node_and_filters:
1198
1024
                    yield info
1199
1025
 
1291
1117
        :return: An iterable of the keys inserted by this operation.
1292
1118
        """
1293
1119
        for node in self._items.itervalues():
1294
 
            if type(node) is StaticTuple:
 
1120
            if type(node) == tuple:
1295
1121
                # Never deserialised.
1296
1122
                continue
1297
1123
            if node._key is not None:
1308
1134
        lines.append('%s\n' % (self._search_prefix,))
1309
1135
        prefix_len = len(self._search_prefix)
1310
1136
        for prefix, node in sorted(self._items.items()):
1311
 
            if type(node) is StaticTuple:
 
1137
            if type(node) == tuple:
1312
1138
                key = node[0]
1313
1139
            else:
1314
1140
                key = node._key[0]
1318
1144
                    % (serialised, self._search_prefix))
1319
1145
            lines.append(serialised[prefix_len:])
1320
1146
        sha1, _, _ = store.add_lines((None,), (), lines)
1321
 
        self._key = StaticTuple("sha1:" + sha1,).intern()
1322
 
        _get_cache().add(self._key, ''.join(lines))
 
1147
        self._key = ("sha1:" + sha1,)
 
1148
        _page_cache.add(self._key, ''.join(lines))
1323
1149
        yield self._key
1324
1150
 
1325
1151
    def _search_key(self, key):
1353
1179
            raise AssertionError("unserialised nodes have no refs.")
1354
1180
        refs = []
1355
1181
        for value in self._items.itervalues():
1356
 
            if type(value) is StaticTuple:
 
1182
            if type(value) == tuple:
1357
1183
                refs.append(value)
1358
1184
            else:
1359
1185
                refs.append(value.key())
1369
1195
        return self._search_prefix
1370
1196
 
1371
1197
    def unmap(self, store, key, check_remap=True):
1372
 
        """Remove key from this node and its children."""
 
1198
        """Remove key from this node and it's children."""
1373
1199
        if not len(self._items):
1374
1200
            raise AssertionError("can't unmap in an empty InternalNode.")
1375
1201
        children = [node for node, _
1460
1286
    return node
1461
1287
 
1462
1288
 
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)
 
1289
def _find_children_info(store, interesting_keys, uninteresting_keys, pb):
 
1290
    """Read the associated records, and determine what is interesting."""
 
1291
    uninteresting_keys = set(uninteresting_keys)
 
1292
    chks_to_read = uninteresting_keys.union(interesting_keys)
 
1293
    next_uninteresting = set()
 
1294
    next_interesting = set()
 
1295
    uninteresting_items = set()
 
1296
    interesting_items = set()
 
1297
    interesting_to_yield = []
 
1298
    for record in store.get_record_stream(chks_to_read, 'unordered', True):
 
1299
        # records_read.add(record.key())
 
1300
        if pb is not None:
 
1301
            pb.tick()
 
1302
        bytes = record.get_bytes_as('fulltext')
 
1303
        # We don't care about search_key_func for this code, because we only
 
1304
        # care about external references.
 
1305
        node = _deserialise(bytes, record.key, search_key_func=None)
 
1306
        if record.key in uninteresting_keys:
 
1307
            if type(node) is InternalNode:
 
1308
                next_uninteresting.update(node.refs())
 
1309
            else:
 
1310
                # We know we are at a LeafNode, so we can pass None for the
 
1311
                # store
 
1312
                uninteresting_items.update(node.iteritems(None))
 
1313
        else:
 
1314
            interesting_to_yield.append(record.key)
 
1315
            if type(node) is InternalNode:
 
1316
                next_interesting.update(node.refs())
 
1317
            else:
 
1318
                interesting_items.update(node.iteritems(None))
 
1319
    return (next_uninteresting, uninteresting_items,
 
1320
            next_interesting, interesting_to_yield, interesting_items)
 
1321
 
 
1322
 
 
1323
def _find_all_uninteresting(store, interesting_root_keys,
 
1324
                            uninteresting_root_keys, pb):
 
1325
    """Determine the full set of uninteresting keys."""
 
1326
    # What about duplicates between interesting_root_keys and
 
1327
    # uninteresting_root_keys?
 
1328
    if not uninteresting_root_keys:
 
1329
        # Shortcut case. We know there is nothing uninteresting to filter out
 
1330
        # So we just let the rest of the algorithm do the work
 
1331
        # We know there is nothing uninteresting, and we didn't have to read
 
1332
        # any interesting records yet.
 
1333
        return (set(), set(), set(interesting_root_keys), [], set())
 
1334
    all_uninteresting_chks = set(uninteresting_root_keys)
 
1335
    all_uninteresting_items = set()
 
1336
 
 
1337
    # First step, find the direct children of both the interesting and
 
1338
    # uninteresting set
 
1339
    (uninteresting_keys, uninteresting_items,
 
1340
     interesting_keys, interesting_to_yield,
 
1341
     interesting_items) = _find_children_info(store, interesting_root_keys,
 
1342
                                              uninteresting_root_keys,
 
1343
                                              pb=pb)
 
1344
    all_uninteresting_chks.update(uninteresting_keys)
 
1345
    all_uninteresting_items.update(uninteresting_items)
 
1346
    del uninteresting_items
 
1347
    # Note: Exact matches between interesting and uninteresting do not need
 
1348
    #       to be search further. Non-exact matches need to be searched in case
 
1349
    #       there is a future exact-match
 
1350
    uninteresting_keys.difference_update(interesting_keys)
 
1351
 
 
1352
    # Second, find the full set of uninteresting bits reachable by the
 
1353
    # uninteresting roots
 
1354
    chks_to_read = uninteresting_keys
 
1355
    while chks_to_read:
 
1356
        next_chks = set()
 
1357
        for record in store.get_record_stream(chks_to_read, 'unordered', False):
 
1358
            # TODO: Handle 'absent'
 
1359
            if pb is not None:
 
1360
                pb.tick()
1523
1361
            bytes = record.get_bytes_as('fulltext')
1524
 
            node = _deserialise(bytes, record.key,
1525
 
                                search_key_func=self._search_key_func)
 
1362
            # We don't care about search_key_func for this code, because we
 
1363
            # only care about external references.
 
1364
            node = _deserialise(bytes, record.key, search_key_func=None)
1526
1365
            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 = []
 
1366
                # uninteresting_prefix_chks.update(node._items.iteritems())
 
1367
                chks = node._items.values()
 
1368
                # TODO: We remove the entries that are already in
 
1369
                #       uninteresting_chks ?
 
1370
                next_chks.update(chks)
 
1371
                all_uninteresting_chks.update(chks)
1536
1372
            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
 
1373
                all_uninteresting_items.update(node._items.iteritems())
 
1374
        chks_to_read = next_chks
 
1375
    return (all_uninteresting_chks, all_uninteresting_items,
 
1376
            interesting_keys, interesting_to_yield, interesting_items)
1702
1377
 
1703
1378
 
1704
1379
def iter_interesting_nodes(store, interesting_root_keys,
1715
1390
    :return: Yield
1716
1391
        (interesting record, {interesting key:values})
1717
1392
    """
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()
 
1393
    # TODO: consider that it may be more memory efficient to use the 20-byte
 
1394
    #       sha1 string, rather than tuples of hexidecimal sha1 strings.
 
1395
    # TODO: Try to factor out a lot of the get_record_stream() calls into a
 
1396
    #       helper function similar to _read_bytes. This function should be
 
1397
    #       able to use nodes from the _page_cache as well as actually
 
1398
    #       requesting bytes from the store.
 
1399
 
 
1400
    (all_uninteresting_chks, all_uninteresting_items, interesting_keys,
 
1401
     interesting_to_yield, interesting_items) = _find_all_uninteresting(store,
 
1402
        interesting_root_keys, uninteresting_root_keys, pb)
 
1403
 
 
1404
    # Now that we know everything uninteresting, we can yield information from
 
1405
    # our first request
 
1406
    interesting_items.difference_update(all_uninteresting_items)
 
1407
    interesting_to_yield = set(interesting_to_yield) - all_uninteresting_chks
 
1408
    if interesting_items:
 
1409
        yield None, interesting_items
 
1410
    if interesting_to_yield:
 
1411
        # We request these records again, rather than buffering the root
 
1412
        # records, most likely they are still in the _group_cache anyway.
 
1413
        for record in store.get_record_stream(interesting_to_yield,
 
1414
                                              'unordered', False):
 
1415
            yield record, []
 
1416
    all_uninteresting_chks.update(interesting_to_yield)
 
1417
    interesting_keys.difference_update(all_uninteresting_chks)
 
1418
 
 
1419
    chks_to_read = interesting_keys
 
1420
    counter = 0
 
1421
    while chks_to_read:
 
1422
        next_chks = set()
 
1423
        for record in store.get_record_stream(chks_to_read, 'unordered', False):
 
1424
            counter += 1
 
1425
            if pb is not None:
 
1426
                pb.update('find chk pages', counter)
 
1427
            # TODO: Handle 'absent'?
 
1428
            bytes = record.get_bytes_as('fulltext')
 
1429
            # We don't care about search_key_func for this code, because we
 
1430
            # only care about external references.
 
1431
            node = _deserialise(bytes, record.key, search_key_func=None)
 
1432
            if type(node) is InternalNode:
 
1433
                # all_uninteresting_chks grows large, as it lists all nodes we
 
1434
                # don't want to process (including already seen interesting
 
1435
                # nodes).
 
1436
                # small.difference_update(large) scales O(large), but
 
1437
                # small.difference(large) scales O(small).
 
1438
                # Also, we know we just _deserialised this node, so we can
 
1439
                # access the dict directly.
 
1440
                chks = set(node._items.itervalues()).difference(
 
1441
                            all_uninteresting_chks)
 
1442
                # Is set() and .difference_update better than:
 
1443
                # chks = [chk for chk in node.refs()
 
1444
                #              if chk not in all_uninteresting_chks]
 
1445
                next_chks.update(chks)
 
1446
                # These are now uninteresting everywhere else
 
1447
                all_uninteresting_chks.update(chks)
 
1448
                interesting_items = []
 
1449
            else:
 
1450
                interesting_items = [item for item in node._items.iteritems()
 
1451
                                     if item not in all_uninteresting_items]
 
1452
                # TODO: Do we need to filter out items that we have already
 
1453
                #       seen on other pages? We don't really want to buffer the
 
1454
                #       whole thing, but it does mean that callers need to
 
1455
                #       understand they may get duplicate values.
 
1456
                # all_uninteresting_items.update(interesting_items)
 
1457
            yield record, interesting_items
 
1458
        chks_to_read = next_chks
1723
1459
 
1724
1460
 
1725
1461
try:
1726
1462
    from bzrlib._chk_map_pyx import (
1727
 
        _bytes_to_text_key,
1728
1463
        _search_key_16,
1729
1464
        _search_key_255,
1730
1465
        _deserialise_leaf_node,
1731
1466
        _deserialise_internal_node,
1732
1467
        )
1733
 
except ImportError, e:
1734
 
    osutils.failed_to_load_extension(e)
 
1468
except ImportError:
1735
1469
    from bzrlib._chk_map_py import (
1736
 
        _bytes_to_text_key,
1737
1470
        _search_key_16,
1738
1471
        _search_key_255,
1739
1472
        _deserialise_leaf_node,
1741
1474
        )
1742
1475
search_key_registry.register('hash-16-way', _search_key_16)
1743
1476
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