179
186
combine mem with the first and second indexes, creating a new one of
180
187
size 4x. On the fifth create a single new one, etc.
189
if self._combine_backing_indices:
190
(new_backing_file, size,
191
backing_pos) = self._spill_mem_keys_and_combine()
193
new_backing_file, size = self._spill_mem_keys_without_combining()
194
# Note: The transport here isn't strictly needed, because we will use
195
# direct access to the new_backing._file object
196
new_backing = BTreeGraphIndex(get_transport('.'), '<temp>', size)
197
# GC will clean up the file
198
new_backing._file = new_backing_file
199
if self._combine_backing_indices:
200
if len(self._backing_indices) == backing_pos:
201
self._backing_indices.append(None)
202
self._backing_indices[backing_pos] = new_backing
203
for backing_pos in range(backing_pos):
204
self._backing_indices[backing_pos] = None
206
self._backing_indices.append(new_backing)
208
self._nodes_by_key = None
210
def _spill_mem_keys_without_combining(self):
211
return self._write_nodes(self._iter_mem_nodes(), allow_optimize=False)
213
def _spill_mem_keys_and_combine(self):
182
214
iterators_to_combine = [self._iter_mem_nodes()]
184
216
for pos, backing in enumerate(self._backing_indices):
340
372
self.row_lengths = []
341
373
# Loop over all nodes adding them to the bottom row
342
374
# (rows[-1]). When we finish a chunk in a row,
343
# propogate the key that didn't fit (comes after the chunk) to the
375
# propagate the key that didn't fit (comes after the chunk) to the
344
376
# row above, transitively.
345
377
for node in node_iterator:
346
378
if key_count == 0:
347
379
# First key triggers the first row
348
380
rows.append(_LeafBuilderRow())
350
# TODO: Flattening the node into a string key and a line should
351
# probably be put into a pyrex function. We can do a quick
352
# iter over all the entries to determine the final length,
353
# and then do a single malloc() rather than lots of
354
# intermediate mallocs as we build everything up.
355
# ATM 3 / 13s are spent flattening nodes (10s is compressing)
356
382
string_key, line = _btree_serializer._flatten_node(node,
357
383
self.reference_lists)
358
self._add_key(string_key, line, rows)
384
self._add_key(string_key, line, rows, allow_optimize=allow_optimize)
359
385
for row in reversed(rows):
360
386
pad = (type(row) != _LeafBuilderRow)
361
387
row.finish_node(pad=pad)
362
result = tempfile.NamedTemporaryFile()
363
388
lines = [_BTSIGNATURE]
364
389
lines.append(_OPTION_NODE_REFS + str(self.reference_lists) + '\n')
365
390
lines.append(_OPTION_KEY_ELEMENTS + str(self._key_length) + '\n')
366
391
lines.append(_OPTION_LEN + str(key_count) + '\n')
367
392
row_lengths = [row.nodes for row in rows]
368
393
lines.append(_OPTION_ROW_LENGTHS + ','.join(map(str, row_lengths)) + '\n')
394
if row_lengths and row_lengths[-1] > 1:
395
result = tempfile.NamedTemporaryFile(prefix='bzr-index-')
397
result = cStringIO.StringIO()
369
398
result.writelines(lines)
370
399
position = sum(map(len, lines))
602
658
the initial read (to read the root node header) can be done
603
659
without over-reading even on empty indices, and on small indices
604
660
allows single-IO to read the entire index.
661
:param unlimited_cache: If set to True, then instead of using an
662
LRUCache with size _NODE_CACHE_SIZE, we will use a dict and always
663
cache all leaf nodes.
664
:param offset: The start of the btree index data isn't byte 0 of the
665
file. Instead it starts at some point later.
606
667
self._transport = transport
607
668
self._name = name
608
669
self._size = size
609
670
self._file = None
610
self._page_size = transport.recommended_page_size()
671
self._recommended_pages = self._compute_recommended_pages()
611
672
self._root_node = None
673
self._base_offset = offset
612
674
# Default max size is 100,000 leave values
613
675
self._leaf_value_cache = None # lru_cache.LRUCache(100*1000)
614
self._leaf_node_cache = lru_cache.LRUCache(_NODE_CACHE_SIZE)
615
self._internal_node_cache = lru_cache.LRUCache()
677
self._leaf_node_cache = {}
678
self._internal_node_cache = {}
680
self._leaf_node_cache = lru_cache.LRUCache(_NODE_CACHE_SIZE)
681
# We use a FIFO here just to prevent possible blowout. However, a
682
# 300k record btree has only 3k leaf nodes, and only 20 internal
683
# nodes. A value of 100 scales to ~100*100*100 = 1M records.
684
self._internal_node_cache = fifo_cache.FIFOCache(100)
616
685
self._key_count = None
617
686
self._row_lengths = None
618
687
self._row_offsets = None # Start of each row, [-1] is the end
650
711
:return: A dict of {node_pos: node}
652
if len(nodes) > cache._max_cache:
653
trace.mutter('Requesting %s > %s nodes, not all will be cached',
654
len(nodes), cache._max_cache)
714
start_of_leaves = None
656
715
for node_pos, node in self._read_nodes(sorted(nodes)):
657
716
if node_pos == 0: # Special case
658
717
self._root_node = node
660
cache.add(node_pos, node)
719
if start_of_leaves is None:
720
start_of_leaves = self._row_offsets[-2]
721
if node_pos < start_of_leaves:
722
self._internal_node_cache[node_pos] = node
724
self._leaf_node_cache[node_pos] = node
661
725
found[node_pos] = node
728
def _compute_recommended_pages(self):
729
"""Convert transport's recommended_page_size into btree pages.
731
recommended_page_size is in bytes, we want to know how many _PAGE_SIZE
732
pages fit in that length.
734
recommended_read = self._transport.recommended_page_size()
735
recommended_pages = int(math.ceil(recommended_read /
737
return recommended_pages
739
def _compute_total_pages_in_index(self):
740
"""How many pages are in the index.
742
If we have read the header we will use the value stored there.
743
Otherwise it will be computed based on the length of the index.
745
if self._size is None:
746
raise AssertionError('_compute_total_pages_in_index should not be'
747
' called when self._size is None')
748
if self._root_node is not None:
749
# This is the number of pages as defined by the header
750
return self._row_offsets[-1]
751
# This is the number of pages as defined by the size of the index. They
752
# should be indentical.
753
total_pages = int(math.ceil(self._size / float(_PAGE_SIZE)))
756
def _expand_offsets(self, offsets):
757
"""Find extra pages to download.
759
The idea is that we always want to make big-enough requests (like 64kB
760
for http), so that we don't waste round trips. So given the entries
761
that we already have cached and the new pages being downloaded figure
762
out what other pages we might want to read.
764
See also doc/developers/btree_index_prefetch.txt for more details.
766
:param offsets: The offsets to be read
767
:return: A list of offsets to download
769
if 'index' in debug.debug_flags:
770
trace.mutter('expanding: %s\toffsets: %s', self._name, offsets)
772
if len(offsets) >= self._recommended_pages:
773
# Don't add more, we are already requesting more than enough
774
if 'index' in debug.debug_flags:
775
trace.mutter(' not expanding large request (%s >= %s)',
776
len(offsets), self._recommended_pages)
778
if self._size is None:
779
# Don't try anything, because we don't know where the file ends
780
if 'index' in debug.debug_flags:
781
trace.mutter(' not expanding without knowing index size')
783
total_pages = self._compute_total_pages_in_index()
784
cached_offsets = self._get_offsets_to_cached_pages()
785
# If reading recommended_pages would read the rest of the index, just
787
if total_pages - len(cached_offsets) <= self._recommended_pages:
788
# Read whatever is left
790
expanded = [x for x in xrange(total_pages)
791
if x not in cached_offsets]
793
expanded = range(total_pages)
794
if 'index' in debug.debug_flags:
795
trace.mutter(' reading all unread pages: %s', expanded)
798
if self._root_node is None:
799
# ATM on the first read of the root node of a large index, we don't
800
# bother pre-reading any other pages. This is because the
801
# likelyhood of actually reading interesting pages is very low.
802
# See doc/developers/btree_index_prefetch.txt for a discussion, and
803
# a possible implementation when we are guessing that the second
804
# layer index is small
805
final_offsets = offsets
807
tree_depth = len(self._row_lengths)
808
if len(cached_offsets) < tree_depth and len(offsets) == 1:
809
# We haven't read enough to justify expansion
810
# If we are only going to read the root node, and 1 leaf node,
811
# then it isn't worth expanding our request. Once we've read at
812
# least 2 nodes, then we are probably doing a search, and we
813
# start expanding our requests.
814
if 'index' in debug.debug_flags:
815
trace.mutter(' not expanding on first reads')
817
final_offsets = self._expand_to_neighbors(offsets, cached_offsets,
820
final_offsets = sorted(final_offsets)
821
if 'index' in debug.debug_flags:
822
trace.mutter('expanded: %s', final_offsets)
825
def _expand_to_neighbors(self, offsets, cached_offsets, total_pages):
826
"""Expand requests to neighbors until we have enough pages.
828
This is called from _expand_offsets after policy has determined that we
830
We only want to expand requests within a given layer. We cheat a little
831
bit and assume all requests will be in the same layer. This is true
832
given the current design, but if it changes this algorithm may perform
835
:param offsets: requested offsets
836
:param cached_offsets: offsets for pages we currently have cached
837
:return: A set() of offsets after expansion
839
final_offsets = set(offsets)
841
new_tips = set(final_offsets)
842
while len(final_offsets) < self._recommended_pages and new_tips:
846
first, end = self._find_layer_first_and_end(pos)
849
and previous not in cached_offsets
850
and previous not in final_offsets
851
and previous >= first):
852
next_tips.add(previous)
854
if (after < total_pages
855
and after not in cached_offsets
856
and after not in final_offsets
859
# This would keep us from going bigger than
860
# recommended_pages by only expanding the first offsets.
861
# However, if we are making a 'wide' request, it is
862
# reasonable to expand all points equally.
863
# if len(final_offsets) > recommended_pages:
865
final_offsets.update(next_tips)
869
def clear_cache(self):
870
"""Clear out any cached/memoized values.
872
This can be called at any time, but generally it is used when we have
873
extracted some information, but don't expect to be requesting any more
876
# Note that we don't touch self._root_node or self._internal_node_cache
877
# We don't expect either of those to be big, and it can save
878
# round-trips in the future. We may re-evaluate this if InternalNode
879
# memory starts to be an issue.
880
self._leaf_node_cache.clear()
882
def external_references(self, ref_list_num):
883
if self._root_node is None:
884
self._get_root_node()
885
if ref_list_num + 1 > self.node_ref_lists:
886
raise ValueError('No ref list %d, index has %d ref lists'
887
% (ref_list_num, self.node_ref_lists))
890
for node in self.iter_all_entries():
892
refs.update(node[3][ref_list_num])
895
def _find_layer_first_and_end(self, offset):
896
"""Find the start/stop nodes for the layer corresponding to offset.
898
:return: (first, end)
899
first is the first node in this layer
900
end is the first node of the next layer
903
for roffset in self._row_offsets:
910
def _get_offsets_to_cached_pages(self):
911
"""Determine what nodes we already have cached."""
912
cached_offsets = set(self._internal_node_cache.keys())
913
cached_offsets.update(self._leaf_node_cache.keys())
914
if self._root_node is not None:
915
cached_offsets.add(0)
916
return cached_offsets
918
def _get_root_node(self):
919
if self._root_node is None:
920
# We may not have a root node yet
921
self._get_internal_nodes([0])
922
return self._root_node
664
924
def _get_nodes(self, cache, node_indexes):
709
976
"iter_all_entries scales with size of history.")
710
977
if not self.key_count():
979
if self._row_offsets[-1] == 1:
980
# There is only the root node, and we read that via key_count()
981
if self.node_ref_lists:
982
for key, (value, refs) in sorted(self._root_node.keys.items()):
983
yield (self, key, value, refs)
985
for key, (value, refs) in sorted(self._root_node.keys.items()):
986
yield (self, key, value)
712
988
start_of_leaves = self._row_offsets[-2]
713
989
end_of_leaves = self._row_offsets[-1]
714
needed_nodes = range(start_of_leaves, end_of_leaves)
990
needed_offsets = range(start_of_leaves, end_of_leaves)
991
if needed_offsets == [0]:
992
# Special case when we only have a root node, as we have already
994
nodes = [(0, self._root_node)]
996
nodes = self._read_nodes(needed_offsets)
715
997
# We iterate strictly in-order so that we can use this function
716
998
# for spilling index builds to disk.
717
999
if self.node_ref_lists:
718
for _, node in self._read_nodes(needed_nodes):
1000
for _, node in nodes:
719
1001
for key, (value, refs) in sorted(node.keys.items()):
720
1002
yield (self, key, value, refs)
722
for _, node in self._read_nodes(needed_nodes):
1004
for _, node in nodes:
723
1005
for key, (value, refs) in sorted(node.keys.items()):
724
1006
yield (self, key, value)
805
1087
output.append(cur_out)
1090
def _walk_through_internal_nodes(self, keys):
1091
"""Take the given set of keys, and find the corresponding LeafNodes.
1093
:param keys: An unsorted iterable of keys to search for
1094
:return: (nodes, index_and_keys)
1095
nodes is a dict mapping {index: LeafNode}
1096
keys_at_index is a list of tuples of [(index, [keys for Leaf])]
1098
# 6 seconds spent in miss_torture using the sorted() line.
1099
# Even with out of order disk IO it seems faster not to sort it when
1100
# large queries are being made.
1101
keys_at_index = [(0, sorted(keys))]
1103
for row_pos, next_row_start in enumerate(self._row_offsets[1:-1]):
1104
node_indexes = [idx for idx, s_keys in keys_at_index]
1105
nodes = self._get_internal_nodes(node_indexes)
1107
next_nodes_and_keys = []
1108
for node_index, sub_keys in keys_at_index:
1109
node = nodes[node_index]
1110
positions = self._multi_bisect_right(sub_keys, node.keys)
1111
node_offset = next_row_start + node.offset
1112
next_nodes_and_keys.extend([(node_offset + pos, s_keys)
1113
for pos, s_keys in positions])
1114
keys_at_index = next_nodes_and_keys
1115
# We should now be at the _LeafNodes
1116
node_indexes = [idx for idx, s_keys in keys_at_index]
1118
# TODO: We may *not* want to always read all the nodes in one
1119
# big go. Consider setting a max size on this.
1120
nodes = self._get_leaf_nodes(node_indexes)
1121
return nodes, keys_at_index
808
1123
def iter_entries(self, keys):
809
1124
"""Iterate over keys within the index.
887
1177
yield (self, next_sub_key, value)
1179
def _find_ancestors(self, keys, ref_list_num, parent_map, missing_keys):
1180
"""Find the parent_map information for the set of keys.
1182
This populates the parent_map dict and missing_keys set based on the
1183
queried keys. It also can fill out an arbitrary number of parents that
1184
it finds while searching for the supplied keys.
1186
It is unlikely that you want to call this directly. See
1187
"CombinedGraphIndex.find_ancestry()" for a more appropriate API.
1189
:param keys: A keys whose ancestry we want to return
1190
Every key will either end up in 'parent_map' or 'missing_keys'.
1191
:param ref_list_num: This index in the ref_lists is the parents we
1193
:param parent_map: {key: parent_keys} for keys that are present in this
1194
index. This may contain more entries than were in 'keys', that are
1195
reachable ancestors of the keys requested.
1196
:param missing_keys: keys which are known to be missing in this index.
1197
This may include parents that were not directly requested, but we
1198
were able to determine that they are not present in this index.
1199
:return: search_keys parents that were found but not queried to know
1200
if they are missing or present. Callers can re-query this index for
1201
those keys, and they will be placed into parent_map or missing_keys
1203
if not self.key_count():
1204
# We use key_count() to trigger reading the root node and
1205
# determining info about this BTreeGraphIndex
1206
# If we don't have any keys, then everything is missing
1207
missing_keys.update(keys)
1209
if ref_list_num >= self.node_ref_lists:
1210
raise ValueError('No ref list %d, index has %d ref lists'
1211
% (ref_list_num, self.node_ref_lists))
1213
# The main trick we are trying to accomplish is that when we find a
1214
# key listing its parents, we expect that the parent key is also likely
1215
# to sit on the same page. Allowing us to expand parents quickly
1216
# without suffering the full stack of bisecting, etc.
1217
nodes, nodes_and_keys = self._walk_through_internal_nodes(keys)
1219
# These are parent keys which could not be immediately resolved on the
1220
# page where the child was present. Note that we may already be
1221
# searching for that key, and it may actually be present [or known
1222
# missing] on one of the other pages we are reading.
1224
# We could try searching for them in the immediate previous or next
1225
# page. If they occur "later" we could put them in a pending lookup
1226
# set, and then for each node we read thereafter we could check to
1227
# see if they are present.
1228
# However, we don't know the impact of keeping this list of things
1229
# that I'm going to search for every node I come across from here on
1231
# It doesn't handle the case when the parent key is missing on a
1232
# page that we *don't* read. So we already have to handle being
1233
# re-entrant for that.
1234
# Since most keys contain a date string, they are more likely to be
1235
# found earlier in the file than later, but we would know that right
1236
# away (key < min_key), and wouldn't keep searching it on every other
1237
# page that we read.
1238
# Mostly, it is an idea, one which should be benchmarked.
1239
parents_not_on_page = set()
1241
for node_index, sub_keys in nodes_and_keys:
1244
# sub_keys is all of the keys we are looking for that should exist
1245
# on this page, if they aren't here, then they won't be found
1246
node = nodes[node_index]
1247
node_keys = node.keys
1248
parents_to_check = set()
1249
for next_sub_key in sub_keys:
1250
if next_sub_key not in node_keys:
1251
# This one is just not present in the index at all
1252
missing_keys.add(next_sub_key)
1254
value, refs = node_keys[next_sub_key]
1255
parent_keys = refs[ref_list_num]
1256
parent_map[next_sub_key] = parent_keys
1257
parents_to_check.update(parent_keys)
1258
# Don't look for things we've already found
1259
parents_to_check = parents_to_check.difference(parent_map)
1260
# this can be used to test the benefit of having the check loop
1262
# parents_not_on_page.update(parents_to_check)
1264
while parents_to_check:
1265
next_parents_to_check = set()
1266
for key in parents_to_check:
1267
if key in node_keys:
1268
value, refs = node_keys[key]
1269
parent_keys = refs[ref_list_num]
1270
parent_map[key] = parent_keys
1271
next_parents_to_check.update(parent_keys)
1273
# This parent either is genuinely missing, or should be
1274
# found on another page. Perf test whether it is better
1275
# to check if this node should fit on this page or not.
1276
# in the 'everything-in-one-pack' scenario, this *not*
1277
# doing the check is 237ms vs 243ms.
1278
# So slightly better, but I assume the standard 'lots
1279
# of packs' is going to show a reasonable improvement
1280
# from the check, because it avoids 'going around
1281
# again' for everything that is in another index
1282
# parents_not_on_page.add(key)
1283
# Missing for some reason
1284
if key < node.min_key:
1285
# in the case of bzr.dev, 3.4k/5.3k misses are
1286
# 'earlier' misses (65%)
1287
parents_not_on_page.add(key)
1288
elif key > node.max_key:
1289
# This parent key would be present on a different
1291
parents_not_on_page.add(key)
1293
# assert key != node.min_key and key != node.max_key
1294
# If it was going to be present, it would be on
1295
# *this* page, so mark it missing.
1296
missing_keys.add(key)
1297
parents_to_check = next_parents_to_check.difference(parent_map)
1298
# Might want to do another .difference() from missing_keys
1299
# parents_not_on_page could have been found on a different page, or be
1300
# known to be missing. So cull out everything that has already been
1302
search_keys = parents_not_on_page.difference(
1303
parent_map).difference(missing_keys)
889
1306
def iter_entries_prefix(self, keys):
890
1307
"""Iterate over keys within the index using prefix matching.
1065
1486
"""Read some nodes from disk into the LRU cache.
1067
1488
This performs a readv to get the node data into memory, and parses each
1068
node, the yields it to the caller. The nodes are requested in the
1489
node, then yields it to the caller. The nodes are requested in the
1069
1490
supplied order. If possible doing sort() on the list before requesting
1070
1491
a read may improve performance.
1072
1493
:param nodes: The nodes to read. 0 - first node, 1 - second node etc.
1496
# may be the byte string of the whole file
1498
# list of (offset, length) regions of the file that should, evenually
1499
# be read in to data_ranges, either from 'bytes' or from the transport
1501
base_offset = self._base_offset
1076
1502
for index in nodes:
1077
offset = index * _PAGE_SIZE
1503
offset = (index * _PAGE_SIZE)
1078
1504
size = _PAGE_SIZE
1080
1506
# Root node - special case
1082
1508
size = min(_PAGE_SIZE, self._size)
1084
stream = self._transport.get(self._name)
1085
start = stream.read(_PAGE_SIZE)
1086
# Avoid doing this again
1087
self._size = len(start)
1088
size = min(_PAGE_SIZE, self._size)
1510
# The only case where we don't know the size, is for very
1511
# small indexes. So we read the whole thing
1512
bytes = self._transport.get_bytes(self._name)
1513
num_bytes = len(bytes)
1514
self._size = num_bytes - base_offset
1515
# the whole thing should be parsed out of 'bytes'
1516
ranges = [(start, min(_PAGE_SIZE, num_bytes - start))
1517
for start in xrange(base_offset, num_bytes, _PAGE_SIZE)]
1520
if offset > self._size:
1521
raise AssertionError('tried to read past the end'
1522
' of the file %s > %s'
1523
% (offset, self._size))
1090
1524
size = min(size, self._size - offset)
1091
ranges.append((offset, size))
1525
ranges.append((base_offset + offset, size))
1094
if self._file is None:
1528
elif bytes is not None:
1529
# already have the whole file
1530
data_ranges = [(start, bytes[start:start+size])
1531
for start, size in ranges]
1532
elif self._file is None:
1095
1533
data_ranges = self._transport.readv(self._name, ranges)
1097
1535
data_ranges = []
added
removed
bzrlib/btree_index.py
