82
84
recreate_search trusts that clients will look for missing things
83
85
they expected and get it from elsewhere.
87
if search_bytes == 'everything':
88
return graph.EverythingResult(repository), None
85
89
lines = search_bytes.split('\n')
86
90
if lines[0] == 'ancestry-of':
180
185
repository.unlock()
182
def _do_repository_request(self, body_bytes):
183
repository = self._repository
184
revision_ids = set(self._revision_ids)
185
include_missing = 'include-missing:' in revision_ids
187
revision_ids.remove('include-missing:')
188
body_lines = body_bytes.split('\n')
189
search_result, error = self.recreate_search_from_recipe(
190
repository, body_lines)
191
if error is not None:
193
# TODO might be nice to start up the search again; but thats not
194
# written or tested yet.
195
client_seen_revs = set(search_result.get_keys())
196
# Always include the requested ids.
197
client_seen_revs.difference_update(revision_ids)
199
repo_graph = repository.get_graph()
187
def _expand_requested_revs(self, repo_graph, revision_ids, client_seen_revs,
188
include_missing, max_size=65536):
201
190
queried_revs = set()
191
estimator = estimate_compressed_size.ZLibEstimator(max_size)
203
192
next_revs = revision_ids
204
193
first_loop_done = False
227
216
# add parents to the result
228
217
result[encoded_id] = parents
229
218
# Approximate the serialized cost of this revision_id.
230
size_so_far += 2 + len(encoded_id) + sum(map(len, parents))
219
line = '%s %s\n' % (encoded_id, ' '.join(parents))
220
estimator.add_content(line)
231
221
# get all the directly asked for parents, and then flesh out to
232
222
# 64K (compressed) or so. We do one level of depth at a time to
233
223
# stay in sync with the client. The 250000 magic number is
234
224
# estimated compression ratio taken from bzr.dev itself.
235
if self.no_extra_results or (
236
first_loop_done and size_so_far > 250000):
225
if self.no_extra_results or (first_loop_done and estimator.full()):
226
trace.mutter('size: %d, z_size: %d'
227
% (estimator._uncompressed_size_added,
228
estimator._compressed_size_added))
237
229
next_revs = set()
239
231
# don't query things we've already queried
240
next_revs.difference_update(queried_revs)
232
next_revs = next_revs.difference(queried_revs)
241
233
first_loop_done = True
236
def _do_repository_request(self, body_bytes):
237
repository = self._repository
238
revision_ids = set(self._revision_ids)
239
include_missing = 'include-missing:' in revision_ids
241
revision_ids.remove('include-missing:')
242
body_lines = body_bytes.split('\n')
243
search_result, error = self.recreate_search_from_recipe(
244
repository, body_lines)
245
if error is not None:
247
# TODO might be nice to start up the search again; but thats not
248
# written or tested yet.
249
client_seen_revs = set(search_result.get_keys())
250
# Always include the requested ids.
251
client_seen_revs.difference_update(revision_ids)
253
repo_graph = repository.get_graph()
254
result = self._expand_requested_revs(repo_graph, revision_ids,
255
client_seen_revs, include_missing)
243
257
# sorting trivially puts lexographically similar revision ids together.
244
258
# Compression FTW.
245
260
for revision, parents in sorted(result.items()):
246
261
lines.append(' '.join((revision, ) + tuple(parents)))
395
token = repository.lock_write(token=token)
410
token = repository.lock_write(token=token).repository_token
396
411
except errors.LockContention, e:
397
412
return FailedSmartServerResponse(('LockContention',))
398
413
except errors.UnlockableTransport:
413
428
def do_repository_request(self, repository, to_network_name):
414
429
"""Get a stream for inserting into a to_format repository.
431
The request body is 'search_bytes', a description of the revisions
434
In 2.3 this verb added support for search_bytes == 'everything'. Older
435
implementations will respond with a BadSearch error, and clients should
436
catch this and fallback appropriately.
416
438
:param repository: The repository to stream from.
417
439
:param to_network_name: The network name of the format of the target
492
514
class SmartServerRepositoryGetStream_1_19(SmartServerRepositoryGetStream):
515
"""The same as Repository.get_stream, but will return stream CHK formats to
518
See SmartServerRepositoryGetStream._should_fake_unknown.
494
523
def _should_fake_unknown(self):
495
524
"""Returns False; we don't need to workaround bugs in 1.19+ clients."""
505
534
for record in substream:
506
535
if record.storage_kind in ('chunked', 'fulltext'):
507
536
serialised = record_to_fulltext_bytes(record)
508
elif record.storage_kind == 'inventory-delta':
509
serialised = record_to_inventory_delta_bytes(record)
510
537
elif record.storage_kind == 'absent':
511
538
raise ValueError("Absent factory for %s" % (record.key,))
544
571
:ivar first_bytes: The first bytes to give the next NetworkRecordStream.
547
def __init__(self, byte_stream):
574
def __init__(self, byte_stream, record_counter):
548
575
"""Create a _ByteStreamDecoder."""
549
576
self.stream_decoder = pack.ContainerPushParser()
550
577
self.current_type = None
551
578
self.first_bytes = None
552
579
self.byte_stream = byte_stream
580
self._record_counter = record_counter
554
583
def iter_stream_decoder(self):
555
584
"""Iterate the contents of the pack from stream_decoder."""
581
610
def record_stream(self):
582
611
"""Yield substream_type, substream from the byte stream."""
612
def wrap_and_count(pb, rc, substream):
613
"""Yield records from stream while showing progress."""
616
if self.current_type != 'revisions' and self.key_count != 0:
617
# As we know the number of revisions now (in self.key_count)
618
# we can setup and use record_counter (rc).
619
if not rc.is_initialized():
620
rc.setup(self.key_count, self.key_count)
621
for record in substream.read():
623
if rc.is_initialized() and counter == rc.STEP:
624
rc.increment(counter)
625
pb.update('Estimate', rc.current, rc.max)
627
if self.current_type == 'revisions':
628
# Total records is proportional to number of revs
629
# to fetch. With remote, we used self.key_count to
630
# track the number of revs. Once we have the revs
631
# counts in self.key_count, the progress bar changes
632
# from 'Estimating..' to 'Estimate' above.
634
if counter == rc.STEP:
635
pb.update('Estimating..', self.key_count)
583
640
self.seed_state()
641
pb = ui.ui_factory.nested_progress_bar()
642
rc = self._record_counter
584
643
# Make and consume sub generators, one per substream type:
585
644
while self.first_bytes is not None:
586
645
substream = NetworkRecordStream(self.iter_substream_bytes())
587
646
# after substream is fully consumed, self.current_type is set to
588
647
# the next type, and self.first_bytes is set to the matching bytes.
589
yield self.current_type, substream.read()
648
yield self.current_type, wrap_and_count(pb, rc, substream)
650
pb.update('Done', rc.max, rc.max)
591
653
def seed_state(self):
592
654
"""Prepare the _ByteStreamDecoder to decode from the pack stream."""
597
659
list(self.iter_substream_bytes())
600
def _byte_stream_to_stream(byte_stream):
662
def _byte_stream_to_stream(byte_stream, record_counter=None):
601
663
"""Convert a byte stream into a format and a stream.
603
665
:param byte_stream: A bytes iterator, as output by _stream_to_byte_stream.
604
666
:return: (RepositoryFormat, stream_generator)
606
decoder = _ByteStreamDecoder(byte_stream)
668
decoder = _ByteStreamDecoder(byte_stream, record_counter)
607
669
for bytes in byte_stream:
608
670
decoder.stream_decoder.accept_bytes(bytes)
609
671
for record in decoder.stream_decoder.read_pending_records(max=1):
added
removed
bzrlib/smart/repository.py
