Merge lp:~jelmer/brz/hpss-get-missing-keys into lp:brz

[Jelmer Vernooij]

Add HPSS call for Repository.get_stream_for_missing_keys.

Add HPSS call for Repository.get_stream_for_missing_keys.

[Martin Packman (gz)]

Some conflicts to resolve but logic all looked fine to me.

review: Approve

[The Breezy Bot (the-breezy-bot)]

Running landing tests failed
https://ci.breezy-vcs.org/job/land-brz/192/

[The Breezy Bot (the-breezy-bot)]

Running landing tests failed
https://ci.breezy-vcs.org/job/land-brz/198/

=== modified file 'breezy/bzr/remote.py'
--- breezy/bzr/remote.py	2018-06-29 23:46:33 +0000
+++ breezy/bzr/remote.py	2018-06-30 14:39:04 +0000
@@ -2883,6 +2883,17 @@
             self.target_repo.autopack()
         return result
 
+    def insert_missing_keys(self, source, missing_keys):
+        if (isinstance(source, RemoteStreamSource) and
+                source.from_repository._client._medium == self.target_repo._client._medium):
+            # Streaming from and to the same medium is tricky, since we don't support
+            # more than one concurrent request. For now, just force VFS.
+            stream = source._get_real_stream_for_missing_keys(missing_keys)
+        else:
+            stream = source.get_stream_for_missing_keys(missing_keys)
+        return self.insert_stream_without_locking(stream,
+            self.target_repo._format)
+
     def insert_stream(self, stream, src_format, resume_tokens):
         target = self.target_repo
         target._unstacked_provider.missing_keys.clear()
@@ -3010,12 +3021,38 @@
             sources.append(repo)
         return self.missing_parents_chain(search, sources)
 
-    def get_stream_for_missing_keys(self, missing_keys):
+    def _get_real_stream_for_missing_keys(self, missing_keys):
         self.from_repository._ensure_real()
         real_repo = self.from_repository._real_repository
         real_source = real_repo._get_source(self.to_format)
         return real_source.get_stream_for_missing_keys(missing_keys)
 
+    def get_stream_for_missing_keys(self, missing_keys):
+        if not isinstance(self.from_repository, RemoteRepository):
+            return self._get_real_stream_for_missing_keys(missing_keys)
+        client = self.from_repository._client
+        medium = client._medium
+        if medium._is_remote_before((3, 0)):
+            return self._get_real_stream_for_missing_keys(missing_keys)
+        path = self.from_repository.controldir._path_for_remote_call(client)
+        args = (path, self.to_format.network_name())
+        search_bytes = b'\n'.join([b'\t'.join(key) for key in missing_keys])
+        try:
+            response, handler = self.from_repository._call_with_body_bytes_expecting_body(
+                b'Repository.get_stream_for_missing_keys', args, search_bytes)
+        except (errors.UnknownSmartMethod, errors.UnknownFormatError):
+            return self._get_real_stream_for_missing_keys(missing_keys)
+        if response[0] != b'ok':
+            raise errors.UnexpectedSmartServerResponse(response)
+        byte_stream = handler.read_streamed_body()
+        src_format, stream = smart_repo._byte_stream_to_stream(byte_stream,
+            self._record_counter)
+        if src_format.network_name() != self.from_repository._format.network_name():
+            raise AssertionError(
+                "Mismatched RemoteRepository and stream src %r, %r" % (
+                src_format.network_name(), repo._format.network_name()))
+        return stream
+
     def _real_stream(self, repo, search):
         """Get a stream for search from repo.
 
@@ -3716,7 +3753,7 @@
         err_context = {'token': str((branch_token, repo_token))}
         response = self._call(
             b'Branch.unlock', self._remote_path(), branch_token,
-            repo_token or '', **err_context)
+            repo_token or b'', **err_context)
         if response == (b'ok',):
             return
         raise errors.UnexpectedSmartServerResponse(response)
@@ -4330,6 +4367,9 @@
     lambda err: errors.FileExists(err.error_args[0].decode('utf-8')))
 no_context_error_translators.register(b'DirectoryNotEmpty',
     lambda err: errors.DirectoryNotEmpty(err.error_args[0].decode('utf-8')))
+no_context_error_translators.register(b'UnknownFormat',
+    lambda err: errors.UnknownFormatError(
+        err.error_args[0].decode('ascii'), err.error_args[0].decode('ascii')))
 
 def _translate_short_readv_error(err):
     args = err.error_args

=== modified file 'breezy/bzr/smart/repository.py'
--- breezy/bzr/smart/repository.py	2018-06-30 10:19:02 +0000
+++ breezy/bzr/smart/repository.py	2018-06-30 14:39:04 +0000
@@ -1276,12 +1276,57 @@
         return None
 
 
+class SmartServerRepositoryGetStreamForMissingKeys(SmartServerRepositoryRequest):
+
+    def do_repository_request(self, repository, to_network_name):
+        """Get a stream for missing keys.
+
+        :param repository: The repository to stream from.
+        :param to_network_name: The network name of the format of the target
+            repository.
+        """
+        try:
+            self._to_format = network_format_registry.get(to_network_name)
+        except KeyError:
+            return FailedSmartServerResponse(
+                (b'UnknownFormat', b'repository', to_network_name))
+        return None # Signal that we want a body.
+
+    def do_body(self, body_bytes):
+        repository = self._repository
+        repository.lock_read()
+        try:
+            source = repository._get_source(self._to_format)
+            stream = source.get_stream_for_missing_keys(
+                [tuple(k.split(b'\t')) for k in body_bytes.split(b'\n')])
+        except Exception:
+            try:
+                # On non-error, unlocking is done by the body stream handler.
+                repository.unlock()
+            finally:
+                raise
+        return SuccessfulSmartServerResponse((b'ok',),
+            body_stream=self.body_stream(stream, repository))
+
+    def body_stream(self, stream, repository):
+        byte_stream = _stream_to_byte_stream(stream, repository._format)
+        try:
+            for bytes in byte_stream:
+                yield bytes
+        except errors.RevisionNotPresent as e:
+            # This shouldn't be able to happen, but as we don't buffer
+            # everything it can in theory happen.
+            repository.unlock()
+            yield FailedSmartServerResponse((b'NoSuchRevision', e.revision_id))
+        else:
+            repository.unlock()
+
+
 class SmartServerRepositoryRevisionArchive(SmartServerRepositoryRequest):
 
     def do_repository_request(self, repository, revision_id, format, name,
                              root, subdir=None, force_mtime=None):
         """Stream an archive file for a specific revision.
-
         :param repository: The repository to stream from.
         :param revision_id: Revision for which to export the tree
         :param format: Format (tar, tgz, tbz2, etc)

=== modified file 'breezy/bzr/smart/request.py'
--- breezy/bzr/smart/request.py	2018-06-23 12:51:06 +0000
+++ breezy/bzr/smart/request.py	2018-06-30 14:39:04 +0000
@@ -759,6 +759,9 @@
     b'Repository.get_stream_1.19', 'breezy.bzr.smart.repository',
     'SmartServerRepositoryGetStream_1_19', info='read')
 request_handlers.register_lazy(
+    b'Repository.get_stream_for_missing_keys', 'breezy.bzr.smart.repository',
+    'SmartServerRepositoryGetStreamForMissingKeys', info='read')
+request_handlers.register_lazy(
     b'Repository.iter_revisions', 'breezy.bzr.smart.repository',
     'SmartServerRepositoryIterRevisions', info='read')
 request_handlers.register_lazy(

=== modified file 'breezy/bzr/vf_repository.py'
--- breezy/bzr/vf_repository.py	2018-06-30 10:19:02 +0000
+++ breezy/bzr/vf_repository.py	2018-06-30 14:39:04 +0000
@@ -166,9 +166,7 @@
             fallback_repo = fallback_repos.pop()
             source = fallback_repo._get_source(self.repository._format)
             sink = self.repository._get_sink()
-            stream = source.get_stream_for_missing_keys(missing_keys)
-            missing_keys = sink.insert_stream_without_locking(stream,
-                self.repository._format)
+            missing_keys = sink.insert_missing_keys(source, missing_keys)
         if missing_keys:
             raise errors.BzrError('Unable to fill in parent inventories for a'
                                   ' stacked branch')
@@ -1718,6 +1716,17 @@
     def __init__(self, target_repo):
         self.target_repo = target_repo
 
+    def insert_missing_keys(self, source, missing_keys):
+        """Insert missing keys from another source.
+
+        :param source: StreamSource to stream from
+        :param missing_keys: Keys to insert
+        :return: keys still missing
+        """
+        stream = source.get_stream_for_missing_keys(missing_keys)
+        return self.insert_stream_without_locking(stream,
+            self.target_repo._format)
+
     def insert_stream(self, stream, src_format, resume_tokens):
         """Insert a stream's content into the target repository.
 

=== modified file 'breezy/tests/blackbox/test_branch.py'
--- breezy/tests/blackbox/test_branch.py	2018-06-17 11:15:04 +0000
+++ breezy/tests/blackbox/test_branch.py	2018-06-30 14:39:04 +0000
@@ -588,6 +588,25 @@
         self.expectFailure("branching to stacked requires VFS access",
             self.assertThat, self.hpss_calls, ContainsNoVfsCalls)
 
+    def test_branch_from_branch_with_ghosts(self):
+        self.setup_smart_server_with_call_log()
+        t = self.make_branch_and_tree('from')
+        for count in range(9):
+            t.commit(message='commit %d' % count)
+        t.set_parent_ids([t.last_revision(), b'ghost'])
+        t.commit(message='add commit with parent')
+        self.reset_smart_call_log()
+        out, err = self.run_bzr(['branch', self.get_url('from'),
+            'local-target'])
+        # This figure represent the amount of work to perform this use case. It
+        # is entirely ok to reduce this number if a test fails due to rpc_count
+        # being too low. If rpc_count increases, more network roundtrips have
+        # become necessary for this use case. Please do not adjust this number
+        # upwards without agreement from bzr's network support maintainers.
+        self.assertThat(self.hpss_calls, ContainsNoVfsCalls)
+        self.assertLength(11, self.hpss_calls)
+        self.assertLength(1, self.hpss_connections)
+
 
 class TestRemoteBranch(TestCaseWithSFTPServer):
 

=== modified file 'breezy/tests/per_branch/test_create_clone.py'
--- breezy/tests/per_branch/test_create_clone.py	2018-03-27 00:51:45 +0000
+++ breezy/tests/per_branch/test_create_clone.py	2018-06-30 14:39:04 +0000
@@ -140,8 +140,8 @@
         branch.Branch.hooks.install_named_hook(
             'pre_change_branch_tip', self.assertBranchHookBranchIsStacked, None)
         try:
-            result = tree.branch.create_clone_on_transport(target_transport,
-                stacked_on=trunk.base)
+            result = tree.branch.create_clone_on_transport(
+                    target_transport, stacked_on=trunk.base)
         except branch.UnstackableBranchFormat:
             if not trunk.repository._format.supports_full_versioned_files:
                 raise tests.TestNotApplicable("can not stack on format")

=== modified file 'breezy/tests/test_smart.py'
--- breezy/tests/test_smart.py	2018-06-30 12:28:52 +0000
+++ breezy/tests/test_smart.py	2018-06-30 14:39:04 +0000
@@ -2719,8 +2719,33 @@
             b"Bazaar pack format 1 (introduced in 0.18)\nB54\n\nBazaar repository format 2a (needs bzr 1.16 or later)\nE")
 
 
+class TestSmartServerRepositoryGetStreamForMissingKeys(GetStreamTestBase):
+
+    def test_missing(self):
+        """The search argument may be a 'ancestry-of' some heads'."""
+        backing = self.get_transport()
+        request = smart_repo.SmartServerRepositoryGetStreamForMissingKeys(
+            backing)
+        repo, r1, r2 = self.make_two_commit_repo()
+        request.execute(b'', repo._format.network_name())
+        lines = b'inventories\t' + r1
+        response = request.do_body(lines)
+        self.assertEqual((b'ok',), response.args)
+        stream_bytes = b''.join(response.body_stream)
+        self.assertStartsWith(stream_bytes, b'Bazaar pack format 1')
+
+    def test_unknown_format(self):
+        """The format may not be known by the remote server."""
+        backing = self.get_transport()
+        request = smart_repo.SmartServerRepositoryGetStreamForMissingKeys(
+            backing)
+        repo, r1, r2 = self.make_two_commit_repo()
+        request.execute(b'', b'yada yada yada')
+        expected = smart_req.FailedSmartServerResponse(
+            (b'UnknownFormat', b'repository', b'yada yada yada'))
+
+
 class TestSmartServerRepositoryRevisionArchive(tests.TestCaseWithTransport):
-
     def test_get(self):
         backing = self.get_transport()
         request = smart_repo.SmartServerRepositoryRevisionArchive(backing)