~bzr-pqm/bzr/bzr.dev

#

# This program is free software; you can redistribute it and/or modify

# it under the terms of the GNU General Public License as published by

except ImportError:

    pylzma = None

 

from bzrlib import (

    annotate,

    debug,

    errors,

    graph as _mod_graph,

    osutils,

    pack,

    trace,

    )

from bzrlib.btree_index import BTreeBuilder

from bzrlib.lru_cache import LRUSizeCache

from bzrlib.versionedfile import (

    adapter_registry,

    AbsentContentFactory,

    ChunkedContentFactory,

    FulltextContentFactory,

    )

 

# Minimum number of uncompressed bytes to try fetch at once when retrieving

 

    present_keys = []

    for prefix in sorted(per_prefix_map):

    return present_keys

 

 

    def __init__(self):

        # map by key? or just order in file?

        self._compressor_name = None

        self._z_content_decompressor = None

        self._z_content_length = None

        self._content_length = None

        :param num_bytes: Ensure that we have extracted at least num_bytes of

            content. If None, consume everything

        """

        if num_bytes is None:

            num_bytes = self._content_length

        elif (self._content_length is not None

                self._content = ''.join(self._content_chunks)

                self._content_chunks = None

        if self._content is None:

                raise AssertionError('No content to decompress')

                self._content = ''

            elif self._compressor_name == 'lzma':

                # We don't do partial lzma decomp yet

            elif self._compressor_name == 'zlib':

                # Start a zlib decompressor

                else:

                    self._z_content_decompressor = zlib.decompressobj()

                    # Seed the decompressor with the uncompressed bytes, so

                    # that the rest of the code is simplified

                    self._content = self._z_content_decompressor.decompress(

            else:

                raise AssertionError('Unknown compressor: %r'

                                     % self._compressor_name)

        # 'unconsumed_tail'

 

        # Do we have enough bytes already?

            return

        # If we got this far, and don't have a decompressor, something is wrong

        if self._z_content_decompressor is None:

            raise AssertionError(

                'No decompressor to decompress %d bytes' % num_bytes)

        remaining_decomp = self._z_content_decompressor.unconsumed_tail

 

    def _parse_bytes(self, bytes, pos):

        """Read the various lengths from the header.

            # XXX: Define some GCCorrupt error ?

            raise AssertionError('Invalid bytes: (%d) != %d + %d' %

                                 (len(bytes), pos, self._z_content_length))

 

    @classmethod

    def from_bytes(cls, bytes):

        self._content_length = length

        self._content_chunks = content_chunks

        self._content = None

 

    def set_content(self, content):

        """Set the content of this block."""

        self._content_length = len(content)

        self._content = content

 

    def _create_z_content_using_lzma(self):

        if self._content_chunks is not None:

            self._content_chunks = None

        if self._content is None:

            raise AssertionError('Nothing to compress')

 

        compressor = zlib.compressobj(zlib.Z_DEFAULT_COMPRESSION)

        compressed_chunks.append(compressor.flush())

 

    def _create_z_content(self):

            return

        if _USE_LZMA:

            self._create_z_content_using_lzma()

            return

        if self._content_chunks is not None:

 

        self._create_z_content()

        if _USE_LZMA:

            header = self.GCB_LZ_HEADER

        else:

            header = self.GCB_HEADER

                 ]

        return ''.join(chunks)

 

    def _dump(self, include_text=False):

                # Grab and cache the raw bytes for this entry

                # and break the ref-cycle with _manager since we don't need it

                # anymore

                block = self._manager._block

                self._bytes = block.extract(self.key, self._start, self._end)

                # There are code paths that first extract as fulltext, and then

    _full_enough_block_size = 3*1024*1024 # size at which we won't repack

    _full_enough_mixed_block_size = 2*768*1024 # 1.5MB

 

        self._block = block

        # We need to preserve the ordering

        self._factories = []

        self._last_byte = 0

 

    def add_factory(self, key, parents, start, end):

        if not self._factories:

        new_block.set_content(self._block._content[:last_byte])

        self._block = new_block

 

    def _rebuild_block(self):

        """Create a new GroupCompressBlock with only the referenced texts."""

        tstart = time.time()

        old_length = self._block._content_length

        end_point = 0

        #       block? It seems hard to come up with a method that it would

        #       expand, since we do full compression again. Perhaps based on a

        #       request that ends up poorly ordered?

        delta = time.time() - tstart

        self._block = new_block

        trace.mutter('creating new compressed block on-the-fly in %.3fs'

        z_header_bytes = zlib.compress(header_bytes)

        del header_bytes

        z_header_bytes_len = len(z_header_bytes)

        lines.append('%d\n%d\n%d\n' % (z_header_bytes_len, header_bytes_len,

        lines.append(z_header_bytes)

        return ''.join(lines)

 

    @classmethod

    def from_bytes(cls, bytes):

        # TODO: This does extra string copying, probably better to do it a

        (storage_kind, z_header_len, header_len,

         block_len, rest) = bytes.split('\n', 4)

        del bytes

 

class _CommonGroupCompressor(object):

 

        """Create a GroupCompressor."""

        self.chunks = []

        self._last = None

        self.labels_deltas = {}

        self._delta_index = None # Set by the children

        self._block = GroupCompressBlock()

 

    def compress(self, key, bytes, expected_sha, nostore_sha=None, soft=False):

        """Compress lines with label key.

 

        After calling this, the compressor should no longer be used

        """

        self._block.set_chunked_content(self.chunks, self.endpoint)

        self.chunks = None

        self._delta_index = None

 

class PythonGroupCompressor(_CommonGroupCompressor):

 

        """Create a GroupCompressor.

 

        Used only if the pyrex version is not available.

        """

        self._delta_index = LinesDeltaIndex([])

        # The actual content is managed by LinesDeltaIndex

        self.chunks = self._delta_index.lines

 

    It contains code very similar to SequenceMatcher because of having a similar

    task. However some key differences apply:

    """

 

 

    def _compress(self, key, bytes, max_delta_size, soft=False):

        """see _CommonGroupCompressor._compress"""

        index = _GCGraphIndex(graph_index, lambda:True, parents=parents,

            add_callback=graph_index.add_nodes,

            inconsistency_fatal=inconsistency_fatal)

        access.set_writer(writer, graph_index, (transport, 'newpack'))

        result = GroupCompressVersionedFiles(index, access, delta)

        result.stream = stream

 

class _BatchingBlockFetcher(object):

    """Fetch group compress blocks in batches.

    :ivar total_bytes: int of expected number of bytes needed to fetch the

        currently pending batch.

    """

 

        self.gcvf = gcvf

        self.locations = locations

        self.keys = []

        self.total_bytes = 0

        self.last_read_memo = None

        self.manager = None

 

    def add_key(self, key):

        """Add another to key to fetch.

        :return: The estimated number of bytes needed to fetch the batch so

            far.

        """

            # and then.

            self.batch_memos[read_memo] = cached_block

        return self.total_bytes

    def _flush_manager(self):

        if self.manager is not None:

            for factory in self.manager.get_record_stream():

    def yield_factories(self, full_flush=False):

        """Yield factories for keys added since the last yield.  They will be

        returned in the order they were added via add_key.

        :param full_flush: by default, some results may not be returned in case

            they can be part of the next batch.  If full_flush is True, then

            all results are returned.

                    memos_to_get_stack.pop()

                else:

                    block = self.batch_memos[read_memo]

                self.last_read_memo = read_memo

            start, end = index_memo[3:5]

            self.manager.add_factory(key, parents, start, end)

        self.total_bytes = 0

 

 

    """A group-compress based VersionedFiles implementation."""

 

        """Create a GroupCompressVersionedFiles object.

 

        :param index: The index object storing access and graph data.

        :param access: The access object storing raw data.

        :param delta: Whether to delta compress or just entropy compress.

        :param _unadded_refs: private parameter, don't use.

        """

        self._index = index

        self._access = access

        if _unadded_refs is None:

            _unadded_refs = {}

        self._unadded_refs = _unadded_refs

 

    def without_fallbacks(self):

        """Return a clone of this object without any fallbacks configured."""

        return GroupCompressVersionedFiles(self._index, self._access,

 

    def add_lines(self, key, parents, lines, parent_texts=None,

        left_matching_blocks=None, nostore_sha=None, random_id=False,

        :param key: The key tuple of the text to add.

        :param parents: The parents key tuples of the text to add.

        :param lines: A list of lines. Each line must be a bytestring. And all

            correctly is expensive and extremely unlikely to catch bugs so it

            is not done at runtime unless check_content is True.)

        :param parent_texts: An optional dictionary containing the opaque

 

        :param a_versioned_files: A VersionedFiles object.

        """

 

    def annotate(self, key):

        """See VersionedFiles.annotate."""

        else:

            return self.get_record_stream(keys, 'unordered', True)

 

    def _check_add(self, key, lines, random_id, check_content):

        """check that version_id and lines are safe to add."""

        version_id = key[-1]

            self._check_lines_not_unicode(lines)

            self._check_lines_are_lines(lines)

 

    def get_parent_map(self, keys):

        """Get a map of the graph parents of keys.

 

            and so on.

        """

        result = {}

        source_results = []

        missing = set(keys)

        for source in sources:

        parent_map = {}

        key_to_source_map = {}

        source_results = []

            if not missing:

                break

            source_parents = source.get_parent_map(missing)

 

        The returned objects should be in the order defined by 'ordering',

        which can weave between different sources.

        :param ordering: Must be one of 'topological' or 'groupcompress'

        :return: List of [(source, [keys])] tuples, such that all keys are in

            the defined order, regardless of source.

        """

        if ordering == 'topological':

        else:

            # ordering == 'groupcompress'

            # XXX: This only optimizes for the target ordering. We may need

        #  - we encounter an unadded ref, or

        #  - we run out of keys, or

        #  - the total bytes to retrieve for this batch > BATCH_SIZE

        for source, keys in source_keys:

            if source is self:

                for key in keys:

        for _ in self._insert_record_stream(stream, random_id=False):

            pass

 

    def _insert_record_stream(self, stream, random_id=False, nostore_sha=None,

                              reuse_blocks=True):

        """Internal core to insert a record stream into this container.

                return adapter

        # This will go up to fulltexts for gc to gc fetching, which isn't

        # ideal.

        self._unadded_refs = {}

        keys_to_add = []

        def flush():

            index, start, length = self._access.add_raw_records(

                [(None, len(bytes))], bytes)[0]

            nodes = []

            self._index.add_records(nodes, random_id=random_id)

            self._unadded_refs = {}

            del keys_to_add[:]

 

        last_prefix = None

        max_fulltext_len = 0

                key = record.key

            self._unadded_refs[key] = record.parents

            yield found_sha1

        if len(keys_to_add):

            flush()

        self._compressor = None

        """See VersionedFiles.keys."""

        if 'evil' in debug.debug_flags:

            trace.mutter_callsite(2, "keys scales with size of history")

        result = set()

        for source in sources:

            result.update(source.keys())

        return result

 

 

class _GCGraphIndex(object):

    """Mapper from GroupCompressVersionedFiles needs into GraphIndex storage."""

 

        self.has_graph = parents

        self._is_locked = is_locked

        self._inconsistency_fatal = inconsistency_fatal

        if track_external_parent_refs:

                track_new_keys=track_new_keys)

        else:

            self._key_dependencies = None

        if not random_id:

            present_nodes = self._get_entries(keys)

            for (index, key, value, node_refs) in present_nodes:

                    if self._inconsistency_fatal:

                        raise errors.KnitCorrupt(self, "inconsistent details"

                                                 " in add_records: %s" %

        :param keys: An iterable of keys.

        :return: A dict of key:

            (index_memo, compression_parent, parents, record_details).