~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

#

# You should have received a copy of the GNU General Public License

# along with this program; if not, write to the Free Software

 

import re

import sys

import time

 

from bzrlib import (

    debug,

    graph,

    osutils,

    )

from bzrlib.index import (

    CombinedGraphIndex,

    GraphIndexPrefixAdapter,

    )

from bzrlib.knit import (

    KnitPlainFactory,

    errors,

    lockable_files,

    lockdir,

    )

 

from bzrlib.btree_index import (

    BTreeGraphIndex,

    BTreeBuilder,

    MetaDirRepositoryFormat,

    RepositoryFormat,

    RootCommitBuilder,

    )

from bzrlib.trace import (

    mutter,

    warning,

    )

 

    # A map of index 'type' to the file extension and position in the

    # index_sizes array.

    index_definitions = {

        'revision': ('.rix', 0),

        'inventory': ('.iix', 1),

        'text': ('.tix', 2),

        }

 

    def __init__(self, revision_index, inventory_index, text_index,

        """Create a pack instance.

 

        :param revision_index: A GraphIndex for determining what revisions are

            texts/deltas (via (fileid, revisionid) tuples).

        :param signature_index: A GraphIndex for determining what signatures are

            present in the Pack and accessing the locations of their texts.

        """

        self.revision_index = revision_index

        self.inventory_index = inventory_index

        self.text_index = text_index

        self.signature_index = signature_index

 

    def access_tuple(self):

        """Return a tuple (transport, name) for the pack content."""

        return self.index_name('text', name)

 

    def _replace_index_with_readonly(self, index_type):

        setattr(self, index_type + '_index',

            self.index_class(self.index_transport,

                self.index_name(index_type, self.name),

 

 

class ExistingPack(Pack):

    """An in memory proxy for an existing .pack and its disk indices."""

 

    def __init__(self, pack_transport, name, revision_index, inventory_index,

        """Create an ExistingPack object.

 

        :param pack_transport: The transport where the pack file resides.

        :param name: The name of the pack on disk in the pack_transport.

        """

        Pack.__init__(self, revision_index, inventory_index, text_index,

        self.name = name

        self.pack_transport = pack_transport

        if None in (revision_index, inventory_index, text_index,

 

    def __init__(self, name, revision_index, inventory_index, text_index,

        signature_index, upload_transport, pack_transport, index_transport,

        """Create a ResumedPack object."""

        ExistingPack.__init__(self, pack_transport, name, revision_index,

        self.upload_transport = upload_transport

        self.index_transport = index_transport

        self.index_sizes = [None, None, None, None]

            ('text', text_index),

            ('signature', signature_index),

            ]

        for index_type, index in indices:

            offset = self.index_offset(index_type)

            self.index_sizes[offset] = index._size

        self.upload_transport.delete(self.file_name())

        indices = [self.revision_index, self.inventory_index, self.text_index,

            self.signature_index]

        for index in indices:

            index._transport.delete(index._name)

 

    def finish(self):

        self._check_references()

            old_name = self.index_name(index_type, self.name)

            new_name = '../indices/' + old_name

            self.upload_transport.rename(old_name, new_name)

            self._replace_index_with_readonly(index_type)

        self._state = 'finished'

 

    def _get_external_refs(self, index):

        return index.external_references(1)

 

 

        # The relative locations of the packs are constrained, but all are

        # passed in because the caller has them, so as to avoid object churn.

        index_builder_class = pack_collection._index_builder_class

        Pack.__init__(self,

            # Revisions: parents list, no text compression.

            index_builder_class(reference_lists=1),

            # Signatures: Just blobs to store, no compression, no parents

            # listing.

            index_builder_class(reference_lists=0),

            )

        self._pack_collection = pack_collection

        # When we make readonly indices, we need this.

        self._file_mode = file_mode

        # tracks the content written to the .pack file.

        self._hash = osutils.md5()

        self.index_sizes = None

        # How much data to cache when writing packs. Note that this is not

        # synchronised with reads, because it's not in the transport layer, so

        self._writer.begin()

        # what state is the pack in? (open, finished, aborted)

        self._state = 'open'

 

    def abort(self):

        """Cancel creating this pack."""

        return bool(self.get_revision_count() or

            self.inventory_index.key_count() or

            self.text_index.key_count() or

 

    def finish(self, suspend=False):

        """Finish the new pack.

         - stores the index size tuple for the pack in the index_sizes

           attribute.

        """

        if not suspend:

            self._check_references()

        # write indices

        self._write_index('text', self.text_index, 'file texts', suspend)

        self._write_index('signature', self.signature_index,

            'revision signatures', suspend)

        self.write_stream.close()

        # Note that this will clobber an existing pack with the same name,

        # without checking for hash collisions. While this is undesirable this

    # XXX: Probably 'can be written to' could/should be separated from 'acts

    # like a knit index' -- mbp 20071024

 

        """Create an AggregateIndex.

 

        :param reload_func: A function to call if we find we are missing an

        self.index_to_pack = {}

        self.combined_index = CombinedGraphIndex([], reload_func=reload_func)

        self.data_access = _DirectPackAccess(self.index_to_pack,

        self.add_callback = None

 

    def replace_indices(self, index_to_pack, indices):

 

    def open_pack(self):

        """Open a pack for the pack we are creating."""

                file_mode=self._pack_collection.repo.bzrdir._get_file_mode())

 

    def _update_pack_order(self, entries, index_to_pack_map):

        """Determine how we want our packs to be ordered.

            if missing_text_keys:

                # TODO: raise a specific error that can handle many missing

                # keys.

                a_missing_key = missing_text_keys.pop()

                raise errors.RevisionNotPresent(a_missing_key[1],

                    a_missing_key[0])

                time.ctime(), self._pack_collection._upload_transport.base, new_pack.random_name,

                new_pack.signature_index.key_count(),

                time.time() - new_pack.start_time)

        new_pack._check_references()

        if not self._use_pack(new_pack):

            new_pack.abort()

        self._pack_collection.allocate(new_pack)

        return new_pack

 

        pb = ui.ui_factory.nested_progress_bar()

        try:

            return self._do_copy_nodes(nodes, index_map, writer,

        finally:

            pb.finished()

 

        # for record verification

        knit = KnitVersionedFiles(None, None)

        # plan a readv on each source pack:

                izip(reader.iter_records(), pack_readv_requests):

                raw_data = read_func(None)

                # check the header only

                pos, size = writer.add_bytes_record(raw_data, names)

                write_index.add_node(key, eol_flag + "%d %d" % (pos, size))

                pb.update("Copied record", record_index)

            iterator is a tuple with:

            index, readv_vector, node_vector. readv_vector is a list ready to

            hand to the transport readv method, and node_vector is a list of

            matching readv_vector.

        """

        # group by pack so we do one readv per pack

        # space (we only topo sort the revisions, which is smaller).

        topo_order = tsort.topo_sort(ancestors)

        rev_order = dict(zip(topo_order, range(len(topo_order))))

        transaction = repo.get_transaction()

        file_id_index = GraphIndexPrefixAdapter(

            self.new_pack.text_index,

    :ivar _names: map of {pack_name: (index_size,)}

    """

 

    def __init__(self, repo, transport, index_transport, upload_transport,

        """Create a new RepositoryPackCollection.

 

        :param transport: Addresses the repository base directory

        :param pack_transport: Addresses the directory of existing complete packs.

        :param index_builder_class: The index builder class to use.

        :param index_class: The index class to use.

        """

        self.repo = repo

        self.transport = transport

        self._index_transport = index_transport

        self._pack_transport = pack_transport

        self._index_builder_class = index_builder_class

        self._index_class = index_class

        self.packs = []

        # name:Pack mapping

        self._packs_by_name = {}

        # the previous pack-names content

        self._packs_at_load = None

        # when a pack is being created by this object, the state of that pack.

        self._new_pack = None

        # aggregated revision index data

        # resumed packs

        self._resumed_packs = []

 

    def add_pack_to_memory(self, pack):

        """Make a Pack object available to the repository to satisfy queries.

 

        self.inventory_index.add_index(pack.inventory_index, pack)

        self.text_index.add_index(pack.text_index, pack)

        self.signature_index.add_index(pack.signature_index, pack)

 

    def all_packs(self):

        """Return a list of all the Pack objects this repository has.

        in synchronisation with certain steps. Otherwise the names collection

        is not flushed.

 

        """

        while True:

            try:

                return self._do_autopack()

                # If we get a RetryAutopack exception, we should abort the

                # current action, and retry.

                pass

        total_revisions = self.revision_index.combined_index.key_count()

        total_packs = len(self._names)

        if self._max_pack_count(total_revisions) >= total_packs:

        # determine which packs need changing

        pack_distribution = self.pack_distribution(total_revisions)

        existing_packs = []

            'containing %d revisions. Packing %d files into %d affecting %d'

            ' revisions', self, total_packs, total_revisions, num_old_packs,

            num_new_packs, num_revs_affected)

                                      reload_func=self._restart_autopack)

 

    def _execute_pack_operations(self, pack_operations, _packer_class=Packer,

                                 reload_func=None):

 

        :param pack_operations: A list of [revision_count, packs_to_combine].

        :param _packer_class: The class of packer to use (default: Packer).

        """

        for revision_count, packs in pack_operations:

            # we may have no-ops from the setup logic

                self._remove_pack_from_memory(pack)

        # record the newly available packs and stop advertising the old

        # packs

 

    def lock_names(self):

        """Acquire the mutex around the pack-names index.

        """

        self.repo.control_files.lock_write()

 

        """Pack the pack collection totally."""

        self.ensure_loaded()

        total_packs = len(self._names)

            return

        total_revisions = self.revision_index.combined_index.key_count()

        # XXX: the following may want to be a class, to pack with a given

        # policy.

        mutter('Packing repository %s, which has %d pack files, '

        # determine which packs need changing

        pack_operations = [[0, []]]

        for pack in self.all_packs():

        self._execute_pack_operations(pack_operations, OptimisingPacker)

 

    def plan_autopack_combinations(self, existing_packs, pack_distribution):

        return [[final_rev_count, final_pack_list]]

 

    def ensure_loaded(self):

        # NB: if you see an assertion error here, its probably access against

        # an unlocked repo. Naughty.

        if not self.repo.is_locked():

                name = key[0]

                self._names[name] = self._parse_index_sizes(value)

                self._packs_at_load.add((key, value))

        # populate all the metadata.

        self.all_packs()

 

    def _parse_index_sizes(self, value):

        """Parse a string of index sizes."""

            inv_index = self._make_index(name, '.iix')

            txt_index = self._make_index(name, '.tix')

            sig_index = self._make_index(name, '.six')

            result = ExistingPack(self._pack_transport, name, rev_index,

            self.add_pack_to_memory(result)

            return result

 

            inv_index = self._make_index(name, '.iix', resume=True)

            txt_index = self._make_index(name, '.tix', resume=True)

            sig_index = self._make_index(name, '.six', resume=True)

        except errors.NoSuchFile, e:

            raise errors.UnresumableWriteGroup(self.repo, [name], str(e))

        self.add_pack_to_memory(result)

        return self._index_class(self.transport, 'pack-names', None

                ).iter_all_entries()

 

        size_offset = self._suffix_offsets[suffix]

        index_name = name + suffix

        if resume:

        else:

            transport = self._index_transport

            index_size = self._names[name][size_offset]

 

    def _max_pack_count(self, total_revisions):

        """Return the maximum number of packs to use for total revisions.

        :param return: None.

        """

        for pack in packs:

            # TODO: Probably needs to know all possible indices for this pack

            # - or maybe list the directory and move all indices matching this

            # name whether we recognize it or not?

 

    def pack_distribution(self, total_revisions):

        """Generate a list of the number of revisions to put in each pack.

        self.inventory_index.remove_index(pack.inventory_index, pack)

        self.text_index.remove_index(pack.text_index, pack)

        self.signature_index.remove_index(pack.signature_index, pack)

 

    def reset(self):

        """Clear all cached data."""

        # cached revision data

        self.revision_index.clear()

        # cached signature data

        self.signature_index.clear()

        # cached file text data

        self.text_index.clear()

        # cached inventory data

        self.inventory_index.clear()

        # remove the open pack

        self._new_pack = None

        # information about packs.

        disk_nodes = set()

        for index, key, value in self._iter_disk_pack_index():

            disk_nodes.add((key, value))

 

        # do a two-way diff against our original content

        current_nodes = set()

        disk_nodes.difference_update(deleted_nodes)

        disk_nodes.update(new_nodes)

 

 

    def _syncronize_pack_names_from_disk_nodes(self, disk_nodes):

        """Given the correct set of pack files, update our saved info.

                added.append(name)

        return removed, added, modified

 

        """Save the list of packs.

 

        This will take out the mutex around the pack names list for the

 

        :param clear_obsolete_packs: If True, clear out the contents of the

            obsolete_packs directory.

        """

        self.lock_names()

        try:

            builder = self._index_builder_class()

            # TODO: handle same-name, index-size-changes here -

            # e.g. use the value from disk, not ours, *unless* we're the one

            # changing it.

                builder.add_node(key, value)

            self.transport.put_file('pack-names', builder.finish(),

                mode=self.repo.bzrdir._get_file_mode())

            self._packs_at_load = disk_nodes

            if clear_obsolete_packs:

        finally:

            self._unlock_names()

        # synchronise the memory packs list with what we just wrote:

        self._syncronize_pack_names_from_disk_nodes(disk_nodes)

 

    def reload_pack_names(self):

        """Sync our pack listing with what is present in the repository.

        This should be called when we find out that something we thought was

        present is now missing. This happens when another process re-packs the

        repository, etc.

        """

        # out the new value.

        (removed, added,

         modified) = self._syncronize_pack_names_from_disk_nodes(disk_nodes)

        if removed or added or modified:

            raise

        raise errors.RetryAutopack(self.repo, False, sys.exc_info())

 

        """Delete everything from the obsolete-packs directory.

        """

        obsolete_pack_transport = self.transport.clone('obsolete_packs')

        for filename in obsolete_pack_transport.list_dir('.'):

            try:

                obsolete_pack_transport.delete(filename)

            except (errors.PathError, errors.TransportError), e:

 

    def _start_write_group(self):

        # Do not permit preparation for writing if we're not in a 'write lock'.

        if not self.repo.is_write_locked():

            raise errors.NotWriteLocked(self)

            file_mode=self.repo.bzrdir._get_file_mode())

        # allow writing: queue writes to a new index

        self.revision_index.add_writable_index(self._new_pack.revision_index,

            self._new_pack)

        self.text_index.add_writable_index(self._new_pack.text_index,

            self._new_pack)

        self.signature_index.add_writable_index(self._new_pack.signature_index,

            self._new_pack)

 

        self.repo.inventories._index._add_callback = self.inventory_index.add_callback

        self.repo.revisions._index._add_callback = self.revision_index.add_callback

                except KeyError:

                    pass

        del self._resumed_packs[:]

 

    def _remove_resumed_pack_indices(self):

        for resumed_pack in self._resumed_packs:

            self._remove_pack_indices(resumed_pack)

        del self._resumed_packs[:]