← Back to branch summary

~bzr-pqm/bzr/bzr.dev

« back to all changes in this revision

Viewing changes to bzrlib/smart/repository.py

  • Committer: Canonical.com Patch Queue Manager
  • Date: 2009-06-18 20:25:52 UTC
  • mfrom: (4413.5.15 1.16-chk-direct)
  • Revision ID: pqm@pqm.ubuntu.com-20090618202552-xyl6tcvbxtm8bupf
(jam) Improve initial commit performance by creating a CHKMap in bulk,
        rather than via O(tree) map() calls.

Show diffs side-by-side

added added

removed removed

Lines of Context:
bzrlib/smart/repository.py
1
 
# Copyright (C) 2006-2010 Canonical Ltd
 
1
# Copyright (C) 2006, 2007 Canonical Ltd
2
2
#
3
3
# This program is free software; you can redistribute it and/or modify
4
4
# it under the terms of the GNU General Public License as published by
20
20
import os
21
21
import Queue
22
22
import sys
 
23
import tarfile
23
24
import tempfile
24
25
import threading
25
26
 
29
30
    graph,
30
31
    osutils,
31
32
    pack,
32
 
    ui,
33
 
    versionedfile,
34
33
    )
35
34
from bzrlib.bzrdir import BzrDir
36
35
from bzrlib.smart.request import (
39
38
    SuccessfulSmartServerResponse,
40
39
    )
41
40
from bzrlib.repository import _strip_NULL_ghosts, network_format_registry
42
 
from bzrlib.recordcounter import RecordCounter
43
41
from bzrlib import revision as _mod_revision
44
 
from bzrlib.versionedfile import (
45
 
    NetworkRecordStream,
46
 
    record_to_fulltext_bytes,
47
 
    )
 
42
from bzrlib.versionedfile import NetworkRecordStream, record_to_fulltext_bytes
48
43
 
49
44
 
50
45
class SmartServerRepositoryRequest(SmartServerRequest):
393
388
        if token == '':
394
389
            token = None
395
390
        try:
396
 
            token = repository.lock_write(token=token).repository_token
 
391
            token = repository.lock_write(token=token)
397
392
        except errors.LockContention, e:
398
393
            return FailedSmartServerResponse(('LockContention',))
399
394
        except errors.UnlockableTransport:
419
414
            repository.
420
415
        """
421
416
        self._to_format = network_format_registry.get(to_network_name)
422
 
        if self._should_fake_unknown():
423
 
            return FailedSmartServerResponse(
424
 
                ('UnknownMethod', 'Repository.get_stream'))
425
417
        return None # Signal that we want a body.
426
418
 
427
 
    def _should_fake_unknown(self):
428
 
        """Return True if we should return UnknownMethod to the client.
429
 
        
430
 
        This is a workaround for bugs in pre-1.19 clients that claim to
431
 
        support receiving streams of CHK repositories.  The pre-1.19 client
432
 
        expects inventory records to be serialized in the format defined by
433
 
        to_network_name, but in pre-1.19 (at least) that format definition
434
 
        tries to use the xml5 serializer, which does not correctly handle
435
 
        rich-roots.  After 1.19 the client can also accept inventory-deltas
436
 
        (which avoids this issue), and those clients will use the
437
 
        Repository.get_stream_1.19 verb instead of this one.
438
 
        So: if this repository is CHK, and the to_format doesn't match,
439
 
        we should just fake an UnknownSmartMethod error so that the client
440
 
        will fallback to VFS, rather than sending it a stream we know it
441
 
        cannot handle.
442
 
        """
443
 
        from_format = self._repository._format
444
 
        to_format = self._to_format
445
 
        if not from_format.supports_chks:
446
 
            # Source not CHK: that's ok
447
 
            return False
448
 
        if (to_format.supports_chks and
449
 
            from_format.repository_class is to_format.repository_class and
450
 
            from_format._serializer == to_format._serializer):
451
 
            # Source is CHK, but target matches: that's ok
452
 
            # (e.g. 2a->2a, or CHK2->2a)
453
 
            return False
454
 
        # Source is CHK, and target is not CHK or incompatible CHK.  We can't
455
 
        # generate a compatible stream.
456
 
        return True
457
 
 
458
419
    def do_body(self, body_bytes):
459
420
        repository = self._repository
460
421
        repository.lock_read()
490
451
            repository.unlock()
491
452
 
492
453
 
493
 
class SmartServerRepositoryGetStream_1_19(SmartServerRepositoryGetStream):
494
 
 
495
 
    def _should_fake_unknown(self):
496
 
        """Returns False; we don't need to workaround bugs in 1.19+ clients."""
497
 
        return False
498
 
 
499
 
 
500
454
def _stream_to_byte_stream(stream, src_format):
501
455
    """Convert a record stream to a self delimited byte stream."""
502
456
    pack_writer = pack.ContainerSerialiser()
506
460
        for record in substream:
507
461
            if record.storage_kind in ('chunked', 'fulltext'):
508
462
                serialised = record_to_fulltext_bytes(record)
509
 
            elif record.storage_kind == 'inventory-delta':
510
 
                serialised = record_to_inventory_delta_bytes(record)
511
463
            elif record.storage_kind == 'absent':
512
464
                raise ValueError("Absent factory for %s" % (record.key,))
513
465
            else:
520
472
    yield pack_writer.end()
521
473
 
522
474
 
523
 
class _ByteStreamDecoder(object):
524
 
    """Helper for _byte_stream_to_stream.
525
 
 
526
 
    The expected usage of this class is via the function _byte_stream_to_stream
527
 
    which creates a _ByteStreamDecoder, pops off the stream format and then
528
 
    yields the output of record_stream(), the main entry point to
529
 
    _ByteStreamDecoder.
530
 
 
531
 
    Broadly this class has to unwrap two layers of iterators:
532
 
    (type, substream)
533
 
    (substream details)
534
 
 
535
 
    This is complicated by wishing to return type, iterator_for_type, but
536
 
    getting the data for iterator_for_type when we find out type: we can't
537
 
    simply pass a generator down to the NetworkRecordStream parser, instead
538
 
    we have a little local state to seed each NetworkRecordStream instance,
539
 
    and gather the type that we'll be yielding.
540
 
 
541
 
    :ivar byte_stream: The byte stream being decoded.
542
 
    :ivar stream_decoder: A pack parser used to decode the bytestream
543
 
    :ivar current_type: The current type, used to join adjacent records of the
544
 
        same type into a single stream.
545
 
    :ivar first_bytes: The first bytes to give the next NetworkRecordStream.
 
475
def _byte_stream_to_stream(byte_stream):
 
476
    """Convert a byte stream into a format and a stream.
 
477
 
 
478
    :param byte_stream: A bytes iterator, as output by _stream_to_byte_stream.
 
479
    :return: (RepositoryFormat, stream_generator)
546
480
    """
547
 
 
548
 
    def __init__(self, byte_stream, record_counter):
549
 
        """Create a _ByteStreamDecoder."""
550
 
        self.stream_decoder = pack.ContainerPushParser()
551
 
        self.current_type = None
552
 
        self.first_bytes = None
553
 
        self.byte_stream = byte_stream
554
 
        self._record_counter = record_counter
555
 
        self.key_count = 0
556
 
 
557
 
    def iter_stream_decoder(self):
558
 
        """Iterate the contents of the pack from stream_decoder."""
559
 
        # dequeue pending items
560
 
        for record in self.stream_decoder.read_pending_records():
561
 
            yield record
562
 
        # Pull bytes of the wire, decode them to records, yield those records.
563
 
        for bytes in self.byte_stream:
564
 
            self.stream_decoder.accept_bytes(bytes)
565
 
            for record in self.stream_decoder.read_pending_records():
566
 
                yield record
567
 
 
568
 
    def iter_substream_bytes(self):
569
 
        if self.first_bytes is not None:
570
 
            yield self.first_bytes
571
 
            # If we run out of pack records, single the outer layer to stop.
572
 
            self.first_bytes = None
573
 
        for record in self.iter_pack_records:
 
481
    stream_decoder = pack.ContainerPushParser()
 
482
    def record_stream():
 
483
        """Closure to return the substreams."""
 
484
        # May have fully parsed records already.
 
485
        for record in stream_decoder.read_pending_records():
574
486
            record_names, record_bytes = record
575
487
            record_name, = record_names
576
488
            substream_type = record_name[0]
577
 
            if substream_type != self.current_type:
578
 
                # end of a substream, seed the next substream.
579
 
                self.current_type = substream_type
580
 
                self.first_bytes = record_bytes
581
 
                return
582
 
            yield record_bytes
583
 
 
584
 
    def record_stream(self):
585
 
        """Yield substream_type, substream from the byte stream."""
586
 
        def wrap_and_count(pb, rc, substream):
587
 
            """Yield records from stream while showing progress."""
588
 
            counter = 0
589
 
            if rc:
590
 
                if self.current_type != 'revisions' and self.key_count != 0:
591
 
                    # As we know the number of revisions now (in self.key_count)
592
 
                    # we can setup and use record_counter (rc).
593
 
                    if not rc.is_initialized():
594
 
                        rc.setup(self.key_count, self.key_count)
595
 
            for record in substream.read():
596
 
                if rc:
597
 
                    if rc.is_initialized() and counter == rc.STEP:
598
 
                        rc.increment(counter)
599
 
                        pb.update('Estimate', rc.current, rc.max)
600
 
                        counter = 0
601
 
                    if self.current_type == 'revisions':
602
 
                        # Total records is proportional to number of revs
603
 
                        # to fetch. With remote, we used self.key_count to
604
 
                        # track the number of revs. Once we have the revs
605
 
                        # counts in self.key_count, the progress bar changes
606
 
                        # from 'Estimating..' to 'Estimate' above.
607
 
                        self.key_count += 1
608
 
                        if counter == rc.STEP:
609
 
                            pb.update('Estimating..', self.key_count)
610
 
                            counter = 0
611
 
                counter += 1
612
 
                yield record
613
 
 
614
 
        self.seed_state()
615
 
        pb = ui.ui_factory.nested_progress_bar()
616
 
        rc = self._record_counter
617
 
        # Make and consume sub generators, one per substream type:
618
 
        while self.first_bytes is not None:
619
 
            substream = NetworkRecordStream(self.iter_substream_bytes())
620
 
            # after substream is fully consumed, self.current_type is set to
621
 
            # the next type, and self.first_bytes is set to the matching bytes.
622
 
            yield self.current_type, wrap_and_count(pb, rc, substream)
623
 
        if rc:
624
 
            pb.update('Done', rc.max, rc.max)
625
 
        pb.finished()
626
 
 
627
 
    def seed_state(self):
628
 
        """Prepare the _ByteStreamDecoder to decode from the pack stream."""
629
 
        # Set a single generator we can use to get data from the pack stream.
630
 
        self.iter_pack_records = self.iter_stream_decoder()
631
 
        # Seed the very first subiterator with content; after this each one
632
 
        # seeds the next.
633
 
        list(self.iter_substream_bytes())
634
 
 
635
 
 
636
 
def _byte_stream_to_stream(byte_stream, record_counter=None):
637
 
    """Convert a byte stream into a format and a stream.
638
 
 
639
 
    :param byte_stream: A bytes iterator, as output by _stream_to_byte_stream.
640
 
    :return: (RepositoryFormat, stream_generator)
641
 
    """
642
 
    decoder = _ByteStreamDecoder(byte_stream, record_counter)
 
489
            substream = NetworkRecordStream([record_bytes])
 
490
            yield substream_type, substream.read()
 
491
        for bytes in byte_stream:
 
492
            stream_decoder.accept_bytes(bytes)
 
493
            for record in stream_decoder.read_pending_records():
 
494
                record_names, record_bytes = record
 
495
                record_name, = record_names
 
496
                substream_type = record_name[0]
 
497
                substream = NetworkRecordStream([record_bytes])
 
498
                yield substream_type, substream.read()
643
499
    for bytes in byte_stream:
644
 
        decoder.stream_decoder.accept_bytes(bytes)
645
 
        for record in decoder.stream_decoder.read_pending_records(max=1):
 
500
        stream_decoder.accept_bytes(bytes)
 
501
        for record in stream_decoder.read_pending_records(max=1):
646
502
            record_names, src_format_name = record
647
503
            src_format = network_format_registry.get(src_format_name)
648
 
            return src_format, decoder.record_stream()
 
504
            return src_format, record_stream()
649
505
 
650
506
 
651
507
class SmartServerRepositoryUnlock(SmartServerRepositoryRequest):
711
567
            temp.close()
712
568
 
713
569
    def _tarball_of_dir(self, dirname, compression, ofile):
714
 
        import tarfile
715
570
        filename = os.path.basename(ofile.name)
716
571
        tarball = tarfile.open(fileobj=ofile, name=filename,
717
572
            mode='w|' + compression)
795
650
            return SuccessfulSmartServerResponse(('ok', ))
796
651
 
797
652
 
798
 
class SmartServerRepositoryInsertStream_1_19(SmartServerRepositoryInsertStreamLocked):
799
 
    """Insert a record stream from a RemoteSink into a repository.
800
 
 
801
 
    Same as SmartServerRepositoryInsertStreamLocked, except:
802
 
     - the lock token argument is optional
803
 
     - servers that implement this verb accept 'inventory-delta' records in the
804
 
       stream.
805
 
 
806
 
    New in 1.19.
807
 
    """
808
 
 
809
 
    def do_repository_request(self, repository, resume_tokens, lock_token=None):
810
 
        """StreamSink.insert_stream for a remote repository."""
811
 
        SmartServerRepositoryInsertStreamLocked.do_repository_request(
812
 
            self, repository, resume_tokens, lock_token)
813
 
 
814
 
 
815
653
class SmartServerRepositoryInsertStream(SmartServerRepositoryInsertStreamLocked):
816
654
    """Insert a record stream from a RemoteSink into an unlocked repository.
817
655