1
# Copyright (C) 2008-2011 Canonical Ltd
3
# This program is free software; you can redistribute it and/or modify
4
# it under the terms of the GNU General Public License as published by
5
# the Free Software Foundation; either version 2 of the License, or
6
# (at your option) any later version.
8
# This program is distributed in the hope that it will be useful,
9
# but WITHOUT ANY WARRANTY; without even the implied warranty of
10
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
11
# GNU General Public License for more details.
13
# You should have received a copy of the GNU General Public License
14
# along with this program; if not, write to the Free Software
15
# Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA
17
"""Core compression logic for compressing streams of related files."""
26
from bzrlib.lazy_import import lazy_import
27
lazy_import(globals(), """
41
from bzrlib.repofmt import pack_repo
44
from bzrlib.btree_index import BTreeBuilder
45
from bzrlib.lru_cache import LRUSizeCache
46
from bzrlib.versionedfile import (
50
ChunkedContentFactory,
51
FulltextContentFactory,
52
VersionedFilesWithFallbacks,
55
# Minimum number of uncompressed bytes to try fetch at once when retrieving
56
# groupcompress blocks.
59
_USE_LZMA = False and (pylzma is not None)
61
# osutils.sha_string('')
62
_null_sha1 = 'da39a3ee5e6b4b0d3255bfef95601890afd80709'
64
def sort_gc_optimal(parent_map):
65
"""Sort and group the keys in parent_map into groupcompress order.
67
groupcompress is defined (currently) as reverse-topological order, grouped
70
:return: A sorted-list of keys
72
# groupcompress ordering is approximately reverse topological,
73
# properly grouped by file-id.
75
for key, value in parent_map.iteritems():
76
if isinstance(key, str) or len(key) == 1:
81
per_prefix_map[prefix][key] = value
83
per_prefix_map[prefix] = {key: value}
86
for prefix in sorted(per_prefix_map):
87
present_keys.extend(reversed(tsort.topo_sort(per_prefix_map[prefix])))
91
# The max zlib window size is 32kB, so if we set 'max_size' output of the
92
# decompressor to the requested bytes + 32kB, then we should guarantee
93
# num_bytes coming out.
94
_ZLIB_DECOMP_WINDOW = 32*1024
96
class GroupCompressBlock(object):
97
"""An object which maintains the internal structure of the compressed data.
99
This tracks the meta info (start of text, length, type, etc.)
102
# Group Compress Block v1 Zlib
103
GCB_HEADER = 'gcb1z\n'
104
# Group Compress Block v1 Lzma
105
GCB_LZ_HEADER = 'gcb1l\n'
106
GCB_KNOWN_HEADERS = (GCB_HEADER, GCB_LZ_HEADER)
109
# map by key? or just order in file?
110
self._compressor_name = None
111
self._z_content_chunks = None
112
self._z_content_decompressor = None
113
self._z_content_length = None
114
self._content_length = None
116
self._content_chunks = None
119
# This is the maximum number of bytes this object will reference if
120
# everything is decompressed. However, if we decompress less than
121
# everything... (this would cause some problems for LRUSizeCache)
122
return self._content_length + self._z_content_length
124
def _ensure_content(self, num_bytes=None):
125
"""Make sure that content has been expanded enough.
127
:param num_bytes: Ensure that we have extracted at least num_bytes of
128
content. If None, consume everything
130
if self._content_length is None:
131
raise AssertionError('self._content_length should never be None')
132
if num_bytes is None:
133
num_bytes = self._content_length
134
elif (self._content_length is not None
135
and num_bytes > self._content_length):
136
raise AssertionError(
137
'requested num_bytes (%d) > content length (%d)'
138
% (num_bytes, self._content_length))
139
# Expand the content if required
140
if self._content is None:
141
if self._content_chunks is not None:
142
self._content = ''.join(self._content_chunks)
143
self._content_chunks = None
144
if self._content is None:
145
# We join self._z_content_chunks here, because if we are
146
# decompressing, then it is *very* likely that we have a single
148
if self._z_content_chunks is None:
149
raise AssertionError('No content to decompress')
150
z_content = ''.join(self._z_content_chunks)
153
elif self._compressor_name == 'lzma':
154
# We don't do partial lzma decomp yet
155
self._content = pylzma.decompress(z_content)
156
elif self._compressor_name == 'zlib':
157
# Start a zlib decompressor
158
if num_bytes * 4 > self._content_length * 3:
159
# If we are requesting more that 3/4ths of the content,
160
# just extract the whole thing in a single pass
161
num_bytes = self._content_length
162
self._content = zlib.decompress(z_content)
164
self._z_content_decompressor = zlib.decompressobj()
165
# Seed the decompressor with the uncompressed bytes, so
166
# that the rest of the code is simplified
167
self._content = self._z_content_decompressor.decompress(
168
z_content, num_bytes + _ZLIB_DECOMP_WINDOW)
169
if not self._z_content_decompressor.unconsumed_tail:
170
self._z_content_decompressor = None
172
raise AssertionError('Unknown compressor: %r'
173
% self._compressor_name)
174
# Any bytes remaining to be decompressed will be in the decompressors
177
# Do we have enough bytes already?
178
if len(self._content) >= num_bytes:
180
# If we got this far, and don't have a decompressor, something is wrong
181
if self._z_content_decompressor is None:
182
raise AssertionError(
183
'No decompressor to decompress %d bytes' % num_bytes)
184
remaining_decomp = self._z_content_decompressor.unconsumed_tail
185
if not remaining_decomp:
186
raise AssertionError('Nothing left to decompress')
187
needed_bytes = num_bytes - len(self._content)
188
# We always set max_size to 32kB over the minimum needed, so that
189
# zlib will give us as much as we really want.
190
# TODO: If this isn't good enough, we could make a loop here,
191
# that keeps expanding the request until we get enough
192
self._content += self._z_content_decompressor.decompress(
193
remaining_decomp, needed_bytes + _ZLIB_DECOMP_WINDOW)
194
if len(self._content) < num_bytes:
195
raise AssertionError('%d bytes wanted, only %d available'
196
% (num_bytes, len(self._content)))
197
if not self._z_content_decompressor.unconsumed_tail:
198
# The stream is finished
199
self._z_content_decompressor = None
201
def _parse_bytes(self, bytes, pos):
202
"""Read the various lengths from the header.
204
This also populates the various 'compressed' buffers.
206
:return: The position in bytes just after the last newline
208
# At present, we have 2 integers for the compressed and uncompressed
209
# content. In base10 (ascii) 14 bytes can represent > 1TB, so to avoid
210
# checking too far, cap the search to 14 bytes.
211
pos2 = bytes.index('\n', pos, pos + 14)
212
self._z_content_length = int(bytes[pos:pos2])
214
pos2 = bytes.index('\n', pos, pos + 14)
215
self._content_length = int(bytes[pos:pos2])
217
if len(bytes) != (pos + self._z_content_length):
218
# XXX: Define some GCCorrupt error ?
219
raise AssertionError('Invalid bytes: (%d) != %d + %d' %
220
(len(bytes), pos, self._z_content_length))
221
self._z_content_chunks = (bytes[pos:],)
224
def _z_content(self):
225
"""Return z_content_chunks as a simple string.
227
Meant only to be used by the test suite.
229
if self._z_content_chunks is not None:
230
return ''.join(self._z_content_chunks)
234
def from_bytes(cls, bytes):
236
if bytes[:6] not in cls.GCB_KNOWN_HEADERS:
237
raise ValueError('bytes did not start with any of %r'
238
% (cls.GCB_KNOWN_HEADERS,))
239
# XXX: why not testing the whole header ?
241
out._compressor_name = 'zlib'
242
elif bytes[4] == 'l':
243
out._compressor_name = 'lzma'
245
raise ValueError('unknown compressor: %r' % (bytes,))
246
out._parse_bytes(bytes, 6)
249
def extract(self, key, start, end, sha1=None):
250
"""Extract the text for a specific key.
252
:param key: The label used for this content
253
:param sha1: TODO (should we validate only when sha1 is supplied?)
254
:return: The bytes for the content
256
if start == end == 0:
258
self._ensure_content(end)
259
# The bytes are 'f' or 'd' for the type, then a variable-length
260
# base128 integer for the content size, then the actual content
261
# We know that the variable-length integer won't be longer than 5
262
# bytes (it takes 5 bytes to encode 2^32)
263
c = self._content[start]
268
raise ValueError('Unknown content control code: %s'
271
content_len, len_len = decode_base128_int(
272
self._content[start + 1:start + 6])
273
content_start = start + 1 + len_len
274
if end != content_start + content_len:
275
raise ValueError('end != len according to field header'
276
' %s != %s' % (end, content_start + content_len))
278
bytes = self._content[content_start:end]
280
bytes = apply_delta_to_source(self._content, content_start, end)
283
def set_chunked_content(self, content_chunks, length):
284
"""Set the content of this block to the given chunks."""
285
# If we have lots of short lines, it is may be more efficient to join
286
# the content ahead of time. If the content is <10MiB, we don't really
287
# care about the extra memory consumption, so we can just pack it and
288
# be done. However, timing showed 18s => 17.9s for repacking 1k revs of
289
# mysql, which is below the noise margin
290
self._content_length = length
291
self._content_chunks = content_chunks
293
self._z_content_chunks = None
295
def set_content(self, content):
296
"""Set the content of this block."""
297
self._content_length = len(content)
298
self._content = content
299
self._z_content_chunks = None
301
def _create_z_content_using_lzma(self):
302
if self._content_chunks is not None:
303
self._content = ''.join(self._content_chunks)
304
self._content_chunks = None
305
if self._content is None:
306
raise AssertionError('Nothing to compress')
307
z_content = pylzma.compress(self._content)
308
self._z_content_chunks = (z_content,)
309
self._z_content_length = len(z_content)
311
def _create_z_content_from_chunks(self, chunks):
312
compressor = zlib.compressobj(zlib.Z_DEFAULT_COMPRESSION)
313
# Peak in this point is 1 fulltext, 1 compressed text, + zlib overhead
314
# (measured peak is maybe 30MB over the above...)
315
compressed_chunks = map(compressor.compress, chunks)
316
compressed_chunks.append(compressor.flush())
317
# Ignore empty chunks
318
self._z_content_chunks = [c for c in compressed_chunks if c]
319
self._z_content_length = sum(map(len, self._z_content_chunks))
321
def _create_z_content(self):
322
if self._z_content_chunks is not None:
325
self._create_z_content_using_lzma()
327
if self._content_chunks is not None:
328
chunks = self._content_chunks
330
chunks = (self._content,)
331
self._create_z_content_from_chunks(chunks)
334
"""Create the byte stream as a series of 'chunks'"""
335
self._create_z_content()
337
header = self.GCB_LZ_HEADER
339
header = self.GCB_HEADER
340
chunks = ['%s%d\n%d\n'
341
% (header, self._z_content_length, self._content_length),
343
chunks.extend(self._z_content_chunks)
344
total_len = sum(map(len, chunks))
345
return total_len, chunks
348
"""Encode the information into a byte stream."""
349
total_len, chunks = self.to_chunks()
350
return ''.join(chunks)
352
def _dump(self, include_text=False):
353
"""Take this block, and spit out a human-readable structure.
355
:param include_text: Inserts also include text bits, chose whether you
356
want this displayed in the dump or not.
357
:return: A dump of the given block. The layout is something like:
358
[('f', length), ('d', delta_length, text_length, [delta_info])]
359
delta_info := [('i', num_bytes, text), ('c', offset, num_bytes),
362
self._ensure_content()
365
while pos < self._content_length:
366
kind = self._content[pos]
368
if kind not in ('f', 'd'):
369
raise ValueError('invalid kind character: %r' % (kind,))
370
content_len, len_len = decode_base128_int(
371
self._content[pos:pos + 5])
373
if content_len + pos > self._content_length:
374
raise ValueError('invalid content_len %d for record @ pos %d'
375
% (content_len, pos - len_len - 1))
376
if kind == 'f': # Fulltext
378
text = self._content[pos:pos+content_len]
379
result.append(('f', content_len, text))
381
result.append(('f', content_len))
382
elif kind == 'd': # Delta
383
delta_content = self._content[pos:pos+content_len]
385
# The first entry in a delta is the decompressed length
386
decomp_len, delta_pos = decode_base128_int(delta_content)
387
result.append(('d', content_len, decomp_len, delta_info))
389
while delta_pos < content_len:
390
c = ord(delta_content[delta_pos])
394
delta_pos) = decode_copy_instruction(delta_content, c,
397
text = self._content[offset:offset+length]
398
delta_info.append(('c', offset, length, text))
400
delta_info.append(('c', offset, length))
401
measured_len += length
404
txt = delta_content[delta_pos:delta_pos+c]
407
delta_info.append(('i', c, txt))
410
if delta_pos != content_len:
411
raise ValueError('Delta consumed a bad number of bytes:'
412
' %d != %d' % (delta_pos, content_len))
413
if measured_len != decomp_len:
414
raise ValueError('Delta claimed fulltext was %d bytes, but'
415
' extraction resulted in %d bytes'
416
% (decomp_len, measured_len))
421
class _LazyGroupCompressFactory(object):
422
"""Yield content from a GroupCompressBlock on demand."""
424
def __init__(self, key, parents, manager, start, end, first):
425
"""Create a _LazyGroupCompressFactory
427
:param key: The key of just this record
428
:param parents: The parents of this key (possibly None)
429
:param gc_block: A GroupCompressBlock object
430
:param start: Offset of the first byte for this record in the
432
:param end: Offset of the byte just after the end of this record
433
(ie, bytes = content[start:end])
434
:param first: Is this the first Factory for the given block?
437
self.parents = parents
439
# Note: This attribute coupled with Manager._factories creates a
440
# reference cycle. Perhaps we would rather use a weakref(), or
441
# find an appropriate time to release the ref. After the first
442
# get_bytes_as call? After Manager.get_record_stream() returns
444
self._manager = manager
446
self.storage_kind = 'groupcompress-block'
448
self.storage_kind = 'groupcompress-block-ref'
454
return '%s(%s, first=%s)' % (self.__class__.__name__,
455
self.key, self._first)
457
def get_bytes_as(self, storage_kind):
458
if storage_kind == self.storage_kind:
460
# wire bytes, something...
461
return self._manager._wire_bytes()
464
if storage_kind in ('fulltext', 'chunked'):
465
if self._bytes is None:
466
# Grab and cache the raw bytes for this entry
467
# and break the ref-cycle with _manager since we don't need it
470
self._manager._prepare_for_extract()
471
except zlib.error as value:
472
raise errors.DecompressCorruption("zlib: " + str(value))
473
block = self._manager._block
474
self._bytes = block.extract(self.key, self._start, self._end)
475
# There are code paths that first extract as fulltext, and then
476
# extract as storage_kind (smart fetch). So we don't break the
477
# refcycle here, but instead in manager.get_record_stream()
478
if storage_kind == 'fulltext':
482
raise errors.UnavailableRepresentation(self.key, storage_kind,
486
class _LazyGroupContentManager(object):
487
"""This manages a group of _LazyGroupCompressFactory objects."""
489
_max_cut_fraction = 0.75 # We allow a block to be trimmed to 75% of
490
# current size, and still be considered
492
_full_block_size = 4*1024*1024
493
_full_mixed_block_size = 2*1024*1024
494
_full_enough_block_size = 3*1024*1024 # size at which we won't repack
495
_full_enough_mixed_block_size = 2*768*1024 # 1.5MB
497
def __init__(self, block, get_compressor_settings=None):
499
# We need to preserve the ordering
502
self._get_settings = get_compressor_settings
503
self._compressor_settings = None
505
def _get_compressor_settings(self):
506
if self._compressor_settings is not None:
507
return self._compressor_settings
509
if self._get_settings is not None:
510
settings = self._get_settings()
512
vf = GroupCompressVersionedFiles
513
settings = vf._DEFAULT_COMPRESSOR_SETTINGS
514
self._compressor_settings = settings
515
return self._compressor_settings
517
def add_factory(self, key, parents, start, end):
518
if not self._factories:
522
# Note that this creates a reference cycle....
523
factory = _LazyGroupCompressFactory(key, parents, self,
524
start, end, first=first)
525
# max() works here, but as a function call, doing a compare seems to be
526
# significantly faster, timeit says 250ms for max() and 100ms for the
528
if end > self._last_byte:
529
self._last_byte = end
530
self._factories.append(factory)
532
def get_record_stream(self):
533
"""Get a record for all keys added so far."""
534
for factory in self._factories:
536
# Break the ref-cycle
537
factory._bytes = None
538
factory._manager = None
539
# TODO: Consider setting self._factories = None after the above loop,
540
# as it will break the reference cycle
542
def _trim_block(self, last_byte):
543
"""Create a new GroupCompressBlock, with just some of the content."""
544
# None of the factories need to be adjusted, because the content is
545
# located in an identical place. Just that some of the unreferenced
546
# trailing bytes are stripped
547
trace.mutter('stripping trailing bytes from groupcompress block'
548
' %d => %d', self._block._content_length, last_byte)
549
new_block = GroupCompressBlock()
550
self._block._ensure_content(last_byte)
551
new_block.set_content(self._block._content[:last_byte])
552
self._block = new_block
554
def _make_group_compressor(self):
555
return GroupCompressor(self._get_compressor_settings())
557
def _rebuild_block(self):
558
"""Create a new GroupCompressBlock with only the referenced texts."""
559
compressor = self._make_group_compressor()
561
old_length = self._block._content_length
563
for factory in self._factories:
564
bytes = factory.get_bytes_as('fulltext')
565
(found_sha1, start_point, end_point,
566
type) = compressor.compress(factory.key, bytes, factory.sha1)
567
# Now update this factory with the new offsets, etc
568
factory.sha1 = found_sha1
569
factory._start = start_point
570
factory._end = end_point
571
self._last_byte = end_point
572
new_block = compressor.flush()
573
# TODO: Should we check that new_block really *is* smaller than the old
574
# block? It seems hard to come up with a method that it would
575
# expand, since we do full compression again. Perhaps based on a
576
# request that ends up poorly ordered?
577
# TODO: If the content would have expanded, then we would want to
578
# handle a case where we need to split the block.
579
# Now that we have a user-tweakable option
580
# (max_bytes_to_index), it is possible that one person set it
581
# to a very low value, causing poor compression.
582
delta = time.time() - tstart
583
self._block = new_block
584
trace.mutter('creating new compressed block on-the-fly in %.3fs'
585
' %d bytes => %d bytes', delta, old_length,
586
self._block._content_length)
588
def _prepare_for_extract(self):
589
"""A _LazyGroupCompressFactory is about to extract to fulltext."""
590
# We expect that if one child is going to fulltext, all will be. This
591
# helps prevent all of them from extracting a small amount at a time.
592
# Which in itself isn't terribly expensive, but resizing 2MB 32kB at a
593
# time (self._block._content) is a little expensive.
594
self._block._ensure_content(self._last_byte)
596
def _check_rebuild_action(self):
597
"""Check to see if our block should be repacked."""
600
for factory in self._factories:
601
total_bytes_used += factory._end - factory._start
602
if last_byte_used < factory._end:
603
last_byte_used = factory._end
604
# If we are using more than half of the bytes from the block, we have
605
# nothing else to check
606
if total_bytes_used * 2 >= self._block._content_length:
607
return None, last_byte_used, total_bytes_used
608
# We are using less than 50% of the content. Is the content we are
609
# using at the beginning of the block? If so, we can just trim the
610
# tail, rather than rebuilding from scratch.
611
if total_bytes_used * 2 > last_byte_used:
612
return 'trim', last_byte_used, total_bytes_used
614
# We are using a small amount of the data, and it isn't just packed
615
# nicely at the front, so rebuild the content.
616
# Note: This would be *nicer* as a strip-data-from-group, rather than
617
# building it up again from scratch
618
# It might be reasonable to consider the fulltext sizes for
619
# different bits when deciding this, too. As you may have a small
620
# fulltext, and a trivial delta, and you are just trading around
621
# for another fulltext. If we do a simple 'prune' you may end up
622
# expanding many deltas into fulltexts, as well.
623
# If we build a cheap enough 'strip', then we could try a strip,
624
# if that expands the content, we then rebuild.
625
return 'rebuild', last_byte_used, total_bytes_used
627
def check_is_well_utilized(self):
628
"""Is the current block considered 'well utilized'?
630
This heuristic asks if the current block considers itself to be a fully
631
developed group, rather than just a loose collection of data.
633
if len(self._factories) == 1:
634
# A block of length 1 could be improved by combining with other
635
# groups - don't look deeper. Even larger than max size groups
636
# could compress well with adjacent versions of the same thing.
638
action, last_byte_used, total_bytes_used = self._check_rebuild_action()
639
block_size = self._block._content_length
640
if total_bytes_used < block_size * self._max_cut_fraction:
641
# This block wants to trim itself small enough that we want to
642
# consider it under-utilized.
644
# TODO: This code is meant to be the twin of _insert_record_stream's
645
# 'start_new_block' logic. It would probably be better to factor
646
# out that logic into a shared location, so that it stays
648
# We currently assume a block is properly utilized whenever it is >75%
649
# of the size of a 'full' block. In normal operation, a block is
650
# considered full when it hits 4MB of same-file content. So any block
651
# >3MB is 'full enough'.
652
# The only time this isn't true is when a given block has large-object
653
# content. (a single file >4MB, etc.)
654
# Under these circumstances, we allow a block to grow to
655
# 2 x largest_content. Which means that if a given block had a large
656
# object, it may actually be under-utilized. However, given that this
657
# is 'pack-on-the-fly' it is probably reasonable to not repack large
658
# content blobs on-the-fly. Note that because we return False for all
659
# 1-item blobs, we will repack them; we may wish to reevaluate our
660
# treatment of large object blobs in the future.
661
if block_size >= self._full_enough_block_size:
663
# If a block is <3MB, it still may be considered 'full' if it contains
664
# mixed content. The current rule is 2MB of mixed content is considered
665
# full. So check to see if this block contains mixed content, and
666
# set the threshold appropriately.
668
for factory in self._factories:
669
prefix = factory.key[:-1]
670
if common_prefix is None:
671
common_prefix = prefix
672
elif prefix != common_prefix:
673
# Mixed content, check the size appropriately
674
if block_size >= self._full_enough_mixed_block_size:
677
# The content failed both the mixed check and the single-content check
678
# so obviously it is not fully utilized
679
# TODO: there is one other constraint that isn't being checked
680
# namely, that the entries in the block are in the appropriate
681
# order. For example, you could insert the entries in exactly
682
# reverse groupcompress order, and we would think that is ok.
683
# (all the right objects are in one group, and it is fully
684
# utilized, etc.) For now, we assume that case is rare,
685
# especially since we should always fetch in 'groupcompress'
689
def _check_rebuild_block(self):
690
action, last_byte_used, total_bytes_used = self._check_rebuild_action()
694
self._trim_block(last_byte_used)
695
elif action == 'rebuild':
696
self._rebuild_block()
698
raise ValueError('unknown rebuild action: %r' % (action,))
700
def _wire_bytes(self):
701
"""Return a byte stream suitable for transmitting over the wire."""
702
self._check_rebuild_block()
703
# The outer block starts with:
704
# 'groupcompress-block\n'
705
# <length of compressed key info>\n
706
# <length of uncompressed info>\n
707
# <length of gc block>\n
710
lines = ['groupcompress-block\n']
711
# The minimal info we need is the key, the start offset, and the
712
# parents. The length and type are encoded in the record itself.
713
# However, passing in the other bits makes it easier. The list of
714
# keys, and the start offset, the length
716
# 1 line with parents, '' for ()
717
# 1 line for start offset
718
# 1 line for end byte
720
for factory in self._factories:
721
key_bytes = '\x00'.join(factory.key)
722
parents = factory.parents
724
parent_bytes = 'None:'
726
parent_bytes = '\t'.join('\x00'.join(key) for key in parents)
727
record_header = '%s\n%s\n%d\n%d\n' % (
728
key_bytes, parent_bytes, factory._start, factory._end)
729
header_lines.append(record_header)
730
# TODO: Can we break the refcycle at this point and set
731
# factory._manager = None?
732
header_bytes = ''.join(header_lines)
734
header_bytes_len = len(header_bytes)
735
z_header_bytes = zlib.compress(header_bytes)
737
z_header_bytes_len = len(z_header_bytes)
738
block_bytes_len, block_chunks = self._block.to_chunks()
739
lines.append('%d\n%d\n%d\n' % (z_header_bytes_len, header_bytes_len,
741
lines.append(z_header_bytes)
742
lines.extend(block_chunks)
743
del z_header_bytes, block_chunks
744
# TODO: This is a point where we will double the memory consumption. To
745
# avoid this, we probably have to switch to a 'chunked' api
746
return ''.join(lines)
749
def from_bytes(cls, bytes):
750
# TODO: This does extra string copying, probably better to do it a
751
# different way. At a minimum this creates 2 copies of the
753
(storage_kind, z_header_len, header_len,
754
block_len, rest) = bytes.split('\n', 4)
756
if storage_kind != 'groupcompress-block':
757
raise ValueError('Unknown storage kind: %s' % (storage_kind,))
758
z_header_len = int(z_header_len)
759
if len(rest) < z_header_len:
760
raise ValueError('Compressed header len shorter than all bytes')
761
z_header = rest[:z_header_len]
762
header_len = int(header_len)
763
header = zlib.decompress(z_header)
764
if len(header) != header_len:
765
raise ValueError('invalid length for decompressed bytes')
767
block_len = int(block_len)
768
if len(rest) != z_header_len + block_len:
769
raise ValueError('Invalid length for block')
770
block_bytes = rest[z_header_len:]
772
# So now we have a valid GCB, we just need to parse the factories that
774
header_lines = header.split('\n')
776
last = header_lines.pop()
778
raise ValueError('header lines did not end with a trailing'
780
if len(header_lines) % 4 != 0:
781
raise ValueError('The header was not an even multiple of 4 lines')
782
block = GroupCompressBlock.from_bytes(block_bytes)
785
for start in xrange(0, len(header_lines), 4):
787
key = tuple(header_lines[start].split('\x00'))
788
parents_line = header_lines[start+1]
789
if parents_line == 'None:':
792
parents = tuple([tuple(segment.split('\x00'))
793
for segment in parents_line.split('\t')
795
start_offset = int(header_lines[start+2])
796
end_offset = int(header_lines[start+3])
797
result.add_factory(key, parents, start_offset, end_offset)
801
def network_block_to_records(storage_kind, bytes, line_end):
802
if storage_kind != 'groupcompress-block':
803
raise ValueError('Unknown storage kind: %s' % (storage_kind,))
804
manager = _LazyGroupContentManager.from_bytes(bytes)
805
return manager.get_record_stream()
808
class _CommonGroupCompressor(object):
810
def __init__(self, settings=None):
811
"""Create a GroupCompressor."""
816
self.labels_deltas = {}
817
self._delta_index = None # Set by the children
818
self._block = GroupCompressBlock()
822
self._settings = settings
824
def compress(self, key, bytes, expected_sha, nostore_sha=None, soft=False):
825
"""Compress lines with label key.
827
:param key: A key tuple. It is stored in the output
828
for identification of the text during decompression. If the last
829
element is 'None' it is replaced with the sha1 of the text -
831
:param bytes: The bytes to be compressed
832
:param expected_sha: If non-None, the sha the lines are believed to
833
have. During compression the sha is calculated; a mismatch will
835
:param nostore_sha: If the computed sha1 sum matches, we will raise
836
ExistingContent rather than adding the text.
837
:param soft: Do a 'soft' compression. This means that we require larger
838
ranges to match to be considered for a copy command.
840
:return: The sha1 of lines, the start and end offsets in the delta, and
841
the type ('fulltext' or 'delta').
843
:seealso VersionedFiles.add_lines:
845
if not bytes: # empty, like a dir entry, etc
846
if nostore_sha == _null_sha1:
847
raise errors.ExistingContent()
848
return _null_sha1, 0, 0, 'fulltext'
849
# we assume someone knew what they were doing when they passed it in
850
if expected_sha is not None:
853
sha1 = osutils.sha_string(bytes)
854
if nostore_sha is not None:
855
if sha1 == nostore_sha:
856
raise errors.ExistingContent()
858
key = key[:-1] + ('sha1:' + sha1,)
860
start, end, type = self._compress(key, bytes, len(bytes) / 2, soft)
861
return sha1, start, end, type
863
def _compress(self, key, bytes, max_delta_size, soft=False):
864
"""Compress lines with label key.
866
:param key: A key tuple. It is stored in the output for identification
867
of the text during decompression.
869
:param bytes: The bytes to be compressed
871
:param max_delta_size: The size above which we issue a fulltext instead
874
:param soft: Do a 'soft' compression. This means that we require larger
875
ranges to match to be considered for a copy command.
877
:return: The sha1 of lines, the start and end offsets in the delta, and
878
the type ('fulltext' or 'delta').
880
raise NotImplementedError(self._compress)
882
def extract(self, key):
883
"""Extract a key previously added to the compressor.
885
:param key: The key to extract.
886
:return: An iterable over bytes and the sha1.
888
(start_byte, start_chunk, end_byte, end_chunk) = self.labels_deltas[key]
889
delta_chunks = self.chunks[start_chunk:end_chunk]
890
stored_bytes = ''.join(delta_chunks)
891
if stored_bytes[0] == 'f':
892
fulltext_len, offset = decode_base128_int(stored_bytes[1:10])
893
data_len = fulltext_len + 1 + offset
894
if data_len != len(stored_bytes):
895
raise ValueError('Index claimed fulltext len, but stored bytes'
897
% (len(stored_bytes), data_len))
898
bytes = stored_bytes[offset + 1:]
900
# XXX: This is inefficient at best
901
source = ''.join(self.chunks[:start_chunk])
902
if stored_bytes[0] != 'd':
903
raise ValueError('Unknown content kind, bytes claim %s'
904
% (stored_bytes[0],))
905
delta_len, offset = decode_base128_int(stored_bytes[1:10])
906
data_len = delta_len + 1 + offset
907
if data_len != len(stored_bytes):
908
raise ValueError('Index claimed delta len, but stored bytes'
910
% (len(stored_bytes), data_len))
911
bytes = apply_delta(source, stored_bytes[offset + 1:])
912
bytes_sha1 = osutils.sha_string(bytes)
913
return bytes, bytes_sha1
916
"""Finish this group, creating a formatted stream.
918
After calling this, the compressor should no longer be used
920
self._block.set_chunked_content(self.chunks, self.endpoint)
922
self._delta_index = None
926
"""Call this if you want to 'revoke' the last compression.
928
After this, the data structures will be rolled back, but you cannot do
931
self._delta_index = None
932
del self.chunks[self._last[0]:]
933
self.endpoint = self._last[1]
937
"""Return the overall compression ratio."""
938
return float(self.input_bytes) / float(self.endpoint)
941
class PythonGroupCompressor(_CommonGroupCompressor):
943
def __init__(self, settings=None):
944
"""Create a GroupCompressor.
946
Used only if the pyrex version is not available.
948
super(PythonGroupCompressor, self).__init__(settings)
949
self._delta_index = LinesDeltaIndex([])
950
# The actual content is managed by LinesDeltaIndex
951
self.chunks = self._delta_index.lines
953
def _compress(self, key, bytes, max_delta_size, soft=False):
954
"""see _CommonGroupCompressor._compress"""
955
input_len = len(bytes)
956
new_lines = osutils.split_lines(bytes)
957
out_lines, index_lines = self._delta_index.make_delta(
958
new_lines, bytes_length=input_len, soft=soft)
959
delta_length = sum(map(len, out_lines))
960
if delta_length > max_delta_size:
961
# The delta is longer than the fulltext, insert a fulltext
963
out_lines = ['f', encode_base128_int(input_len)]
964
out_lines.extend(new_lines)
965
index_lines = [False, False]
966
index_lines.extend([True] * len(new_lines))
968
# this is a worthy delta, output it
971
# Update the delta_length to include those two encoded integers
972
out_lines[1] = encode_base128_int(delta_length)
974
start = self.endpoint
975
chunk_start = len(self.chunks)
976
self._last = (chunk_start, self.endpoint)
977
self._delta_index.extend_lines(out_lines, index_lines)
978
self.endpoint = self._delta_index.endpoint
979
self.input_bytes += input_len
980
chunk_end = len(self.chunks)
981
self.labels_deltas[key] = (start, chunk_start,
982
self.endpoint, chunk_end)
983
return start, self.endpoint, type
986
class PyrexGroupCompressor(_CommonGroupCompressor):
987
"""Produce a serialised group of compressed texts.
989
It contains code very similar to SequenceMatcher because of having a similar
990
task. However some key differences apply:
992
* there is no junk, we want a minimal edit not a human readable diff.
993
* we don't filter very common lines (because we don't know where a good
994
range will start, and after the first text we want to be emitting minmal
996
* we chain the left side, not the right side
997
* we incrementally update the adjacency matrix as new lines are provided.
998
* we look for matches in all of the left side, so the routine which does
999
the analagous task of find_longest_match does not need to filter on the
1003
def __init__(self, settings=None):
1004
super(PyrexGroupCompressor, self).__init__(settings)
1005
max_bytes_to_index = self._settings.get('max_bytes_to_index', 0)
1006
self._delta_index = DeltaIndex(max_bytes_to_index=max_bytes_to_index)
1008
def _compress(self, key, bytes, max_delta_size, soft=False):
1009
"""see _CommonGroupCompressor._compress"""
1010
input_len = len(bytes)
1011
# By having action/label/sha1/len, we can parse the group if the index
1012
# was ever destroyed, we have the key in 'label', we know the final
1013
# bytes are valid from sha1, and we know where to find the end of this
1014
# record because of 'len'. (the delta record itself will store the
1015
# total length for the expanded record)
1016
# 'len: %d\n' costs approximately 1% increase in total data
1017
# Having the labels at all costs us 9-10% increase, 38% increase for
1018
# inventory pages, and 5.8% increase for text pages
1019
# new_chunks = ['label:%s\nsha1:%s\n' % (label, sha1)]
1020
if self._delta_index._source_offset != self.endpoint:
1021
raise AssertionError('_source_offset != endpoint'
1022
' somehow the DeltaIndex got out of sync with'
1023
' the output lines')
1024
delta = self._delta_index.make_delta(bytes, max_delta_size)
1027
enc_length = encode_base128_int(len(bytes))
1028
len_mini_header = 1 + len(enc_length)
1029
self._delta_index.add_source(bytes, len_mini_header)
1030
new_chunks = ['f', enc_length, bytes]
1033
enc_length = encode_base128_int(len(delta))
1034
len_mini_header = 1 + len(enc_length)
1035
new_chunks = ['d', enc_length, delta]
1036
self._delta_index.add_delta_source(delta, len_mini_header)
1038
start = self.endpoint
1039
chunk_start = len(self.chunks)
1040
# Now output these bytes
1041
self._output_chunks(new_chunks)
1042
self.input_bytes += input_len
1043
chunk_end = len(self.chunks)
1044
self.labels_deltas[key] = (start, chunk_start,
1045
self.endpoint, chunk_end)
1046
if not self._delta_index._source_offset == self.endpoint:
1047
raise AssertionError('the delta index is out of sync'
1048
'with the output lines %s != %s'
1049
% (self._delta_index._source_offset, self.endpoint))
1050
return start, self.endpoint, type
1052
def _output_chunks(self, new_chunks):
1053
"""Output some chunks.
1055
:param new_chunks: The chunks to output.
1057
self._last = (len(self.chunks), self.endpoint)
1058
endpoint = self.endpoint
1059
self.chunks.extend(new_chunks)
1060
endpoint += sum(map(len, new_chunks))
1061
self.endpoint = endpoint
1064
def make_pack_factory(graph, delta, keylength, inconsistency_fatal=True):
1065
"""Create a factory for creating a pack based groupcompress.
1067
This is only functional enough to run interface tests, it doesn't try to
1068
provide a full pack environment.
1070
:param graph: Store a graph.
1071
:param delta: Delta compress contents.
1072
:param keylength: How long should keys be.
1074
def factory(transport):
1079
graph_index = BTreeBuilder(reference_lists=ref_length,
1080
key_elements=keylength)
1081
stream = transport.open_write_stream('newpack')
1082
writer = pack.ContainerWriter(stream.write)
1084
index = _GCGraphIndex(graph_index, lambda:True, parents=parents,
1085
add_callback=graph_index.add_nodes,
1086
inconsistency_fatal=inconsistency_fatal)
1087
access = pack_repo._DirectPackAccess({})
1088
access.set_writer(writer, graph_index, (transport, 'newpack'))
1089
result = GroupCompressVersionedFiles(index, access, delta)
1090
result.stream = stream
1091
result.writer = writer
1096
def cleanup_pack_group(versioned_files):
1097
versioned_files.writer.end()
1098
versioned_files.stream.close()
1101
class _BatchingBlockFetcher(object):
1102
"""Fetch group compress blocks in batches.
1104
:ivar total_bytes: int of expected number of bytes needed to fetch the
1105
currently pending batch.
1108
def __init__(self, gcvf, locations, get_compressor_settings=None):
1110
self.locations = locations
1112
self.batch_memos = {}
1113
self.memos_to_get = []
1114
self.total_bytes = 0
1115
self.last_read_memo = None
1117
self._get_compressor_settings = get_compressor_settings
1119
def add_key(self, key):
1120
"""Add another to key to fetch.
1122
:return: The estimated number of bytes needed to fetch the batch so
1125
self.keys.append(key)
1126
index_memo, _, _, _ = self.locations[key]
1127
read_memo = index_memo[0:3]
1128
# Three possibilities for this read_memo:
1129
# - it's already part of this batch; or
1130
# - it's not yet part of this batch, but is already cached; or
1131
# - it's not yet part of this batch and will need to be fetched.
1132
if read_memo in self.batch_memos:
1133
# This read memo is already in this batch.
1134
return self.total_bytes
1136
cached_block = self.gcvf._group_cache[read_memo]
1138
# This read memo is new to this batch, and the data isn't cached
1140
self.batch_memos[read_memo] = None
1141
self.memos_to_get.append(read_memo)
1142
byte_length = read_memo[2]
1143
self.total_bytes += byte_length
1145
# This read memo is new to this batch, but cached.
1146
# Keep a reference to the cached block in batch_memos because it's
1147
# certain that we'll use it when this batch is processed, but
1148
# there's a risk that it would fall out of _group_cache between now
1150
self.batch_memos[read_memo] = cached_block
1151
return self.total_bytes
1153
def _flush_manager(self):
1154
if self.manager is not None:
1155
for factory in self.manager.get_record_stream():
1158
self.last_read_memo = None
1160
def yield_factories(self, full_flush=False):
1161
"""Yield factories for keys added since the last yield. They will be
1162
returned in the order they were added via add_key.
1164
:param full_flush: by default, some results may not be returned in case
1165
they can be part of the next batch. If full_flush is True, then
1166
all results are returned.
1168
if self.manager is None and not self.keys:
1170
# Fetch all memos in this batch.
1171
blocks = self.gcvf._get_blocks(self.memos_to_get)
1172
# Turn blocks into factories and yield them.
1173
memos_to_get_stack = list(self.memos_to_get)
1174
memos_to_get_stack.reverse()
1175
for key in self.keys:
1176
index_memo, _, parents, _ = self.locations[key]
1177
read_memo = index_memo[:3]
1178
if self.last_read_memo != read_memo:
1179
# We are starting a new block. If we have a
1180
# manager, we have found everything that fits for
1181
# now, so yield records
1182
for factory in self._flush_manager():
1184
# Now start a new manager.
1185
if memos_to_get_stack and memos_to_get_stack[-1] == read_memo:
1186
# The next block from _get_blocks will be the block we
1188
block_read_memo, block = blocks.next()
1189
if block_read_memo != read_memo:
1190
raise AssertionError(
1191
"block_read_memo out of sync with read_memo"
1192
"(%r != %r)" % (block_read_memo, read_memo))
1193
self.batch_memos[read_memo] = block
1194
memos_to_get_stack.pop()
1196
block = self.batch_memos[read_memo]
1197
self.manager = _LazyGroupContentManager(block,
1198
get_compressor_settings=self._get_compressor_settings)
1199
self.last_read_memo = read_memo
1200
start, end = index_memo[3:5]
1201
self.manager.add_factory(key, parents, start, end)
1203
for factory in self._flush_manager():
1206
self.batch_memos.clear()
1207
del self.memos_to_get[:]
1208
self.total_bytes = 0
1211
class GroupCompressVersionedFiles(VersionedFilesWithFallbacks):
1212
"""A group-compress based VersionedFiles implementation."""
1214
# This controls how the GroupCompress DeltaIndex works. Basically, we
1215
# compute hash pointers into the source blocks (so hash(text) => text).
1216
# However each of these references costs some memory in trade against a
1217
# more accurate match result. For very large files, they either are
1218
# pre-compressed and change in bulk whenever they change, or change in just
1219
# local blocks. Either way, 'improved resolution' is not very helpful,
1220
# versus running out of memory trying to track everything. The default max
1221
# gives 100% sampling of a 1MB file.
1222
_DEFAULT_MAX_BYTES_TO_INDEX = 1024 * 1024
1223
_DEFAULT_COMPRESSOR_SETTINGS = {'max_bytes_to_index':
1224
_DEFAULT_MAX_BYTES_TO_INDEX}
1226
def __init__(self, index, access, delta=True, _unadded_refs=None,
1228
"""Create a GroupCompressVersionedFiles object.
1230
:param index: The index object storing access and graph data.
1231
:param access: The access object storing raw data.
1232
:param delta: Whether to delta compress or just entropy compress.
1233
:param _unadded_refs: private parameter, don't use.
1234
:param _group_cache: private parameter, don't use.
1237
self._access = access
1239
if _unadded_refs is None:
1241
self._unadded_refs = _unadded_refs
1242
if _group_cache is None:
1243
_group_cache = LRUSizeCache(max_size=50*1024*1024)
1244
self._group_cache = _group_cache
1245
self._immediate_fallback_vfs = []
1246
self._max_bytes_to_index = None
1248
def without_fallbacks(self):
1249
"""Return a clone of this object without any fallbacks configured."""
1250
return GroupCompressVersionedFiles(self._index, self._access,
1251
self._delta, _unadded_refs=dict(self._unadded_refs),
1252
_group_cache=self._group_cache)
1254
def add_lines(self, key, parents, lines, parent_texts=None,
1255
left_matching_blocks=None, nostore_sha=None, random_id=False,
1256
check_content=True):
1257
"""Add a text to the store.
1259
:param key: The key tuple of the text to add.
1260
:param parents: The parents key tuples of the text to add.
1261
:param lines: A list of lines. Each line must be a bytestring. And all
1262
of them except the last must be terminated with \\n and contain no
1263
other \\n's. The last line may either contain no \\n's or a single
1264
terminating \\n. If the lines list does meet this constraint the
1265
add routine may error or may succeed - but you will be unable to
1266
read the data back accurately. (Checking the lines have been split
1267
correctly is expensive and extremely unlikely to catch bugs so it
1268
is not done at runtime unless check_content is True.)
1269
:param parent_texts: An optional dictionary containing the opaque
1270
representations of some or all of the parents of version_id to
1271
allow delta optimisations. VERY IMPORTANT: the texts must be those
1272
returned by add_lines or data corruption can be caused.
1273
:param left_matching_blocks: a hint about which areas are common
1274
between the text and its left-hand-parent. The format is
1275
the SequenceMatcher.get_matching_blocks format.
1276
:param nostore_sha: Raise ExistingContent and do not add the lines to
1277
the versioned file if the digest of the lines matches this.
1278
:param random_id: If True a random id has been selected rather than
1279
an id determined by some deterministic process such as a converter
1280
from a foreign VCS. When True the backend may choose not to check
1281
for uniqueness of the resulting key within the versioned file, so
1282
this should only be done when the result is expected to be unique
1284
:param check_content: If True, the lines supplied are verified to be
1285
bytestrings that are correctly formed lines.
1286
:return: The text sha1, the number of bytes in the text, and an opaque
1287
representation of the inserted version which can be provided
1288
back to future add_lines calls in the parent_texts dictionary.
1290
self._index._check_write_ok()
1291
self._check_add(key, lines, random_id, check_content)
1293
# The caller might pass None if there is no graph data, but kndx
1294
# indexes can't directly store that, so we give them
1295
# an empty tuple instead.
1297
# double handling for now. Make it work until then.
1298
length = sum(map(len, lines))
1299
record = ChunkedContentFactory(key, parents, None, lines)
1300
sha1 = list(self._insert_record_stream([record], random_id=random_id,
1301
nostore_sha=nostore_sha))[0]
1302
return sha1, length, None
1304
def _add_text(self, key, parents, text, nostore_sha=None, random_id=False):
1305
"""See VersionedFiles._add_text()."""
1306
self._index._check_write_ok()
1307
self._check_add(key, None, random_id, check_content=False)
1308
if text.__class__ is not str:
1309
raise errors.BzrBadParameterUnicode("text")
1311
# The caller might pass None if there is no graph data, but kndx
1312
# indexes can't directly store that, so we give them
1313
# an empty tuple instead.
1315
# double handling for now. Make it work until then.
1317
record = FulltextContentFactory(key, parents, None, text)
1318
sha1 = list(self._insert_record_stream([record], random_id=random_id,
1319
nostore_sha=nostore_sha))[0]
1320
return sha1, length, None
1322
def add_fallback_versioned_files(self, a_versioned_files):
1323
"""Add a source of texts for texts not present in this knit.
1325
:param a_versioned_files: A VersionedFiles object.
1327
self._immediate_fallback_vfs.append(a_versioned_files)
1329
def annotate(self, key):
1330
"""See VersionedFiles.annotate."""
1331
ann = annotate.Annotator(self)
1332
return ann.annotate_flat(key)
1334
def get_annotator(self):
1335
return annotate.Annotator(self)
1337
def check(self, progress_bar=None, keys=None):
1338
"""See VersionedFiles.check()."""
1341
for record in self.get_record_stream(keys, 'unordered', True):
1342
record.get_bytes_as('fulltext')
1344
return self.get_record_stream(keys, 'unordered', True)
1346
def clear_cache(self):
1347
"""See VersionedFiles.clear_cache()"""
1348
self._group_cache.clear()
1349
self._index._graph_index.clear_cache()
1350
self._index._int_cache.clear()
1352
def _check_add(self, key, lines, random_id, check_content):
1353
"""check that version_id and lines are safe to add."""
1354
version_id = key[-1]
1355
if version_id is not None:
1356
if osutils.contains_whitespace(version_id):
1357
raise errors.InvalidRevisionId(version_id, self)
1358
self.check_not_reserved_id(version_id)
1359
# TODO: If random_id==False and the key is already present, we should
1360
# probably check that the existing content is identical to what is
1361
# being inserted, and otherwise raise an exception. This would make
1362
# the bundle code simpler.
1364
self._check_lines_not_unicode(lines)
1365
self._check_lines_are_lines(lines)
1367
def get_parent_map(self, keys):
1368
"""Get a map of the graph parents of keys.
1370
:param keys: The keys to look up parents for.
1371
:return: A mapping from keys to parents. Absent keys are absent from
1374
return self._get_parent_map_with_sources(keys)[0]
1376
def _get_parent_map_with_sources(self, keys):
1377
"""Get a map of the parents of keys.
1379
:param keys: The keys to look up parents for.
1380
:return: A tuple. The first element is a mapping from keys to parents.
1381
Absent keys are absent from the mapping. The second element is a
1382
list with the locations each key was found in. The first element
1383
is the in-this-knit parents, the second the first fallback source,
1387
sources = [self._index] + self._immediate_fallback_vfs
1390
for source in sources:
1393
new_result = source.get_parent_map(missing)
1394
source_results.append(new_result)
1395
result.update(new_result)
1396
missing.difference_update(set(new_result))
1397
return result, source_results
1399
def _get_blocks(self, read_memos):
1400
"""Get GroupCompressBlocks for the given read_memos.
1402
:returns: a series of (read_memo, block) pairs, in the order they were
1406
for read_memo in read_memos:
1408
block = self._group_cache[read_memo]
1412
cached[read_memo] = block
1414
not_cached_seen = set()
1415
for read_memo in read_memos:
1416
if read_memo in cached:
1417
# Don't fetch what we already have
1419
if read_memo in not_cached_seen:
1420
# Don't try to fetch the same data twice
1422
not_cached.append(read_memo)
1423
not_cached_seen.add(read_memo)
1424
raw_records = self._access.get_raw_records(not_cached)
1425
for read_memo in read_memos:
1427
yield read_memo, cached[read_memo]
1429
# Read the block, and cache it.
1430
zdata = raw_records.next()
1431
block = GroupCompressBlock.from_bytes(zdata)
1432
self._group_cache[read_memo] = block
1433
cached[read_memo] = block
1434
yield read_memo, block
1436
def get_missing_compression_parent_keys(self):
1437
"""Return the keys of missing compression parents.
1439
Missing compression parents occur when a record stream was missing
1440
basis texts, or a index was scanned that had missing basis texts.
1442
# GroupCompress cannot currently reference texts that are not in the
1443
# group, so this is valid for now
1446
def get_record_stream(self, keys, ordering, include_delta_closure):
1447
"""Get a stream of records for keys.
1449
:param keys: The keys to include.
1450
:param ordering: Either 'unordered' or 'topological'. A topologically
1451
sorted stream has compression parents strictly before their
1453
:param include_delta_closure: If True then the closure across any
1454
compression parents will be included (in the opaque data).
1455
:return: An iterator of ContentFactory objects, each of which is only
1456
valid until the iterator is advanced.
1458
# keys might be a generator
1459
orig_keys = list(keys)
1463
if (not self._index.has_graph
1464
and ordering in ('topological', 'groupcompress')):
1465
# Cannot topological order when no graph has been stored.
1466
# but we allow 'as-requested' or 'unordered'
1467
ordering = 'unordered'
1469
remaining_keys = keys
1472
keys = set(remaining_keys)
1473
for content_factory in self._get_remaining_record_stream(keys,
1474
orig_keys, ordering, include_delta_closure):
1475
remaining_keys.discard(content_factory.key)
1476
yield content_factory
1478
except errors.RetryWithNewPacks, e:
1479
self._access.reload_or_raise(e)
1481
def _find_from_fallback(self, missing):
1482
"""Find whatever keys you can from the fallbacks.
1484
:param missing: A set of missing keys. This set will be mutated as keys
1485
are found from a fallback_vfs
1486
:return: (parent_map, key_to_source_map, source_results)
1487
parent_map the overall key => parent_keys
1488
key_to_source_map a dict from {key: source}
1489
source_results a list of (source: keys)
1492
key_to_source_map = {}
1494
for source in self._immediate_fallback_vfs:
1497
source_parents = source.get_parent_map(missing)
1498
parent_map.update(source_parents)
1499
source_parents = list(source_parents)
1500
source_results.append((source, source_parents))
1501
key_to_source_map.update((key, source) for key in source_parents)
1502
missing.difference_update(source_parents)
1503
return parent_map, key_to_source_map, source_results
1505
def _get_ordered_source_keys(self, ordering, parent_map, key_to_source_map):
1506
"""Get the (source, [keys]) list.
1508
The returned objects should be in the order defined by 'ordering',
1509
which can weave between different sources.
1511
:param ordering: Must be one of 'topological' or 'groupcompress'
1512
:return: List of [(source, [keys])] tuples, such that all keys are in
1513
the defined order, regardless of source.
1515
if ordering == 'topological':
1516
present_keys = tsort.topo_sort(parent_map)
1518
# ordering == 'groupcompress'
1519
# XXX: This only optimizes for the target ordering. We may need
1520
# to balance that with the time it takes to extract
1521
# ordering, by somehow grouping based on
1522
# locations[key][0:3]
1523
present_keys = sort_gc_optimal(parent_map)
1524
# Now group by source:
1526
current_source = None
1527
for key in present_keys:
1528
source = key_to_source_map.get(key, self)
1529
if source is not current_source:
1530
source_keys.append((source, []))
1531
current_source = source
1532
source_keys[-1][1].append(key)
1535
def _get_as_requested_source_keys(self, orig_keys, locations, unadded_keys,
1538
current_source = None
1539
for key in orig_keys:
1540
if key in locations or key in unadded_keys:
1542
elif key in key_to_source_map:
1543
source = key_to_source_map[key]
1546
if source is not current_source:
1547
source_keys.append((source, []))
1548
current_source = source
1549
source_keys[-1][1].append(key)
1552
def _get_io_ordered_source_keys(self, locations, unadded_keys,
1555
# This is the group the bytes are stored in, followed by the
1556
# location in the group
1557
return locations[key][0]
1558
present_keys = sorted(locations.iterkeys(), key=get_group)
1559
# We don't have an ordering for keys in the in-memory object, but
1560
# lets process the in-memory ones first.
1561
present_keys = list(unadded_keys) + present_keys
1562
# Now grab all of the ones from other sources
1563
source_keys = [(self, present_keys)]
1564
source_keys.extend(source_result)
1567
def _get_remaining_record_stream(self, keys, orig_keys, ordering,
1568
include_delta_closure):
1569
"""Get a stream of records for keys.
1571
:param keys: The keys to include.
1572
:param ordering: one of 'unordered', 'topological', 'groupcompress' or
1574
:param include_delta_closure: If True then the closure across any
1575
compression parents will be included (in the opaque data).
1576
:return: An iterator of ContentFactory objects, each of which is only
1577
valid until the iterator is advanced.
1580
locations = self._index.get_build_details(keys)
1581
unadded_keys = set(self._unadded_refs).intersection(keys)
1582
missing = keys.difference(locations)
1583
missing.difference_update(unadded_keys)
1584
(fallback_parent_map, key_to_source_map,
1585
source_result) = self._find_from_fallback(missing)
1586
if ordering in ('topological', 'groupcompress'):
1587
# would be better to not globally sort initially but instead
1588
# start with one key, recurse to its oldest parent, then grab
1589
# everything in the same group, etc.
1590
parent_map = dict((key, details[2]) for key, details in
1591
locations.iteritems())
1592
for key in unadded_keys:
1593
parent_map[key] = self._unadded_refs[key]
1594
parent_map.update(fallback_parent_map)
1595
source_keys = self._get_ordered_source_keys(ordering, parent_map,
1597
elif ordering == 'as-requested':
1598
source_keys = self._get_as_requested_source_keys(orig_keys,
1599
locations, unadded_keys, key_to_source_map)
1601
# We want to yield the keys in a semi-optimal (read-wise) ordering.
1602
# Otherwise we thrash the _group_cache and destroy performance
1603
source_keys = self._get_io_ordered_source_keys(locations,
1604
unadded_keys, source_result)
1606
yield AbsentContentFactory(key)
1607
# Batch up as many keys as we can until either:
1608
# - we encounter an unadded ref, or
1609
# - we run out of keys, or
1610
# - the total bytes to retrieve for this batch > BATCH_SIZE
1611
batcher = _BatchingBlockFetcher(self, locations,
1612
get_compressor_settings=self._get_compressor_settings)
1613
for source, keys in source_keys:
1616
if key in self._unadded_refs:
1617
# Flush batch, then yield unadded ref from
1619
for factory in batcher.yield_factories(full_flush=True):
1621
bytes, sha1 = self._compressor.extract(key)
1622
parents = self._unadded_refs[key]
1623
yield FulltextContentFactory(key, parents, sha1, bytes)
1625
if batcher.add_key(key) > BATCH_SIZE:
1626
# Ok, this batch is big enough. Yield some results.
1627
for factory in batcher.yield_factories():
1630
for factory in batcher.yield_factories(full_flush=True):
1632
for record in source.get_record_stream(keys, ordering,
1633
include_delta_closure):
1635
for factory in batcher.yield_factories(full_flush=True):
1638
def get_sha1s(self, keys):
1639
"""See VersionedFiles.get_sha1s()."""
1641
for record in self.get_record_stream(keys, 'unordered', True):
1642
if record.sha1 != None:
1643
result[record.key] = record.sha1
1645
if record.storage_kind != 'absent':
1646
result[record.key] = osutils.sha_string(
1647
record.get_bytes_as('fulltext'))
1650
def insert_record_stream(self, stream):
1651
"""Insert a record stream into this container.
1653
:param stream: A stream of records to insert.
1655
:seealso VersionedFiles.get_record_stream:
1657
# XXX: Setting random_id=True makes
1658
# test_insert_record_stream_existing_keys fail for groupcompress and
1659
# groupcompress-nograph, this needs to be revisited while addressing
1660
# 'bzr branch' performance issues.
1661
for _ in self._insert_record_stream(stream, random_id=False):
1664
def _get_compressor_settings(self):
1665
if self._max_bytes_to_index is None:
1666
# TODO: VersionedFiles don't know about their containing
1667
# repository, so they don't have much of an idea about their
1668
# location. So for now, this is only a global option.
1669
c = config.GlobalConfig()
1670
val = c.get_user_option('bzr.groupcompress.max_bytes_to_index')
1674
except ValueError, e:
1675
trace.warning('Value for '
1676
'"bzr.groupcompress.max_bytes_to_index"'
1677
' %r is not an integer'
1681
val = self._DEFAULT_MAX_BYTES_TO_INDEX
1682
self._max_bytes_to_index = val
1683
return {'max_bytes_to_index': self._max_bytes_to_index}
1685
def _make_group_compressor(self):
1686
return GroupCompressor(self._get_compressor_settings())
1688
def _insert_record_stream(self, stream, random_id=False, nostore_sha=None,
1690
"""Internal core to insert a record stream into this container.
1692
This helper function has a different interface than insert_record_stream
1693
to allow add_lines to be minimal, but still return the needed data.
1695
:param stream: A stream of records to insert.
1696
:param nostore_sha: If the sha1 of a given text matches nostore_sha,
1697
raise ExistingContent, rather than committing the new text.
1698
:param reuse_blocks: If the source is streaming from
1699
groupcompress-blocks, just insert the blocks as-is, rather than
1700
expanding the texts and inserting again.
1701
:return: An iterator over the sha1 of the inserted records.
1702
:seealso insert_record_stream:
1706
def get_adapter(adapter_key):
1708
return adapters[adapter_key]
1710
adapter_factory = adapter_registry.get(adapter_key)
1711
adapter = adapter_factory(self)
1712
adapters[adapter_key] = adapter
1714
# This will go up to fulltexts for gc to gc fetching, which isn't
1716
self._compressor = self._make_group_compressor()
1717
self._unadded_refs = {}
1720
bytes_len, chunks = self._compressor.flush().to_chunks()
1721
self._compressor = self._make_group_compressor()
1722
# Note: At this point we still have 1 copy of the fulltext (in
1723
# record and the var 'bytes'), and this generates 2 copies of
1724
# the compressed text (one for bytes, one in chunks)
1725
# TODO: Push 'chunks' down into the _access api, so that we don't
1726
# have to double compressed memory here
1727
# TODO: Figure out how to indicate that we would be happy to free
1728
# the fulltext content at this point. Note that sometimes we
1729
# will want it later (streaming CHK pages), but most of the
1730
# time we won't (everything else)
1731
bytes = ''.join(chunks)
1733
index, start, length = self._access.add_raw_records(
1734
[(None, len(bytes))], bytes)[0]
1736
for key, reads, refs in keys_to_add:
1737
nodes.append((key, "%d %d %s" % (start, length, reads), refs))
1738
self._index.add_records(nodes, random_id=random_id)
1739
self._unadded_refs = {}
1743
max_fulltext_len = 0
1744
max_fulltext_prefix = None
1745
insert_manager = None
1748
# XXX: TODO: remove this, it is just for safety checking for now
1749
inserted_keys = set()
1750
reuse_this_block = reuse_blocks
1751
for record in stream:
1752
# Raise an error when a record is missing.
1753
if record.storage_kind == 'absent':
1754
raise errors.RevisionNotPresent(record.key, self)
1756
if record.key in inserted_keys:
1757
trace.note('Insert claimed random_id=True,'
1758
' but then inserted %r two times', record.key)
1760
inserted_keys.add(record.key)
1762
# If the reuse_blocks flag is set, check to see if we can just
1763
# copy a groupcompress block as-is.
1764
# We only check on the first record (groupcompress-block) not
1765
# on all of the (groupcompress-block-ref) entries.
1766
# The reuse_this_block flag is then kept for as long as
1767
if record.storage_kind == 'groupcompress-block':
1768
# Check to see if we really want to re-use this block
1769
insert_manager = record._manager
1770
reuse_this_block = insert_manager.check_is_well_utilized()
1772
reuse_this_block = False
1773
if reuse_this_block:
1774
# We still want to reuse this block
1775
if record.storage_kind == 'groupcompress-block':
1776
# Insert the raw block into the target repo
1777
insert_manager = record._manager
1778
bytes = record._manager._block.to_bytes()
1779
_, start, length = self._access.add_raw_records(
1780
[(None, len(bytes))], bytes)[0]
1783
block_length = length
1784
if record.storage_kind in ('groupcompress-block',
1785
'groupcompress-block-ref'):
1786
if insert_manager is None:
1787
raise AssertionError('No insert_manager set')
1788
if insert_manager is not record._manager:
1789
raise AssertionError('insert_manager does not match'
1790
' the current record, we cannot be positive'
1791
' that the appropriate content was inserted.'
1793
value = "%d %d %d %d" % (block_start, block_length,
1794
record._start, record._end)
1795
nodes = [(record.key, value, (record.parents,))]
1796
# TODO: Consider buffering up many nodes to be added, not
1797
# sure how much overhead this has, but we're seeing
1798
# ~23s / 120s in add_records calls
1799
self._index.add_records(nodes, random_id=random_id)
1802
bytes = record.get_bytes_as('fulltext')
1803
except errors.UnavailableRepresentation:
1804
adapter_key = record.storage_kind, 'fulltext'
1805
adapter = get_adapter(adapter_key)
1806
bytes = adapter.get_bytes(record)
1807
if len(record.key) > 1:
1808
prefix = record.key[0]
1809
soft = (prefix == last_prefix)
1813
if max_fulltext_len < len(bytes):
1814
max_fulltext_len = len(bytes)
1815
max_fulltext_prefix = prefix
1816
(found_sha1, start_point, end_point,
1817
type) = self._compressor.compress(record.key,
1818
bytes, record.sha1, soft=soft,
1819
nostore_sha=nostore_sha)
1820
# delta_ratio = float(len(bytes)) / (end_point - start_point)
1821
# Check if we want to continue to include that text
1822
if (prefix == max_fulltext_prefix
1823
and end_point < 2 * max_fulltext_len):
1824
# As long as we are on the same file_id, we will fill at least
1825
# 2 * max_fulltext_len
1826
start_new_block = False
1827
elif end_point > 4*1024*1024:
1828
start_new_block = True
1829
elif (prefix is not None and prefix != last_prefix
1830
and end_point > 2*1024*1024):
1831
start_new_block = True
1833
start_new_block = False
1834
last_prefix = prefix
1836
self._compressor.pop_last()
1838
max_fulltext_len = len(bytes)
1839
(found_sha1, start_point, end_point,
1840
type) = self._compressor.compress(record.key, bytes,
1842
if record.key[-1] is None:
1843
key = record.key[:-1] + ('sha1:' + found_sha1,)
1846
self._unadded_refs[key] = record.parents
1848
as_st = static_tuple.StaticTuple.from_sequence
1849
if record.parents is not None:
1850
parents = as_st([as_st(p) for p in record.parents])
1853
refs = static_tuple.StaticTuple(parents)
1854
keys_to_add.append((key, '%d %d' % (start_point, end_point), refs))
1855
if len(keys_to_add):
1857
self._compressor = None
1859
def iter_lines_added_or_present_in_keys(self, keys, pb=None):
1860
"""Iterate over the lines in the versioned files from keys.
1862
This may return lines from other keys. Each item the returned
1863
iterator yields is a tuple of a line and a text version that that line
1864
is present in (not introduced in).
1866
Ordering of results is in whatever order is most suitable for the
1867
underlying storage format.
1869
If a progress bar is supplied, it may be used to indicate progress.
1870
The caller is responsible for cleaning up progress bars (because this
1874
* Lines are normalised by the underlying store: they will all have \n
1876
* Lines are returned in arbitrary order.
1878
:return: An iterator over (line, key).
1882
# we don't care about inclusions, the caller cares.
1883
# but we need to setup a list of records to visit.
1884
# we need key, position, length
1885
for key_idx, record in enumerate(self.get_record_stream(keys,
1886
'unordered', True)):
1887
# XXX: todo - optimise to use less than full texts.
1890
pb.update('Walking content', key_idx, total)
1891
if record.storage_kind == 'absent':
1892
raise errors.RevisionNotPresent(key, self)
1893
lines = osutils.split_lines(record.get_bytes_as('fulltext'))
1897
pb.update('Walking content', total, total)
1900
"""See VersionedFiles.keys."""
1901
if 'evil' in debug.debug_flags:
1902
trace.mutter_callsite(2, "keys scales with size of history")
1903
sources = [self._index] + self._immediate_fallback_vfs
1905
for source in sources:
1906
result.update(source.keys())
1910
class _GCBuildDetails(object):
1911
"""A blob of data about the build details.
1913
This stores the minimal data, which then allows compatibility with the old
1914
api, without taking as much memory.
1917
__slots__ = ('_index', '_group_start', '_group_end', '_basis_end',
1918
'_delta_end', '_parents')
1921
compression_parent = None
1923
def __init__(self, parents, position_info):
1924
self._parents = parents
1925
(self._index, self._group_start, self._group_end, self._basis_end,
1926
self._delta_end) = position_info
1929
return '%s(%s, %s)' % (self.__class__.__name__,
1930
self.index_memo, self._parents)
1933
def index_memo(self):
1934
return (self._index, self._group_start, self._group_end,
1935
self._basis_end, self._delta_end)
1938
def record_details(self):
1939
return static_tuple.StaticTuple(self.method, None)
1941
def __getitem__(self, offset):
1942
"""Compatibility thunk to act like a tuple."""
1944
return self.index_memo
1946
return self.compression_parent # Always None
1948
return self._parents
1950
return self.record_details
1952
raise IndexError('offset out of range')
1958
class _GCGraphIndex(object):
1959
"""Mapper from GroupCompressVersionedFiles needs into GraphIndex storage."""
1961
def __init__(self, graph_index, is_locked, parents=True,
1962
add_callback=None, track_external_parent_refs=False,
1963
inconsistency_fatal=True, track_new_keys=False):
1964
"""Construct a _GCGraphIndex on a graph_index.
1966
:param graph_index: An implementation of bzrlib.index.GraphIndex.
1967
:param is_locked: A callback, returns True if the index is locked and
1969
:param parents: If True, record knits parents, if not do not record
1971
:param add_callback: If not None, allow additions to the index and call
1972
this callback with a list of added GraphIndex nodes:
1973
[(node, value, node_refs), ...]
1974
:param track_external_parent_refs: As keys are added, keep track of the
1975
keys they reference, so that we can query get_missing_parents(),
1977
:param inconsistency_fatal: When asked to add records that are already
1978
present, and the details are inconsistent with the existing
1979
record, raise an exception instead of warning (and skipping the
1982
self._add_callback = add_callback
1983
self._graph_index = graph_index
1984
self._parents = parents
1985
self.has_graph = parents
1986
self._is_locked = is_locked
1987
self._inconsistency_fatal = inconsistency_fatal
1988
# GroupCompress records tend to have the same 'group' start + offset
1989
# repeated over and over, this creates a surplus of ints
1990
self._int_cache = {}
1991
if track_external_parent_refs:
1992
self._key_dependencies = _KeyRefs(
1993
track_new_keys=track_new_keys)
1995
self._key_dependencies = None
1997
def add_records(self, records, random_id=False):
1998
"""Add multiple records to the index.
2000
This function does not insert data into the Immutable GraphIndex
2001
backing the KnitGraphIndex, instead it prepares data for insertion by
2002
the caller and checks that it is safe to insert then calls
2003
self._add_callback with the prepared GraphIndex nodes.
2005
:param records: a list of tuples:
2006
(key, options, access_memo, parents).
2007
:param random_id: If True the ids being added were randomly generated
2008
and no check for existence will be performed.
2010
if not self._add_callback:
2011
raise errors.ReadOnlyError(self)
2012
# we hope there are no repositories with inconsistent parentage
2017
for (key, value, refs) in records:
2018
if not self._parents:
2022
raise errors.KnitCorrupt(self,
2023
"attempt to add node with parents "
2024
"in parentless index.")
2027
keys[key] = (value, refs)
2030
present_nodes = self._get_entries(keys)
2031
for (index, key, value, node_refs) in present_nodes:
2032
# Sometimes these are passed as a list rather than a tuple
2033
node_refs = static_tuple.as_tuples(node_refs)
2034
passed = static_tuple.as_tuples(keys[key])
2035
if node_refs != passed[1]:
2036
details = '%s %s %s' % (key, (value, node_refs), passed)
2037
if self._inconsistency_fatal:
2038
raise errors.KnitCorrupt(self, "inconsistent details"
2039
" in add_records: %s" %
2042
trace.warning("inconsistent details in skipped"
2043
" record: %s", details)
2049
for key, (value, node_refs) in keys.iteritems():
2050
result.append((key, value, node_refs))
2052
for key, (value, node_refs) in keys.iteritems():
2053
result.append((key, value))
2055
key_dependencies = self._key_dependencies
2056
if key_dependencies is not None:
2058
for key, value, refs in records:
2060
key_dependencies.add_references(key, parents)
2062
for key, value, refs in records:
2063
new_keys.add_key(key)
2064
self._add_callback(records)
2066
def _check_read(self):
2067
"""Raise an exception if reads are not permitted."""
2068
if not self._is_locked():
2069
raise errors.ObjectNotLocked(self)
2071
def _check_write_ok(self):
2072
"""Raise an exception if writes are not permitted."""
2073
if not self._is_locked():
2074
raise errors.ObjectNotLocked(self)
2076
def _get_entries(self, keys, check_present=False):
2077
"""Get the entries for keys.
2079
Note: Callers are responsible for checking that the index is locked
2080
before calling this method.
2082
:param keys: An iterable of index key tuples.
2087
for node in self._graph_index.iter_entries(keys):
2089
found_keys.add(node[1])
2091
# adapt parentless index to the rest of the code.
2092
for node in self._graph_index.iter_entries(keys):
2093
yield node[0], node[1], node[2], ()
2094
found_keys.add(node[1])
2096
missing_keys = keys.difference(found_keys)
2098
raise errors.RevisionNotPresent(missing_keys.pop(), self)
2100
def find_ancestry(self, keys):
2101
"""See CombinedGraphIndex.find_ancestry"""
2102
return self._graph_index.find_ancestry(keys, 0)
2104
def get_parent_map(self, keys):
2105
"""Get a map of the parents of keys.
2107
:param keys: The keys to look up parents for.
2108
:return: A mapping from keys to parents. Absent keys are absent from
2112
nodes = self._get_entries(keys)
2116
result[node[1]] = node[3][0]
2119
result[node[1]] = None
2122
def get_missing_parents(self):
2123
"""Return the keys of missing parents."""
2124
# Copied from _KnitGraphIndex.get_missing_parents
2125
# We may have false positives, so filter those out.
2126
self._key_dependencies.satisfy_refs_for_keys(
2127
self.get_parent_map(self._key_dependencies.get_unsatisfied_refs()))
2128
return frozenset(self._key_dependencies.get_unsatisfied_refs())
2130
def get_build_details(self, keys):
2131
"""Get the various build details for keys.
2133
Ghosts are omitted from the result.
2135
:param keys: An iterable of keys.
2136
:return: A dict of key:
2137
(index_memo, compression_parent, parents, record_details).
2139
* index_memo: opaque structure to pass to read_records to extract
2141
* compression_parent: Content that this record is built upon, may
2143
* parents: Logical parents of this node
2144
* record_details: extra information about the content which needs
2145
to be passed to Factory.parse_record
2149
entries = self._get_entries(keys)
2150
for entry in entries:
2152
if not self._parents:
2155
parents = entry[3][0]
2156
details = _GCBuildDetails(parents, self._node_to_position(entry))
2157
result[key] = details
2161
"""Get all the keys in the collection.
2163
The keys are not ordered.
2166
return [node[1] for node in self._graph_index.iter_all_entries()]
2168
def _node_to_position(self, node):
2169
"""Convert an index value to position details."""
2170
bits = node[2].split(' ')
2171
# It would be nice not to read the entire gzip.
2172
# start and stop are put into _int_cache because they are very common.
2173
# They define the 'group' that an entry is in, and many groups can have
2174
# thousands of objects.
2175
# Branching Launchpad, for example, saves ~600k integers, at 12 bytes
2176
# each, or about 7MB. Note that it might be even more when you consider
2177
# how PyInt is allocated in separate slabs. And you can't return a slab
2178
# to the OS if even 1 int on it is in use. Note though that Python uses
2179
# a LIFO when re-using PyInt slots, which might cause more
2181
start = int(bits[0])
2182
start = self._int_cache.setdefault(start, start)
2184
stop = self._int_cache.setdefault(stop, stop)
2185
basis_end = int(bits[2])
2186
delta_end = int(bits[3])
2187
# We can't use StaticTuple here, because node[0] is a BTreeGraphIndex
2189
return (node[0], start, stop, basis_end, delta_end)
2191
def scan_unvalidated_index(self, graph_index):
2192
"""Inform this _GCGraphIndex that there is an unvalidated index.
2194
This allows this _GCGraphIndex to keep track of any missing
2195
compression parents we may want to have filled in to make those
2196
indices valid. It also allows _GCGraphIndex to track any new keys.
2198
:param graph_index: A GraphIndex
2200
key_dependencies = self._key_dependencies
2201
if key_dependencies is None:
2203
for node in graph_index.iter_all_entries():
2204
# Add parent refs from graph_index (and discard parent refs
2205
# that the graph_index has).
2206
key_dependencies.add_references(node[1], node[3][0])
2209
from bzrlib._groupcompress_py import (
2211
apply_delta_to_source,
2214
decode_copy_instruction,
2218
from bzrlib._groupcompress_pyx import (
2220
apply_delta_to_source,
2225
GroupCompressor = PyrexGroupCompressor
2226
except ImportError, e:
2227
osutils.failed_to_load_extension(e)
2228
GroupCompressor = PythonGroupCompressor
added
removed
bzrlib/groupcompress.py
