/brz/remove-bazaar

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

« back to all changes in this revision

Viewing changes to bzrlib/btree_index.py

Merge bzr.dev

Show diffs side-by-side

added added

removed removed

Lines of Context:
bzrlib/btree_index.py
139
139
        self._nodes = {}
140
140
        # Indicate it hasn't been built yet
141
141
        self._nodes_by_key = None
 
142
        self._optimize_for_size = False
142
143
 
143
144
    def add_node(self, key, value, references=()):
144
145
        """Add a node to the index.
276
277
                    length = _PAGE_SIZE
277
278
                    if internal_row.nodes == 0:
278
279
                        length -= _RESERVED_HEADER_BYTES # padded
279
 
                    internal_row.writer = chunk_writer.ChunkWriter(length, 0)
 
280
                    internal_row.writer = chunk_writer.ChunkWriter(length, 0,
 
281
                        optimize_for_size=self._optimize_for_size)
280
282
                    internal_row.writer.write(_INTERNAL_FLAG)
281
283
                    internal_row.writer.write(_INTERNAL_OFFSET +
282
284
                        str(rows[pos + 1].nodes) + "\n")
284
286
            length = _PAGE_SIZE
285
287
            if rows[-1].nodes == 0:
286
288
                length -= _RESERVED_HEADER_BYTES # padded
287
 
            rows[-1].writer = chunk_writer.ChunkWriter(length)
 
289
            rows[-1].writer = chunk_writer.ChunkWriter(length,
 
290
                optimize_for_size=self._optimize_for_size)
288
291
            rows[-1].writer.write(_LEAF_FLAG)
289
292
        if rows[-1].writer.write(line):
290
293
            # this key did not fit in the node:
313
316
                # This will be padded, hence the -100
314
317
                new_row.writer = chunk_writer.ChunkWriter(
315
318
                    _PAGE_SIZE - _RESERVED_HEADER_BYTES,
316
 
                    reserved_bytes)
 
319
                    reserved_bytes,
 
320
                    optimize_for_size=self._optimize_for_size)
317
321
                new_row.writer.write(_INTERNAL_FLAG)
318
322
                new_row.writer.write(_INTERNAL_OFFSET +
319
323
                    str(rows[1].nodes - 1) + "\n")
347
351
                # First key triggers the first row
348
352
                rows.append(_LeafBuilderRow())
349
353
            key_count += 1
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
354
            string_key, line = _btree_serializer._flatten_node(node,
357
355
                                    self.reference_lists)
358
356
            self._add_key(string_key, line, rows)
433
431
            efficient order for the index (keys iteration order in this case).
434
432
        """
435
433
        keys = set(keys)
 
434
        local_keys = keys.intersection(self._keys)
436
435
        if self.reference_lists:
437
 
            for key in keys.intersection(self._keys):
 
436
            for key in local_keys:
438
437
                node = self._nodes[key]
439
438
                yield self, key, node[1], node[0]
440
439
        else:
441
 
            for key in keys.intersection(self._keys):
 
440
            for key in local_keys:
442
441
                node = self._nodes[key]
443
442
                yield self, key, node[1]
444
 
        keys.difference_update(self._keys)
 
443
        # Find things that are in backing indices that have not been handled
 
444
        # yet.
 
445
        if not self._backing_indices:
 
446
            return # We won't find anything there either
 
447
        # Remove all of the keys that we found locally
 
448
        keys.difference_update(local_keys)
445
449
        for backing in self._backing_indices:
446
450
            if backing is None:
447
451
                continue
607
611
        self._name = name
608
612
        self._size = size
609
613
        self._file = None
610
 
        self._page_size = transport.recommended_page_size()
 
614
        self._recommended_pages = self._compute_recommended_pages()
611
615
        self._root_node = None
612
616
        # Default max size is 100,000 leave values
613
617
        self._leaf_value_cache = None # lru_cache.LRUCache(100*1000)
628
632
    def __ne__(self, other):
629
633
        return not self.__eq__(other)
630
634
 
631
 
    def _get_root_node(self):
632
 
        if self._root_node is None:
633
 
            # We may not have a root node yet
634
 
            nodes = list(self._read_nodes([0]))
635
 
            if len(nodes):
636
 
                self._root_node = nodes[0][1]
637
 
        return self._root_node
638
 
 
639
 
    def _cache_nodes(self, nodes, cache):
 
635
    def _get_and_cache_nodes(self, nodes):
640
636
        """Read nodes and cache them in the lru.
641
637
 
642
638
        The nodes list supplied is sorted and then read from disk, each node
649
645
 
650
646
        :return: A dict of {node_pos: node}
651
647
        """
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)
655
648
        found = {}
 
649
        start_of_leaves = None
656
650
        for node_pos, node in self._read_nodes(sorted(nodes)):
657
651
            if node_pos == 0: # Special case
658
652
                self._root_node = node
659
653
            else:
660
 
                cache.add(node_pos, node)
 
654
                if start_of_leaves is None:
 
655
                    start_of_leaves = self._row_offsets[-2]
 
656
                if node_pos < start_of_leaves:
 
657
                    self._internal_node_cache.add(node_pos, node)
 
658
                else:
 
659
                    self._leaf_node_cache.add(node_pos, node)
661
660
            found[node_pos] = node
662
661
        return found
663
662
 
 
663
    def _compute_recommended_pages(self):
 
664
        """Convert transport's recommended_page_size into btree pages.
 
665
 
 
666
        recommended_page_size is in bytes, we want to know how many _PAGE_SIZE
 
667
        pages fit in that length.
 
668
        """
 
669
        recommended_read = self._transport.recommended_page_size()
 
670
        recommended_pages = int(math.ceil(recommended_read /
 
671
                                          float(_PAGE_SIZE)))
 
672
        return recommended_pages
 
673
 
 
674
    def _compute_total_pages_in_index(self):
 
675
        """How many pages are in the index.
 
676
 
 
677
        If we have read the header we will use the value stored there.
 
678
        Otherwise it will be computed based on the length of the index.
 
679
        """
 
680
        if self._size is None:
 
681
            raise AssertionError('_compute_total_pages_in_index should not be'
 
682
                                 ' called when self._size is None')
 
683
        if self._root_node is not None:
 
684
            # This is the number of pages as defined by the header
 
685
            return self._row_offsets[-1]
 
686
        # This is the number of pages as defined by the size of the index. They
 
687
        # should be indentical.
 
688
        total_pages = int(math.ceil(self._size / float(_PAGE_SIZE)))
 
689
        return total_pages
 
690
 
 
691
    def _expand_offsets(self, offsets):
 
692
        """Find extra pages to download.
 
693
 
 
694
        The idea is that we always want to make big-enough requests (like 64kB
 
695
        for http), so that we don't waste round trips. So given the entries
 
696
        that we already have cached and the new pages being downloaded figure
 
697
        out what other pages we might want to read.
 
698
 
 
699
        See also doc/developers/btree_index_prefetch.txt for more details.
 
700
 
 
701
        :param offsets: The offsets to be read
 
702
        :return: A list of offsets to download
 
703
        """
 
704
        if 'index' in debug.debug_flags:
 
705
            trace.mutter('expanding: %s\toffsets: %s', self._name, offsets)
 
706
 
 
707
        if len(offsets) >= self._recommended_pages:
 
708
            # Don't add more, we are already requesting more than enough
 
709
            if 'index' in debug.debug_flags:
 
710
                trace.mutter('  not expanding large request (%s >= %s)',
 
711
                             len(offsets), self._recommended_pages)
 
712
            return offsets
 
713
        if self._size is None:
 
714
            # Don't try anything, because we don't know where the file ends
 
715
            if 'index' in debug.debug_flags:
 
716
                trace.mutter('  not expanding without knowing index size')
 
717
            return offsets
 
718
        total_pages = self._compute_total_pages_in_index()
 
719
        cached_offsets = self._get_offsets_to_cached_pages()
 
720
        # If reading recommended_pages would read the rest of the index, just
 
721
        # do so.
 
722
        if total_pages - len(cached_offsets) <= self._recommended_pages:
 
723
            # Read whatever is left
 
724
            if cached_offsets:
 
725
                expanded = [x for x in xrange(total_pages)
 
726
                               if x not in cached_offsets]
 
727
            else:
 
728
                expanded = range(total_pages)
 
729
            if 'index' in debug.debug_flags:
 
730
                trace.mutter('  reading all unread pages: %s', expanded)
 
731
            return expanded
 
732
 
 
733
        if self._root_node is None:
 
734
            # ATM on the first read of the root node of a large index, we don't
 
735
            # bother pre-reading any other pages. This is because the
 
736
            # likelyhood of actually reading interesting pages is very low.
 
737
            # See doc/developers/btree_index_prefetch.txt for a discussion, and
 
738
            # a possible implementation when we are guessing that the second
 
739
            # layer index is small
 
740
            final_offsets = offsets
 
741
        else:
 
742
            tree_depth = len(self._row_lengths)
 
743
            if len(cached_offsets) < tree_depth and len(offsets) == 1:
 
744
                # We haven't read enough to justify expansion
 
745
                # If we are only going to read the root node, and 1 leaf node,
 
746
                # then it isn't worth expanding our request. Once we've read at
 
747
                # least 2 nodes, then we are probably doing a search, and we
 
748
                # start expanding our requests.
 
749
                if 'index' in debug.debug_flags:
 
750
                    trace.mutter('  not expanding on first reads')
 
751
                return offsets
 
752
            final_offsets = self._expand_to_neighbors(offsets, cached_offsets,
 
753
                                                      total_pages)
 
754
 
 
755
        final_offsets = sorted(final_offsets)
 
756
        if 'index' in debug.debug_flags:
 
757
            trace.mutter('expanded:  %s', final_offsets)
 
758
        return final_offsets
 
759
 
 
760
    def _expand_to_neighbors(self, offsets, cached_offsets, total_pages):
 
761
        """Expand requests to neighbors until we have enough pages.
 
762
 
 
763
        This is called from _expand_offsets after policy has determined that we
 
764
        want to expand.
 
765
        We only want to expand requests within a given layer. We cheat a little
 
766
        bit and assume all requests will be in the same layer. This is true
 
767
        given the current design, but if it changes this algorithm may perform
 
768
        oddly.
 
769
 
 
770
        :param offsets: requested offsets
 
771
        :param cached_offsets: offsets for pages we currently have cached
 
772
        :return: A set() of offsets after expansion
 
773
        """
 
774
        final_offsets = set(offsets)
 
775
        first = end = None
 
776
        new_tips = set(final_offsets)
 
777
        while len(final_offsets) < self._recommended_pages and new_tips:
 
778
            next_tips = set()
 
779
            for pos in new_tips:
 
780
                if first is None:
 
781
                    first, end = self._find_layer_first_and_end(pos)
 
782
                previous = pos - 1
 
783
                if (previous > 0
 
784
                    and previous not in cached_offsets
 
785
                    and previous not in final_offsets
 
786
                    and previous >= first):
 
787
                    next_tips.add(previous)
 
788
                after = pos + 1
 
789
                if (after < total_pages
 
790
                    and after not in cached_offsets
 
791
                    and after not in final_offsets
 
792
                    and after < end):
 
793
                    next_tips.add(after)
 
794
                # This would keep us from going bigger than
 
795
                # recommended_pages by only expanding the first offsets.
 
796
                # However, if we are making a 'wide' request, it is
 
797
                # reasonable to expand all points equally.
 
798
                # if len(final_offsets) > recommended_pages:
 
799
                #     break
 
800
            final_offsets.update(next_tips)
 
801
            new_tips = next_tips
 
802
        return final_offsets
 
803
 
 
804
    def _find_layer_first_and_end(self, offset):
 
805
        """Find the start/stop nodes for the layer corresponding to offset.
 
806
 
 
807
        :return: (first, end)
 
808
            first is the first node in this layer
 
809
            end is the first node of the next layer
 
810
        """
 
811
        first = end = 0
 
812
        for roffset in self._row_offsets:
 
813
            first = end
 
814
            end = roffset
 
815
            if offset < roffset:
 
816
                break
 
817
        return first, end
 
818
 
 
819
    def _get_offsets_to_cached_pages(self):
 
820
        """Determine what nodes we already have cached."""
 
821
        cached_offsets = set(self._internal_node_cache.keys())
 
822
        cached_offsets.update(self._leaf_node_cache.keys())
 
823
        if self._root_node is not None:
 
824
            cached_offsets.add(0)
 
825
        return cached_offsets
 
826
 
 
827
    def _get_root_node(self):
 
828
        if self._root_node is None:
 
829
            # We may not have a root node yet
 
830
            self._get_internal_nodes([0])
 
831
        return self._root_node
 
832
 
664
833
    def _get_nodes(self, cache, node_indexes):
665
834
        found = {}
666
835
        needed = []
672
841
                found[idx] = cache[idx]
673
842
            except KeyError:
674
843
                needed.append(idx)
675
 
        found.update(self._cache_nodes(needed, cache))
 
844
        if not needed:
 
845
            return found
 
846
        needed = self._expand_offsets(needed)
 
847
        found.update(self._get_and_cache_nodes(needed))
676
848
        return found
677
849
 
678
850
    def _get_internal_nodes(self, node_indexes):
682
854
        """
683
855
        return self._get_nodes(self._internal_node_cache, node_indexes)
684
856
 
685
 
    def _get_leaf_nodes(self, node_indexes):
686
 
        """Get a bunch of nodes, from cache or disk."""
687
 
        found = self._get_nodes(self._leaf_node_cache, node_indexes)
 
857
    def _cache_leaf_values(self, nodes):
 
858
        """Cache directly from key => value, skipping the btree."""
688
859
        if self._leaf_value_cache is not None:
689
 
            for node in found.itervalues():
 
860
            for node in nodes.itervalues():
690
861
                for key, value in node.keys.iteritems():
691
862
                    if key in self._leaf_value_cache:
692
863
                        # Don't add the rest of the keys, we've seen this node
693
864
                        # before.
694
865
                        break
695
866
                    self._leaf_value_cache[key] = value
 
867
 
 
868
    def _get_leaf_nodes(self, node_indexes):
 
869
        """Get a bunch of nodes, from cache or disk."""
 
870
        found = self._get_nodes(self._leaf_node_cache, node_indexes)
 
871
        self._cache_leaf_values(found)
696
872
        return found
697
873
 
698
874
    def iter_all_entries(self):
711
887
            return
712
888
        start_of_leaves = self._row_offsets[-2]
713
889
        end_of_leaves = self._row_offsets[-1]
714
 
        needed_nodes = range(start_of_leaves, end_of_leaves)
 
890
        needed_offsets = range(start_of_leaves, end_of_leaves)
 
891
        if needed_offsets == [0]:
 
892
            # Special case when we only have a root node, as we have already
 
893
            # read everything
 
894
            nodes = [(0, self._root_node)]
 
895
        else:
 
896
            nodes = self._read_nodes(needed_offsets)
715
897
        # We iterate strictly in-order so that we can use this function
716
898
        # for spilling index builds to disk.
717
899
        if self.node_ref_lists:
718
 
            for _, node in self._read_nodes(needed_nodes):
 
900
            for _, node in nodes:
719
901
                for key, (value, refs) in sorted(node.keys.items()):
720
902
                    yield (self, key, value, refs)
721
903
        else:
722
 
            for _, node in self._read_nodes(needed_nodes):
 
904
            for _, node in nodes:
723
905
                for key, (value, refs) in sorted(node.keys.items()):
724
906
                    yield (self, key, value)
725
907
 
1008
1190
            self._get_root_node()
1009
1191
        return self._key_count
1010
1192
 
 
1193
    def _compute_row_offsets(self):
 
1194
        """Fill out the _row_offsets attribute based on _row_lengths."""
 
1195
        offsets = []
 
1196
        row_offset = 0
 
1197
        for row in self._row_lengths:
 
1198
            offsets.append(row_offset)
 
1199
            row_offset += row
 
1200
        offsets.append(row_offset)
 
1201
        self._row_offsets = offsets
 
1202
 
1011
1203
    def _parse_header_from_bytes(self, bytes):
1012
1204
        """Parse the header from a region of bytes.
1013
1205
 
1049
1241
                if len(length)])
1050
1242
        except ValueError:
1051
1243
            raise errors.BadIndexOptions(self)
1052
 
        offsets = []
1053
 
        row_offset = 0
1054
 
        for row in self._row_lengths:
1055
 
            offsets.append(row_offset)
1056
 
            row_offset += row
1057
 
        offsets.append(row_offset)
1058
 
        self._row_offsets = offsets
 
1244
        self._compute_row_offsets()
1059
1245
 
1060
1246
        # calculate the bytes we have processed
1061
1247
        header_end = (len(signature) + sum(map(len, lines[0:4])) + 4)
1072
1258
        :param nodes: The nodes to read. 0 - first node, 1 - second node etc.
1073
1259
        :return: None
1074
1260
        """
 
1261
        bytes = None
1075
1262
        ranges = []
1076
1263
        for index in nodes:
1077
1264
            offset = index * _PAGE_SIZE
1081
1268
                if self._size:
1082
1269
                    size = min(_PAGE_SIZE, self._size)
1083
1270
                else:
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)
 
1271
                    # The only case where we don't know the size, is for very
 
1272
                    # small indexes. So we read the whole thing
 
1273
                    bytes = self._transport.get_bytes(self._name)
 
1274
                    self._size = len(bytes)
 
1275
                    ranges.append((0, len(bytes)))
 
1276
                    break
1089
1277
            else:
 
1278
                if offset > self._size:
 
1279
                    raise AssertionError('tried to read past the end'
 
1280
                                         ' of the file %s > %s'
 
1281
                                         % (offset, self._size))
1090
1282
                size = min(size, self._size - offset)
1091
1283
            ranges.append((offset, size))
1092
1284
        if not ranges:
1093
1285
            return
1094
 
        if self._file is None:
 
1286
        if bytes:
 
1287
            data_ranges = [(offset, bytes[offset:offset+_PAGE_SIZE])
 
1288
                           for offset in xrange(0, len(bytes), _PAGE_SIZE)]
 
1289
        elif self._file is None:
1095
1290
            data_ranges = self._transport.readv(self._name, ranges)
1096
1291
        else:
1097
1292
            data_ranges = []