← Back to branch summary

~bzr-pqm/bzr/bzr.dev

« back to all changes in this revision

Viewing changes to bzrlib/smart/repository.py

  • Committer: Johan Walles
  • Date: 2009-05-07 05:08:46 UTC
  • mfrom: (4342 +trunk)
  • mto: This revision was merged to the branch mainline in revision 4343.
  • Revision ID: johan.walles@gmail.com-20090507050846-nkwvcyauf1eh653q
Merge from upstream.

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
19
19
import bz2
20
20
import os
21
21
import Queue
 
22
import struct
22
23
import sys
 
24
import tarfile
23
25
import tempfile
24
26
import threading
25
27
 
26
28
from bzrlib import (
27
 
    bencode,
28
29
    errors,
29
30
    graph,
30
31
    osutils,
31
32
    pack,
32
 
    ui,
33
33
    )
34
34
from bzrlib.bzrdir import BzrDir
35
35
from bzrlib.smart.request import (
39
39
    )
40
40
from bzrlib.repository import _strip_NULL_ghosts, network_format_registry
41
41
from bzrlib import revision as _mod_revision
42
 
from bzrlib.versionedfile import (
43
 
    NetworkRecordStream,
44
 
    record_to_fulltext_bytes,
45
 
    )
 
42
from bzrlib.util import bencode
 
43
from bzrlib.versionedfile import NetworkRecordStream, record_to_fulltext_bytes
46
44
 
47
45
 
48
46
class SmartServerRepositoryRequest(SmartServerRequest):
81
79
            recreate_search trusts that clients will look for missing things
82
80
            they expected and get it from elsewhere.
83
81
        """
84
 
        if search_bytes == 'everything':
85
 
            return graph.EverythingResult(repository), None
86
82
        lines = search_bytes.split('\n')
87
83
        if lines[0] == 'ancestry-of':
88
84
            heads = lines[1:]
288
284
        return SuccessfulSmartServerResponse(('ok', ), '\n'.join(lines))
289
285
 
290
286
 
291
 
class SmartServerRepositoryGetRevIdForRevno(SmartServerRepositoryReadLocked):
292
 
 
293
 
    def do_readlocked_repository_request(self, repository, revno,
294
 
            known_pair):
295
 
        """Find the revid for a given revno, given a known revno/revid pair.
296
 
        
297
 
        New in 1.17.
298
 
        """
299
 
        try:
300
 
            found_flag, result = repository.get_rev_id_for_revno(revno, known_pair)
301
 
        except errors.RevisionNotPresent, err:
302
 
            if err.revision_id != known_pair[1]:
303
 
                raise AssertionError(
304
 
                    'get_rev_id_for_revno raised RevisionNotPresent for '
305
 
                    'non-initial revision: ' + err.revision_id)
306
 
            return FailedSmartServerResponse(
307
 
                ('nosuchrevision', err.revision_id))
308
 
        if found_flag:
309
 
            return SuccessfulSmartServerResponse(('ok', result))
310
 
        else:
311
 
            earliest_revno, earliest_revid = result
312
 
            return SuccessfulSmartServerResponse(
313
 
                ('history-incomplete', earliest_revno, earliest_revid))
314
 
 
315
 
 
316
287
class SmartServerRequestHasRevision(SmartServerRepositoryRequest):
317
288
 
318
289
    def do_repository_request(self, repository, revision_id):
393
364
        if token == '':
394
365
            token = None
395
366
        try:
396
 
            token = repository.lock_write(token=token).repository_token
 
367
            token = repository.lock_write(token=token)
397
368
        except errors.LockContention, e:
398
369
            return FailedSmartServerResponse(('LockContention',))
399
370
        except errors.UnlockableTransport:
414
385
    def do_repository_request(self, repository, to_network_name):
415
386
        """Get a stream for inserting into a to_format repository.
416
387
 
417
 
        The request body is 'search_bytes', a description of the revisions
418
 
        being requested.
419
 
 
420
 
        In 2.3 this verb added support for search_bytes == 'everything'.  Older
421
 
        implementations will respond with a BadSearch error, and clients should
422
 
        catch this and fallback appropriately.
423
 
 
424
388
        :param repository: The repository to stream from.
425
389
        :param to_network_name: The network name of the format of the target
426
390
            repository.
427
391
        """
428
392
        self._to_format = network_format_registry.get(to_network_name)
429
 
        if self._should_fake_unknown():
430
 
            return FailedSmartServerResponse(
431
 
                ('UnknownMethod', 'Repository.get_stream'))
432
393
        return None # Signal that we want a body.
433
394
 
434
 
    def _should_fake_unknown(self):
435
 
        """Return True if we should return UnknownMethod to the client.
436
 
        
437
 
        This is a workaround for bugs in pre-1.19 clients that claim to
438
 
        support receiving streams of CHK repositories.  The pre-1.19 client
439
 
        expects inventory records to be serialized in the format defined by
440
 
        to_network_name, but in pre-1.19 (at least) that format definition
441
 
        tries to use the xml5 serializer, which does not correctly handle
442
 
        rich-roots.  After 1.19 the client can also accept inventory-deltas
443
 
        (which avoids this issue), and those clients will use the
444
 
        Repository.get_stream_1.19 verb instead of this one.
445
 
        So: if this repository is CHK, and the to_format doesn't match,
446
 
        we should just fake an UnknownSmartMethod error so that the client
447
 
        will fallback to VFS, rather than sending it a stream we know it
448
 
        cannot handle.
449
 
        """
450
 
        from_format = self._repository._format
451
 
        to_format = self._to_format
452
 
        if not from_format.supports_chks:
453
 
            # Source not CHK: that's ok
454
 
            return False
455
 
        if (to_format.supports_chks and
456
 
            from_format.repository_class is to_format.repository_class and
457
 
            from_format._serializer == to_format._serializer):
458
 
            # Source is CHK, but target matches: that's ok
459
 
            # (e.g. 2a->2a, or CHK2->2a)
460
 
            return False
461
 
        # Source is CHK, and target is not CHK or incompatible CHK.  We can't
462
 
        # generate a compatible stream.
463
 
        return True
464
 
 
465
395
    def do_body(self, body_bytes):
466
396
        repository = self._repository
467
397
        repository.lock_read()
497
427
            repository.unlock()
498
428
 
499
429
 
500
 
class SmartServerRepositoryGetStream_1_19(SmartServerRepositoryGetStream):
501
 
    """The same as Repository.get_stream, but will return stream CHK formats to
502
 
    clients.
503
 
 
504
 
    See SmartServerRepositoryGetStream._should_fake_unknown.
505
 
    
506
 
    New in 1.19.
507
 
    """
508
 
 
509
 
    def _should_fake_unknown(self):
510
 
        """Returns False; we don't need to workaround bugs in 1.19+ clients."""
511
 
        return False
512
 
 
513
 
 
514
430
def _stream_to_byte_stream(stream, src_format):
515
431
    """Convert a record stream to a self delimited byte stream."""
516
432
    pack_writer = pack.ContainerSerialiser()
520
436
        for record in substream:
521
437
            if record.storage_kind in ('chunked', 'fulltext'):
522
438
                serialised = record_to_fulltext_bytes(record)
523
 
            elif record.storage_kind == 'absent':
524
 
                raise ValueError("Absent factory for %s" % (record.key,))
525
439
            else:
526
440
                serialised = record.get_bytes_as(record.storage_kind)
527
441
            if serialised:
532
446
    yield pack_writer.end()
533
447
 
534
448
 
535
 
class _ByteStreamDecoder(object):
536
 
    """Helper for _byte_stream_to_stream.
537
 
 
538
 
    The expected usage of this class is via the function _byte_stream_to_stream
539
 
    which creates a _ByteStreamDecoder, pops off the stream format and then
540
 
    yields the output of record_stream(), the main entry point to
541
 
    _ByteStreamDecoder.
542
 
 
543
 
    Broadly this class has to unwrap two layers of iterators:
544
 
    (type, substream)
545
 
    (substream details)
546
 
 
547
 
    This is complicated by wishing to return type, iterator_for_type, but
548
 
    getting the data for iterator_for_type when we find out type: we can't
549
 
    simply pass a generator down to the NetworkRecordStream parser, instead
550
 
    we have a little local state to seed each NetworkRecordStream instance,
551
 
    and gather the type that we'll be yielding.
552
 
 
553
 
    :ivar byte_stream: The byte stream being decoded.
554
 
    :ivar stream_decoder: A pack parser used to decode the bytestream
555
 
    :ivar current_type: The current type, used to join adjacent records of the
556
 
        same type into a single stream.
557
 
    :ivar first_bytes: The first bytes to give the next NetworkRecordStream.
 
449
def _byte_stream_to_stream(byte_stream):
 
450
    """Convert a byte stream into a format and a stream.
 
451
 
 
452
    :param byte_stream: A bytes iterator, as output by _stream_to_byte_stream.
 
453
    :return: (RepositoryFormat, stream_generator)
558
454
    """
559
 
 
560
 
    def __init__(self, byte_stream, record_counter):
561
 
        """Create a _ByteStreamDecoder."""
562
 
        self.stream_decoder = pack.ContainerPushParser()
563
 
        self.current_type = None
564
 
        self.first_bytes = None
565
 
        self.byte_stream = byte_stream
566
 
        self._record_counter = record_counter
567
 
        self.key_count = 0
568
 
 
569
 
    def iter_stream_decoder(self):
570
 
        """Iterate the contents of the pack from stream_decoder."""
571
 
        # dequeue pending items
572
 
        for record in self.stream_decoder.read_pending_records():
573
 
            yield record
574
 
        # Pull bytes of the wire, decode them to records, yield those records.
575
 
        for bytes in self.byte_stream:
576
 
            self.stream_decoder.accept_bytes(bytes)
577
 
            for record in self.stream_decoder.read_pending_records():
578
 
                yield record
579
 
 
580
 
    def iter_substream_bytes(self):
581
 
        if self.first_bytes is not None:
582
 
            yield self.first_bytes
583
 
            # If we run out of pack records, single the outer layer to stop.
584
 
            self.first_bytes = None
585
 
        for record in self.iter_pack_records:
 
455
    stream_decoder = pack.ContainerPushParser()
 
456
    def record_stream():
 
457
        """Closure to return the substreams."""
 
458
        # May have fully parsed records already.
 
459
        for record in stream_decoder.read_pending_records():
586
460
            record_names, record_bytes = record
587
461
            record_name, = record_names
588
462
            substream_type = record_name[0]
589
 
            if substream_type != self.current_type:
590
 
                # end of a substream, seed the next substream.
591
 
                self.current_type = substream_type
592
 
                self.first_bytes = record_bytes
593
 
                return
594
 
            yield record_bytes
595
 
 
596
 
    def record_stream(self):
597
 
        """Yield substream_type, substream from the byte stream."""
598
 
        def wrap_and_count(pb, rc, substream):
599
 
            """Yield records from stream while showing progress."""
600
 
            counter = 0
601
 
            if rc:
602
 
                if self.current_type != 'revisions' and self.key_count != 0:
603
 
                    # As we know the number of revisions now (in self.key_count)
604
 
                    # we can setup and use record_counter (rc).
605
 
                    if not rc.is_initialized():
606
 
                        rc.setup(self.key_count, self.key_count)
607
 
            for record in substream.read():
608
 
                if rc:
609
 
                    if rc.is_initialized() and counter == rc.STEP:
610
 
                        rc.increment(counter)
611
 
                        pb.update('Estimate', rc.current, rc.max)
612
 
                        counter = 0
613
 
                    if self.current_type == 'revisions':
614
 
                        # Total records is proportional to number of revs
615
 
                        # to fetch. With remote, we used self.key_count to
616
 
                        # track the number of revs. Once we have the revs
617
 
                        # counts in self.key_count, the progress bar changes
618
 
                        # from 'Estimating..' to 'Estimate' above.
619
 
                        self.key_count += 1
620
 
                        if counter == rc.STEP:
621
 
                            pb.update('Estimating..', self.key_count)
622
 
                            counter = 0
623
 
                counter += 1
624
 
                yield record
625
 
 
626
 
        self.seed_state()
627
 
        pb = ui.ui_factory.nested_progress_bar()
628
 
        rc = self._record_counter
629
 
        # Make and consume sub generators, one per substream type:
630
 
        while self.first_bytes is not None:
631
 
            substream = NetworkRecordStream(self.iter_substream_bytes())
632
 
            # after substream is fully consumed, self.current_type is set to
633
 
            # the next type, and self.first_bytes is set to the matching bytes.
634
 
            yield self.current_type, wrap_and_count(pb, rc, substream)
635
 
        if rc:
636
 
            pb.update('Done', rc.max, rc.max)
637
 
        pb.finished()
638
 
 
639
 
    def seed_state(self):
640
 
        """Prepare the _ByteStreamDecoder to decode from the pack stream."""
641
 
        # Set a single generator we can use to get data from the pack stream.
642
 
        self.iter_pack_records = self.iter_stream_decoder()
643
 
        # Seed the very first subiterator with content; after this each one
644
 
        # seeds the next.
645
 
        list(self.iter_substream_bytes())
646
 
 
647
 
 
648
 
def _byte_stream_to_stream(byte_stream, record_counter=None):
649
 
    """Convert a byte stream into a format and a stream.
650
 
 
651
 
    :param byte_stream: A bytes iterator, as output by _stream_to_byte_stream.
652
 
    :return: (RepositoryFormat, stream_generator)
653
 
    """
654
 
    decoder = _ByteStreamDecoder(byte_stream, record_counter)
 
463
            substream = NetworkRecordStream([record_bytes])
 
464
            yield substream_type, substream.read()
 
465
        for bytes in byte_stream:
 
466
            stream_decoder.accept_bytes(bytes)
 
467
            for record in stream_decoder.read_pending_records():
 
468
                record_names, record_bytes = record
 
469
                record_name, = record_names
 
470
                substream_type = record_name[0]
 
471
                substream = NetworkRecordStream([record_bytes])
 
472
                yield substream_type, substream.read()
655
473
    for bytes in byte_stream:
656
 
        decoder.stream_decoder.accept_bytes(bytes)
657
 
        for record in decoder.stream_decoder.read_pending_records(max=1):
 
474
        stream_decoder.accept_bytes(bytes)
 
475
        for record in stream_decoder.read_pending_records(max=1):
658
476
            record_names, src_format_name = record
659
477
            src_format = network_format_registry.get(src_format_name)
660
 
            return src_format, decoder.record_stream()
 
478
            return src_format, record_stream()
661
479
 
662
480
 
663
481
class SmartServerRepositoryUnlock(SmartServerRepositoryRequest):
723
541
            temp.close()
724
542
 
725
543
    def _tarball_of_dir(self, dirname, compression, ofile):
726
 
        import tarfile
727
544
        filename = os.path.basename(ofile.name)
728
545
        tarball = tarfile.open(fileobj=ofile, name=filename,
729
546
            mode='w|' + compression)
807
624
            return SuccessfulSmartServerResponse(('ok', ))
808
625
 
809
626
 
810
 
class SmartServerRepositoryInsertStream_1_19(SmartServerRepositoryInsertStreamLocked):
811
 
    """Insert a record stream from a RemoteSink into a repository.
812
 
 
813
 
    Same as SmartServerRepositoryInsertStreamLocked, except:
814
 
     - the lock token argument is optional
815
 
     - servers that implement this verb accept 'inventory-delta' records in the
816
 
       stream.
817
 
 
818
 
    New in 1.19.
819
 
    """
820
 
 
821
 
    def do_repository_request(self, repository, resume_tokens, lock_token=None):
822
 
        """StreamSink.insert_stream for a remote repository."""
823
 
        SmartServerRepositoryInsertStreamLocked.do_repository_request(
824
 
            self, repository, resume_tokens, lock_token)
825
 
 
826
 
 
827
627
class SmartServerRepositoryInsertStream(SmartServerRepositoryInsertStreamLocked):
828
628
    """Insert a record stream from a RemoteSink into an unlocked repository.
829
629