← Back to branch summary

~bzr-pqm/bzr/bzr.dev

« back to all changes in this revision

Viewing changes to bzrlib/smart/repository.py

  • Committer: Patch Queue Manager
  • Date: 2016-02-01 19:13:13 UTC
  • mfrom: (6614.2.2 trunk)
  • Revision ID: pqm@pqm.ubuntu.com-20160201191313-wdfvmfff1djde6oq
(vila) Release 2.7.0 (Vincent Ladeuil)

Show diffs side-by-side

added added

removed removed

Lines of Context:
bzrlib/smart/repository.py
14
14
# along with this program; if not, write to the Free Software
15
15
# Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA
16
16
 
17
 
"""Server-side repository related request implmentations."""
 
17
"""Server-side repository related request implementations."""
 
18
 
 
19
from __future__ import absolute_import
18
20
 
19
21
import bz2
20
22
import os
22
24
import sys
23
25
import tempfile
24
26
import threading
 
27
import zlib
25
28
 
26
29
from bzrlib import (
27
30
    bencode,
28
31
    errors,
29
32
    estimate_compressed_size,
30
 
    graph,
 
33
    inventory as _mod_inventory,
 
34
    inventory_delta,
31
35
    osutils,
32
36
    pack,
33
37
    trace,
34
38
    ui,
 
39
    vf_search,
35
40
    )
36
41
from bzrlib.bzrdir import BzrDir
37
42
from bzrlib.smart.request import (
42
47
from bzrlib.repository import _strip_NULL_ghosts, network_format_registry
43
48
from bzrlib import revision as _mod_revision
44
49
from bzrlib.versionedfile import (
 
50
    ChunkedContentFactory,
45
51
    NetworkRecordStream,
46
52
    record_to_fulltext_bytes,
47
53
    )
84
90
            they expected and get it from elsewhere.
85
91
        """
86
92
        if search_bytes == 'everything':
87
 
            return graph.EverythingResult(repository), None
 
93
            return vf_search.EverythingResult(repository), None
88
94
        lines = search_bytes.split('\n')
89
95
        if lines[0] == 'ancestry-of':
90
96
            heads = lines[1:]
91
 
            search_result = graph.PendingAncestryResult(heads, repository)
 
97
            search_result = vf_search.PendingAncestryResult(heads, repository)
92
98
            return search_result, None
93
99
        elif lines[0] == 'search':
94
100
            return self.recreate_search_from_recipe(repository, lines[1:],
118
124
                except StopIteration:
119
125
                    break
120
126
                search.stop_searching_any(exclude_keys.intersection(next_revs))
121
 
            search_result = search.get_result()
122
 
            if (not discard_excess and
123
 
                search_result.get_recipe()[3] != revision_count):
 
127
            (started_keys, excludes, included_keys) = search.get_state()
 
128
            if (not discard_excess and len(included_keys) != revision_count):
124
129
                # we got back a different amount of data than expected, this
125
130
                # gets reported as NoSuchRevision, because less revisions
126
131
                # indicates missing revisions, and more should never happen as
127
132
                # the excludes list considers ghosts and ensures that ghost
128
133
                # filling races are not a problem.
129
134
                return (None, FailedSmartServerResponse(('NoSuchRevision',)))
 
135
            search_result = vf_search.SearchResult(started_keys, excludes,
 
136
                len(included_keys), included_keys)
130
137
            return (search_result, None)
131
138
        finally:
132
139
            repository.unlock()
435
442
        return SuccessfulSmartServerResponse(('ok', ), body)
436
443
 
437
444
 
 
445
class SmartServerRepositoryGetRevisionSignatureText(
 
446
        SmartServerRepositoryRequest):
 
447
    """Return the signature text of a revision.
 
448
 
 
449
    New in 2.5.
 
450
    """
 
451
 
 
452
    def do_repository_request(self, repository, revision_id):
 
453
        """Return the result of repository.get_signature_text().
 
454
 
 
455
        :param repository: The repository to query in.
 
456
        :return: A smart server response of with the signature text as
 
457
            body.
 
458
        """
 
459
        try:
 
460
            text = repository.get_signature_text(revision_id)
 
461
        except errors.NoSuchRevision, err:
 
462
            return FailedSmartServerResponse(
 
463
                ('nosuchrevision', err.revision))
 
464
        return SuccessfulSmartServerResponse(('ok', ), text)
 
465
 
 
466
 
438
467
class SmartServerRepositoryIsShared(SmartServerRepositoryRequest):
439
468
 
440
469
    def do_repository_request(self, repository):
707
736
        self.seed_state()
708
737
        pb = ui.ui_factory.nested_progress_bar()
709
738
        rc = self._record_counter
710
 
        # Make and consume sub generators, one per substream type:
711
 
        while self.first_bytes is not None:
712
 
            substream = NetworkRecordStream(self.iter_substream_bytes())
713
 
            # after substream is fully consumed, self.current_type is set to
714
 
            # the next type, and self.first_bytes is set to the matching bytes.
715
 
            yield self.current_type, wrap_and_count(pb, rc, substream)
716
 
        if rc:
717
 
            pb.update('Done', rc.max, rc.max)
718
 
        pb.finished()
 
739
        try:
 
740
            # Make and consume sub generators, one per substream type:
 
741
            while self.first_bytes is not None:
 
742
                substream = NetworkRecordStream(self.iter_substream_bytes())
 
743
                # after substream is fully consumed, self.current_type is set
 
744
                # to the next type, and self.first_bytes is set to the matching
 
745
                # bytes.
 
746
                yield self.current_type, wrap_and_count(pb, rc, substream)
 
747
        finally:
 
748
            if rc:
 
749
                pb.update('Done', rc.max, rc.max)
 
750
            pb.finished()
719
751
 
720
752
    def seed_state(self):
721
753
        """Prepare the _ByteStreamDecoder to decode from the pack stream."""
940
972
    New in 2.5.
941
973
    """
942
974
 
943
 
    def do_repository_request(self, repository, lock_token, write_group_tokens,
944
 
            revision_id):
 
975
    def do_repository_request(self, repository, lock_token, revision_id,
 
976
            *write_group_tokens):
945
977
        """Add a revision signature text.
946
978
 
947
979
        :param repository: Repository to operate on
948
980
        :param lock_token: Lock token
 
981
        :param revision_id: Revision for which to add signature
949
982
        :param write_group_tokens: Write group tokens
950
 
        :param revision_id: Revision for which to add signature
951
983
        """
952
984
        self._lock_token = lock_token
 
985
        self._revision_id = revision_id
953
986
        self._write_group_tokens = write_group_tokens
954
 
        self._revision_id = revision_id
955
987
        return None
956
988
 
957
989
    def do_body(self, body_bytes):
965
997
        try:
966
998
            self._repository.resume_write_group(self._write_group_tokens)
967
999
            try:
968
 
                self._repository.add_signature_text(self._revision_id, body_bytes)
 
1000
                self._repository.add_signature_text(self._revision_id,
 
1001
                    body_bytes)
969
1002
            finally:
970
1003
                new_write_group_tokens = self._repository.suspend_write_group()
971
1004
        finally:
972
1005
            self._repository.unlock()
973
 
        return SuccessfulSmartServerResponse(('ok', ) + tuple(new_write_group_tokens))
 
1006
        return SuccessfulSmartServerResponse(
 
1007
            ('ok', ) + tuple(new_write_group_tokens))
974
1008
 
975
1009
 
976
1010
class SmartServerRepositoryStartWriteGroup(SmartServerRepositoryRequest):
1073
1107
    def do_repository_request(self, repository):
1074
1108
        revids = repository.all_revision_ids()
1075
1109
        return SuccessfulSmartServerResponse(("ok", ), "\n".join(revids))
 
1110
 
 
1111
 
 
1112
class SmartServerRepositoryReconcile(SmartServerRepositoryRequest):
 
1113
    """Reconcile a repository.
 
1114
 
 
1115
    New in 2.5.
 
1116
    """
 
1117
 
 
1118
    def do_repository_request(self, repository, lock_token):
 
1119
        try:
 
1120
            repository.lock_write(token=lock_token)
 
1121
        except errors.TokenLockingNotSupported, e:
 
1122
            return FailedSmartServerResponse(
 
1123
                ('TokenLockingNotSupported', ))
 
1124
        try:
 
1125
            reconciler = repository.reconcile()
 
1126
        finally:
 
1127
            repository.unlock()
 
1128
        body = [
 
1129
            "garbage_inventories: %d\n" % reconciler.garbage_inventories,
 
1130
            "inconsistent_parents: %d\n" % reconciler.inconsistent_parents,
 
1131
            ]
 
1132
        return SuccessfulSmartServerResponse(('ok', ), "".join(body))
 
1133
 
 
1134
 
 
1135
class SmartServerRepositoryPack(SmartServerRepositoryRequest):
 
1136
    """Pack a repository.
 
1137
 
 
1138
    New in 2.5.
 
1139
    """
 
1140
 
 
1141
    def do_repository_request(self, repository, lock_token, clean_obsolete_packs):
 
1142
        self._repository = repository
 
1143
        self._lock_token = lock_token
 
1144
        if clean_obsolete_packs == 'True':
 
1145
            self._clean_obsolete_packs = True
 
1146
        else:
 
1147
            self._clean_obsolete_packs = False
 
1148
        return None
 
1149
 
 
1150
    def do_body(self, body_bytes):
 
1151
        if body_bytes == "":
 
1152
            hint = None
 
1153
        else:
 
1154
            hint = body_bytes.splitlines()
 
1155
        self._repository.lock_write(token=self._lock_token)
 
1156
        try:
 
1157
            self._repository.pack(hint, self._clean_obsolete_packs)
 
1158
        finally:
 
1159
            self._repository.unlock()
 
1160
        return SuccessfulSmartServerResponse(("ok", ), )
 
1161
 
 
1162
 
 
1163
class SmartServerRepositoryIterFilesBytes(SmartServerRepositoryRequest):
 
1164
    """Iterate over the contents of files.
 
1165
 
 
1166
    The client sends a list of desired files to stream, one
 
1167
    per line, and as tuples of file id and revision, separated by
 
1168
    \0.
 
1169
 
 
1170
    The server replies with a stream. Each entry is preceded by a header,
 
1171
    which can either be:
 
1172
 
 
1173
    * "ok\x00IDX\n" where IDX is the index of the entry in the desired files
 
1174
        list sent by the client. This header is followed by the contents of
 
1175
        the file, bzip2-compressed.
 
1176
    * "absent\x00FILEID\x00REVISION\x00IDX" to indicate a text is missing.
 
1177
        The client can then raise an appropriate RevisionNotPresent error
 
1178
        or check its fallback repositories.
 
1179
 
 
1180
    New in 2.5.
 
1181
    """
 
1182
 
 
1183
    def body_stream(self, repository, desired_files):
 
1184
        self._repository.lock_read()
 
1185
        try:
 
1186
            text_keys = {}
 
1187
            for i, key in enumerate(desired_files):
 
1188
                text_keys[key] = i
 
1189
            for record in repository.texts.get_record_stream(text_keys,
 
1190
                    'unordered', True):
 
1191
                identifier = text_keys[record.key]
 
1192
                if record.storage_kind == 'absent':
 
1193
                    yield "absent\0%s\0%s\0%d\n" % (record.key[0],
 
1194
                        record.key[1], identifier)
 
1195
                    # FIXME: Way to abort early?
 
1196
                    continue
 
1197
                yield "ok\0%d\n" % identifier
 
1198
                compressor = zlib.compressobj()
 
1199
                for bytes in record.get_bytes_as('chunked'):
 
1200
                    data = compressor.compress(bytes)
 
1201
                    if data:
 
1202
                        yield data
 
1203
                data = compressor.flush()
 
1204
                if data:
 
1205
                    yield data
 
1206
        finally:
 
1207
            self._repository.unlock()
 
1208
 
 
1209
    def do_body(self, body_bytes):
 
1210
        desired_files = [
 
1211
            tuple(l.split("\0")) for l in body_bytes.splitlines()]
 
1212
        return SuccessfulSmartServerResponse(('ok', ),
 
1213
            body_stream=self.body_stream(self._repository, desired_files))
 
1214
 
 
1215
    def do_repository_request(self, repository):
 
1216
        # Signal that we want a body
 
1217
        return None
 
1218
 
 
1219
 
 
1220
class SmartServerRepositoryIterRevisions(SmartServerRepositoryRequest):
 
1221
    """Stream a list of revisions.
 
1222
 
 
1223
    The client sends a list of newline-separated revision ids in the
 
1224
    body of the request and the server replies with the serializer format,
 
1225
    and a stream of bzip2-compressed revision texts (using the specified
 
1226
    serializer format).
 
1227
 
 
1228
    Any revisions the server does not have are omitted from the stream.
 
1229
 
 
1230
    New in 2.5.
 
1231
    """
 
1232
 
 
1233
    def do_repository_request(self, repository):
 
1234
        self._repository = repository
 
1235
        # Signal there is a body
 
1236
        return None
 
1237
 
 
1238
    def do_body(self, body_bytes):
 
1239
        revision_ids = body_bytes.split("\n")
 
1240
        return SuccessfulSmartServerResponse(
 
1241
            ('ok', self._repository.get_serializer_format()),
 
1242
            body_stream=self.body_stream(self._repository, revision_ids))
 
1243
 
 
1244
    def body_stream(self, repository, revision_ids):
 
1245
        self._repository.lock_read()
 
1246
        try:
 
1247
            for record in repository.revisions.get_record_stream(
 
1248
                [(revid,) for revid in revision_ids], 'unordered', True):
 
1249
                if record.storage_kind == 'absent':
 
1250
                    continue
 
1251
                yield zlib.compress(record.get_bytes_as('fulltext'))
 
1252
        finally:
 
1253
            self._repository.unlock()
 
1254
 
 
1255
 
 
1256
class SmartServerRepositoryGetInventories(SmartServerRepositoryRequest):
 
1257
    """Get the inventory deltas for a set of revision ids.
 
1258
 
 
1259
    This accepts a list of revision ids, and then sends a chain
 
1260
    of deltas for the inventories of those revisions. The first
 
1261
    revision will be empty.
 
1262
 
 
1263
    The server writes back zlibbed serialized inventory deltas,
 
1264
    in the ordering specified. The base for each delta is the
 
1265
    inventory generated by the previous delta.
 
1266
 
 
1267
    New in 2.5.
 
1268
    """
 
1269
 
 
1270
    def _inventory_delta_stream(self, repository, ordering, revids):
 
1271
        prev_inv = _mod_inventory.Inventory(root_id=None,
 
1272
            revision_id=_mod_revision.NULL_REVISION)
 
1273
        serializer = inventory_delta.InventoryDeltaSerializer(
 
1274
            repository.supports_rich_root(),
 
1275
            repository._format.supports_tree_reference)
 
1276
        repository.lock_read()
 
1277
        try:
 
1278
            for inv, revid in repository._iter_inventories(revids, ordering):
 
1279
                if inv is None:
 
1280
                    continue
 
1281
                inv_delta = inv._make_delta(prev_inv)
 
1282
                lines = serializer.delta_to_lines(
 
1283
                    prev_inv.revision_id, inv.revision_id, inv_delta)
 
1284
                yield ChunkedContentFactory(inv.revision_id, None, None, lines)
 
1285
                prev_inv = inv
 
1286
        finally:
 
1287
            repository.unlock()
 
1288
 
 
1289
    def body_stream(self, repository, ordering, revids):
 
1290
        substream = self._inventory_delta_stream(repository,
 
1291
            ordering, revids)
 
1292
        return _stream_to_byte_stream([('inventory-deltas', substream)],
 
1293
            repository._format)
 
1294
 
 
1295
    def do_body(self, body_bytes):
 
1296
        return SuccessfulSmartServerResponse(('ok', ),
 
1297
            body_stream=self.body_stream(self._repository, self._ordering,
 
1298
                body_bytes.splitlines()))
 
1299
 
 
1300
    def do_repository_request(self, repository, ordering):
 
1301
        if ordering == 'unordered':
 
1302
            # inventory deltas for a topologically sorted stream
 
1303
            # are likely to be smaller
 
1304
            ordering = 'topological'
 
1305
        self._ordering = ordering
 
1306
        # Signal that we want a body
 
1307
        return None