276
270
annotated_kind = ''
277
271
self.storage_kind = 'knit-%s%s-gz' % (annotated_kind, kind)
278
272
self._raw_record = raw_record
279
self._network_bytes = network_bytes
280
273
self._build_details = build_details
281
274
self._knit = knit
283
def _create_network_bytes(self):
284
"""Create a fully serialised network version for transmission."""
285
# storage_kind, key, parents, Noeol, raw_record
286
key_bytes = '\x00'.join(self.key)
287
if self.parents is None:
288
parent_bytes = 'None:'
290
parent_bytes = '\t'.join('\x00'.join(key) for key in self.parents)
291
if self._build_details[1]:
295
network_bytes = "%s\n%s\n%s\n%s%s" % (self.storage_kind, key_bytes,
296
parent_bytes, noeol, self._raw_record)
297
self._network_bytes = network_bytes
299
def get_bytes_as(self, storage_kind):
300
if storage_kind == self.storage_kind:
301
if self._network_bytes is None:
302
self._create_network_bytes()
303
return self._network_bytes
304
if ('-ft-' in self.storage_kind and
305
storage_kind in ('chunked', 'fulltext')):
306
adapter_key = (self.storage_kind, 'fulltext')
307
adapter_factory = adapter_registry.get(adapter_key)
308
adapter = adapter_factory(None)
309
bytes = adapter.get_bytes(self)
310
if storage_kind == 'chunked':
314
if self._knit is not None:
315
# Not redundant with direct conversion above - that only handles
317
if storage_kind == 'chunked':
318
return self._knit.get_lines(self.key[0])
319
elif storage_kind == 'fulltext':
320
return self._knit.get_text(self.key[0])
321
raise errors.UnavailableRepresentation(self.key, storage_kind,
325
class LazyKnitContentFactory(ContentFactory):
326
"""A ContentFactory which can either generate full text or a wire form.
328
:seealso ContentFactory:
331
def __init__(self, key, parents, generator, first):
332
"""Create a LazyKnitContentFactory.
334
:param key: The key of the record.
335
:param parents: The parents of the record.
336
:param generator: A _ContentMapGenerator containing the record for this
338
:param first: Is this the first content object returned from generator?
339
if it is, its storage kind is knit-delta-closure, otherwise it is
340
knit-delta-closure-ref
343
self.parents = parents
345
self._generator = generator
346
self.storage_kind = "knit-delta-closure"
348
self.storage_kind = self.storage_kind + "-ref"
351
def get_bytes_as(self, storage_kind):
352
if storage_kind == self.storage_kind:
354
return self._generator._wire_bytes()
356
# all the keys etc are contained in the bytes returned in the
359
if storage_kind in ('chunked', 'fulltext'):
360
chunks = self._generator._get_one_work(self.key).text()
361
if storage_kind == 'chunked':
364
return ''.join(chunks)
365
raise errors.UnavailableRepresentation(self.key, storage_kind,
369
def knit_delta_closure_to_records(storage_kind, bytes, line_end):
370
"""Convert a network record to a iterator over stream records.
372
:param storage_kind: The storage kind of the record.
373
Must be 'knit-delta-closure'.
374
:param bytes: The bytes of the record on the network.
376
generator = _NetworkContentMapGenerator(bytes, line_end)
377
return generator.get_record_stream()
380
def knit_network_to_record(storage_kind, bytes, line_end):
381
"""Convert a network record to a record object.
383
:param storage_kind: The storage kind of the record.
384
:param bytes: The bytes of the record on the network.
387
line_end = bytes.find('\n', start)
388
key = tuple(bytes[start:line_end].split('\x00'))
390
line_end = bytes.find('\n', start)
391
parent_line = bytes[start:line_end]
392
if parent_line == 'None:':
396
[tuple(segment.split('\x00')) for segment in parent_line.split('\t')
399
noeol = bytes[start] == 'N'
400
if 'ft' in storage_kind:
403
method = 'line-delta'
404
build_details = (method, noeol)
406
raw_record = bytes[start:]
407
annotated = 'annotated' in storage_kind
408
return [KnitContentFactory(key, parents, build_details, None, raw_record,
409
annotated, network_bytes=bytes)]
276
def get_bytes_as(self, storage_kind):
277
if storage_kind == self.storage_kind:
278
return self._raw_record
279
if storage_kind == 'fulltext' and self._knit is not None:
280
return self._knit.get_text(self.key[0])
282
raise errors.UnavailableRepresentation(self.key, storage_kind,
412
286
class KnitContent(object):
413
287
"""Content of a knit version to which deltas can be applied.
415
289
This is always stored in memory as a list of lines with \n at the end,
416
plus a flag saying if the final ending is really there or not, because that
290
plus a flag saying if the final ending is really there or not, because that
417
291
corresponds to the on-disk knit representation.
1189
983
if not self.get_parent_map([key]):
1190
984
raise RevisionNotPresent(key, self)
1191
985
return cached_version
1192
generator = _VFContentMapGenerator(self, [key])
1193
return generator._get_content(key)
1195
def get_known_graph_ancestry(self, keys):
1196
"""Get a KnownGraph instance with the ancestry of keys."""
1197
parent_map, missing_keys = self._index.find_ancestry(keys)
1198
for fallback in self._fallback_vfs:
986
text_map, contents_map = self._get_content_maps([key])
987
return contents_map[key]
989
def _get_content_maps(self, keys, nonlocal_keys=None):
990
"""Produce maps of text and KnitContents
992
:param keys: The keys to produce content maps for.
993
:param nonlocal_keys: An iterable of keys(possibly intersecting keys)
994
which are known to not be in this knit, but rather in one of the
996
:return: (text_map, content_map) where text_map contains the texts for
997
the requested versions and content_map contains the KnitContents.
999
# FUTURE: This function could be improved for the 'extract many' case
1000
# by tracking each component and only doing the copy when the number of
1001
# children than need to apply delta's to it is > 1 or it is part of the
1004
multiple_versions = len(keys) != 1
1005
record_map = self._get_record_map(keys, allow_missing=True)
1010
if nonlocal_keys is None:
1011
nonlocal_keys = set()
1013
nonlocal_keys = frozenset(nonlocal_keys)
1014
missing_keys = set(nonlocal_keys)
1015
for source in self._fallback_vfs:
1199
1016
if not missing_keys:
1201
(f_parent_map, f_missing_keys) = fallback._index.find_ancestry(
1203
parent_map.update(f_parent_map)
1204
missing_keys = f_missing_keys
1205
kg = _mod_graph.KnownGraph(parent_map)
1018
for record in source.get_record_stream(missing_keys,
1020
if record.storage_kind == 'absent':
1022
missing_keys.remove(record.key)
1023
lines = split_lines(record.get_bytes_as('fulltext'))
1024
text_map[record.key] = lines
1025
content_map[record.key] = PlainKnitContent(lines, record.key)
1026
if record.key in keys:
1027
final_content[record.key] = content_map[record.key]
1029
if key in nonlocal_keys:
1034
while cursor is not None:
1036
record, record_details, digest, next = record_map[cursor]
1038
raise RevisionNotPresent(cursor, self)
1039
components.append((cursor, record, record_details, digest))
1041
if cursor in content_map:
1042
# no need to plan further back
1043
components.append((cursor, None, None, None))
1047
for (component_id, record, record_details,
1048
digest) in reversed(components):
1049
if component_id in content_map:
1050
content = content_map[component_id]
1052
content, delta = self._factory.parse_record(key[-1],
1053
record, record_details, content,
1054
copy_base_content=multiple_versions)
1055
if multiple_versions:
1056
content_map[component_id] = content
1058
final_content[key] = content
1060
# digest here is the digest from the last applied component.
1061
text = content.text()
1062
actual_sha = sha_strings(text)
1063
if actual_sha != digest:
1064
raise SHA1KnitCorrupt(self, actual_sha, digest, key, text)
1065
text_map[key] = text
1066
return text_map, final_content
1208
1068
def get_parent_map(self, keys):
1209
1069
"""Get a map of the graph parents of keys.
2025
class _ContentMapGenerator(object):
2026
"""Generate texts or expose raw deltas for a set of texts."""
2028
def __init__(self, ordering='unordered'):
2029
self._ordering = ordering
2031
def _get_content(self, key):
2032
"""Get the content object for key."""
2033
# Note that _get_content is only called when the _ContentMapGenerator
2034
# has been constructed with just one key requested for reconstruction.
2035
if key in self.nonlocal_keys:
2036
record = self.get_record_stream().next()
2037
# Create a content object on the fly
2038
lines = osutils.chunks_to_lines(record.get_bytes_as('chunked'))
2039
return PlainKnitContent(lines, record.key)
2041
# local keys we can ask for directly
2042
return self._get_one_work(key)
2044
def get_record_stream(self):
2045
"""Get a record stream for the keys requested during __init__."""
2046
for record in self._work():
2050
"""Produce maps of text and KnitContents as dicts.
2052
:return: (text_map, content_map) where text_map contains the texts for
2053
the requested versions and content_map contains the KnitContents.
2055
# NB: By definition we never need to read remote sources unless texts
2056
# are requested from them: we don't delta across stores - and we
2057
# explicitly do not want to to prevent data loss situations.
2058
if self.global_map is None:
2059
self.global_map = self.vf.get_parent_map(self.keys)
2060
nonlocal_keys = self.nonlocal_keys
2062
missing_keys = set(nonlocal_keys)
2063
# Read from remote versioned file instances and provide to our caller.
2064
for source in self.vf._fallback_vfs:
2065
if not missing_keys:
2067
# Loop over fallback repositories asking them for texts - ignore
2068
# any missing from a particular fallback.
2069
for record in source.get_record_stream(missing_keys,
2070
self._ordering, True):
2071
if record.storage_kind == 'absent':
2072
# Not in thie particular stream, may be in one of the
2073
# other fallback vfs objects.
2075
missing_keys.remove(record.key)
2078
if self._raw_record_map is None:
2079
raise AssertionError('_raw_record_map should have been filled')
2081
for key in self.keys:
2082
if key in self.nonlocal_keys:
2084
yield LazyKnitContentFactory(key, self.global_map[key], self, first)
2087
def _get_one_work(self, requested_key):
2088
# Now, if we have calculated everything already, just return the
2090
if requested_key in self._contents_map:
2091
return self._contents_map[requested_key]
2092
# To simplify things, parse everything at once - code that wants one text
2093
# probably wants them all.
2094
# FUTURE: This function could be improved for the 'extract many' case
2095
# by tracking each component and only doing the copy when the number of
2096
# children than need to apply delta's to it is > 1 or it is part of the
2098
multiple_versions = len(self.keys) != 1
2099
if self._record_map is None:
2100
self._record_map = self.vf._raw_map_to_record_map(
2101
self._raw_record_map)
2102
record_map = self._record_map
2103
# raw_record_map is key:
2104
# Have read and parsed records at this point.
2105
for key in self.keys:
2106
if key in self.nonlocal_keys:
2111
while cursor is not None:
2113
record, record_details, digest, next = record_map[cursor]
2115
raise RevisionNotPresent(cursor, self)
2116
components.append((cursor, record, record_details, digest))
2118
if cursor in self._contents_map:
2119
# no need to plan further back
2120
components.append((cursor, None, None, None))
2124
for (component_id, record, record_details,
2125
digest) in reversed(components):
2126
if component_id in self._contents_map:
2127
content = self._contents_map[component_id]
2129
content, delta = self._factory.parse_record(key[-1],
2130
record, record_details, content,
2131
copy_base_content=multiple_versions)
2132
if multiple_versions:
2133
self._contents_map[component_id] = content
2135
# digest here is the digest from the last applied component.
2136
text = content.text()
2137
actual_sha = sha_strings(text)
2138
if actual_sha != digest:
2139
raise SHA1KnitCorrupt(self, actual_sha, digest, key, text)
2140
if multiple_versions:
2141
return self._contents_map[requested_key]
2145
def _wire_bytes(self):
2146
"""Get the bytes to put on the wire for 'key'.
2148
The first collection of bytes asked for returns the serialised
2149
raw_record_map and the additional details (key, parent) for key.
2150
Subsequent calls return just the additional details (key, parent).
2151
The wire storage_kind given for the first key is 'knit-delta-closure',
2152
For subsequent keys it is 'knit-delta-closure-ref'.
2154
:param key: A key from the content generator.
2155
:return: Bytes to put on the wire.
2158
# kind marker for dispatch on the far side,
2159
lines.append('knit-delta-closure')
2161
if self.vf._factory.annotated:
2162
lines.append('annotated')
2165
# then the list of keys
2166
lines.append('\t'.join(['\x00'.join(key) for key in self.keys
2167
if key not in self.nonlocal_keys]))
2168
# then the _raw_record_map in serialised form:
2170
# for each item in the map:
2172
# 1 line with parents if the key is to be yielded (None: for None, '' for ())
2173
# one line with method
2174
# one line with noeol
2175
# one line with next ('' for None)
2176
# one line with byte count of the record bytes
2178
for key, (record_bytes, (method, noeol), next) in \
2179
self._raw_record_map.iteritems():
2180
key_bytes = '\x00'.join(key)
2181
parents = self.global_map.get(key, None)
2183
parent_bytes = 'None:'
2185
parent_bytes = '\t'.join('\x00'.join(key) for key in parents)
2186
method_bytes = method
2192
next_bytes = '\x00'.join(next)
2195
map_byte_list.append('%s\n%s\n%s\n%s\n%s\n%d\n%s' % (
2196
key_bytes, parent_bytes, method_bytes, noeol_bytes, next_bytes,
2197
len(record_bytes), record_bytes))
2198
map_bytes = ''.join(map_byte_list)
2199
lines.append(map_bytes)
2200
bytes = '\n'.join(lines)
2204
class _VFContentMapGenerator(_ContentMapGenerator):
2205
"""Content map generator reading from a VersionedFiles object."""
2207
def __init__(self, versioned_files, keys, nonlocal_keys=None,
2208
global_map=None, raw_record_map=None, ordering='unordered'):
2209
"""Create a _ContentMapGenerator.
2211
:param versioned_files: The versioned files that the texts are being
2213
:param keys: The keys to produce content maps for.
2214
:param nonlocal_keys: An iterable of keys(possibly intersecting keys)
2215
which are known to not be in this knit, but rather in one of the
2217
:param global_map: The result of get_parent_map(keys) (or a supermap).
2218
This is required if get_record_stream() is to be used.
2219
:param raw_record_map: A unparsed raw record map to use for answering
2222
_ContentMapGenerator.__init__(self, ordering=ordering)
2223
# The vf to source data from
2224
self.vf = versioned_files
2226
self.keys = list(keys)
2227
# Keys known to be in fallback vfs objects
2228
if nonlocal_keys is None:
2229
self.nonlocal_keys = set()
2231
self.nonlocal_keys = frozenset(nonlocal_keys)
2232
# Parents data for keys to be returned in get_record_stream
2233
self.global_map = global_map
2234
# The chunked lists for self.keys in text form
2236
# A cache of KnitContent objects used in extracting texts.
2237
self._contents_map = {}
2238
# All the knit records needed to assemble the requested keys as full
2240
self._record_map = None
2241
if raw_record_map is None:
2242
self._raw_record_map = self.vf._get_record_map_unparsed(keys,
2245
self._raw_record_map = raw_record_map
2246
# the factory for parsing records
2247
self._factory = self.vf._factory
2250
class _NetworkContentMapGenerator(_ContentMapGenerator):
2251
"""Content map generator sourced from a network stream."""
2253
def __init__(self, bytes, line_end):
2254
"""Construct a _NetworkContentMapGenerator from a bytes block."""
2256
self.global_map = {}
2257
self._raw_record_map = {}
2258
self._contents_map = {}
2259
self._record_map = None
2260
self.nonlocal_keys = []
2261
# Get access to record parsing facilities
2262
self.vf = KnitVersionedFiles(None, None)
2265
line_end = bytes.find('\n', start)
2266
line = bytes[start:line_end]
2267
start = line_end + 1
2268
if line == 'annotated':
2269
self._factory = KnitAnnotateFactory()
2271
self._factory = KnitPlainFactory()
2272
# list of keys to emit in get_record_stream
2273
line_end = bytes.find('\n', start)
2274
line = bytes[start:line_end]
2275
start = line_end + 1
2277
tuple(segment.split('\x00')) for segment in line.split('\t')
2279
# now a loop until the end. XXX: It would be nice if this was just a
2280
# bunch of the same records as get_record_stream(..., False) gives, but
2281
# there is a decent sized gap stopping that at the moment.
2285
line_end = bytes.find('\n', start)
2286
key = tuple(bytes[start:line_end].split('\x00'))
2287
start = line_end + 1
2288
# 1 line with parents (None: for None, '' for ())
2289
line_end = bytes.find('\n', start)
2290
line = bytes[start:line_end]
2295
[tuple(segment.split('\x00')) for segment in line.split('\t')
2297
self.global_map[key] = parents
2298
start = line_end + 1
2299
# one line with method
2300
line_end = bytes.find('\n', start)
2301
line = bytes[start:line_end]
2303
start = line_end + 1
2304
# one line with noeol
2305
line_end = bytes.find('\n', start)
2306
line = bytes[start:line_end]
2308
start = line_end + 1
2309
# one line with next ('' for None)
2310
line_end = bytes.find('\n', start)
2311
line = bytes[start:line_end]
2315
next = tuple(bytes[start:line_end].split('\x00'))
2316
start = line_end + 1
2317
# one line with byte count of the record bytes
2318
line_end = bytes.find('\n', start)
2319
line = bytes[start:line_end]
2321
start = line_end + 1
2323
record_bytes = bytes[start:start+count]
2324
start = start + count
2326
self._raw_record_map[key] = (record_bytes, (method, noeol), next)
2328
def get_record_stream(self):
2329
"""Get a record stream for for keys requested by the bytestream."""
2331
for key in self.keys:
2332
yield LazyKnitContentFactory(key, self.global_map[key], self, first)
2335
def _wire_bytes(self):
2339
1764
class _KndxIndex(object):
2340
1765
"""Manages knit index files
3431
2653
annotator = _KnitAnnotator(knit)
3432
return iter(annotator.annotate_flat(revision_id))
3435
class _KnitAnnotator(annotate.Annotator):
2654
return iter(annotator.annotate(revision_id))
2657
class _KnitAnnotator(object):
3436
2658
"""Build up the annotations for a text."""
3438
def __init__(self, vf):
3439
annotate.Annotator.__init__(self, vf)
3441
# TODO: handle Nodes which cannot be extracted
3442
# self._ghosts = set()
3444
# Map from (key, parent_key) => matching_blocks, should be 'use once'
3445
self._matching_blocks = {}
3447
# KnitContent objects
3448
self._content_objects = {}
3449
# The number of children that depend on this fulltext content object
3450
self._num_compression_children = {}
3451
# Delta records that need their compression parent before they can be
3453
self._pending_deltas = {}
3454
# Fulltext records that are waiting for their parents fulltexts before
3455
# they can be yielded for annotation
3456
self._pending_annotation = {}
2660
def __init__(self, knit):
2663
# Content objects, differs from fulltexts because of how final newlines
2664
# are treated by knits. the content objects here will always have a
2666
self._fulltext_contents = {}
2668
# Annotated lines of specific revisions
2669
self._annotated_lines = {}
2671
# Track the raw data for nodes that we could not process yet.
2672
# This maps the revision_id of the base to a list of children that will
2673
# annotated from it.
2674
self._pending_children = {}
2676
# Nodes which cannot be extracted
2677
self._ghosts = set()
2679
# Track how many children this node has, so we know if we need to keep
2681
self._annotate_children = {}
2682
self._compression_children = {}
3458
2684
self._all_build_details = {}
2685
# The children => parent revision_id graph
2686
self._revision_id_graph = {}
2688
self._heads_provider = None
2690
self._nodes_to_keep_annotations = set()
2691
self._generations_until_keep = 100
2693
def set_generations_until_keep(self, value):
2694
"""Set the number of generations before caching a node.
2696
Setting this to -1 will cache every merge node, setting this higher
2697
will cache fewer nodes.
2699
self._generations_until_keep = value
2701
def _add_fulltext_content(self, revision_id, content_obj):
2702
self._fulltext_contents[revision_id] = content_obj
2703
# TODO: jam 20080305 It might be good to check the sha1digest here
2704
return content_obj.text()
2706
def _check_parents(self, child, nodes_to_annotate):
2707
"""Check if all parents have been processed.
2709
:param child: A tuple of (rev_id, parents, raw_content)
2710
:param nodes_to_annotate: If child is ready, add it to
2711
nodes_to_annotate, otherwise put it back in self._pending_children
2713
for parent_id in child[1]:
2714
if (parent_id not in self._annotated_lines):
2715
# This parent is present, but another parent is missing
2716
self._pending_children.setdefault(parent_id,
2720
# This one is ready to be processed
2721
nodes_to_annotate.append(child)
2723
def _add_annotation(self, revision_id, fulltext, parent_ids,
2724
left_matching_blocks=None):
2725
"""Add an annotation entry.
2727
All parents should already have been annotated.
2728
:return: A list of children that now have their parents satisfied.
2730
a = self._annotated_lines
2731
annotated_parent_lines = [a[p] for p in parent_ids]
2732
annotated_lines = list(annotate.reannotate(annotated_parent_lines,
2733
fulltext, revision_id, left_matching_blocks,
2734
heads_provider=self._get_heads_provider()))
2735
self._annotated_lines[revision_id] = annotated_lines
2736
for p in parent_ids:
2737
ann_children = self._annotate_children[p]
2738
ann_children.remove(revision_id)
2739
if (not ann_children
2740
and p not in self._nodes_to_keep_annotations):
2741
del self._annotated_lines[p]
2742
del self._all_build_details[p]
2743
if p in self._fulltext_contents:
2744
del self._fulltext_contents[p]
2745
# Now that we've added this one, see if there are any pending
2746
# deltas to be done, certainly this parent is finished
2747
nodes_to_annotate = []
2748
for child in self._pending_children.pop(revision_id, []):
2749
self._check_parents(child, nodes_to_annotate)
2750
return nodes_to_annotate
3460
2752
def _get_build_graph(self, key):
3461
2753
"""Get the graphs for building texts and annotations.
3468
2760
:return: A list of (key, index_memo) records, suitable for
3469
passing to read_records_iter to start reading in the raw data from
2761
passing to read_records_iter to start reading in the raw data fro/
2764
if key in self._annotated_lines:
3472
2767
pending = set([key])
3475
self._num_needed_children[key] = 1
3477
2772
# get all pending nodes
3478
2774
this_iteration = pending
3479
build_details = self._vf._index.get_build_details(this_iteration)
2775
build_details = self._knit._index.get_build_details(this_iteration)
3480
2776
self._all_build_details.update(build_details)
3481
# new_nodes = self._vf._index._get_entries(this_iteration)
2777
# new_nodes = self._knit._index._get_entries(this_iteration)
3482
2778
pending = set()
3483
2779
for key, details in build_details.iteritems():
3484
(index_memo, compression_parent, parent_keys,
2780
(index_memo, compression_parent, parents,
3485
2781
record_details) = details
3486
self._parent_map[key] = parent_keys
3487
self._heads_provider = None
2782
self._revision_id_graph[key] = parents
3488
2783
records.append((key, index_memo))
3489
2784
# Do we actually need to check _annotated_lines?
3490
pending.update([p for p in parent_keys
3491
if p not in self._all_build_details])
3493
for parent_key in parent_keys:
3494
if parent_key in self._num_needed_children:
3495
self._num_needed_children[parent_key] += 1
3497
self._num_needed_children[parent_key] = 1
2785
pending.update(p for p in parents
2786
if p not in self._all_build_details)
3498
2787
if compression_parent:
3499
if compression_parent in self._num_compression_children:
3500
self._num_compression_children[compression_parent] += 1
3502
self._num_compression_children[compression_parent] = 1
2788
self._compression_children.setdefault(compression_parent,
2791
for parent in parents:
2792
self._annotate_children.setdefault(parent,
2794
num_gens = generation - kept_generation
2795
if ((num_gens >= self._generations_until_keep)
2796
and len(parents) > 1):
2797
kept_generation = generation
2798
self._nodes_to_keep_annotations.add(key)
3504
2800
missing_versions = this_iteration.difference(build_details.keys())
3505
if missing_versions:
3506
for key in missing_versions:
3507
if key in self._parent_map and key in self._text_cache:
3508
# We already have this text ready, we just need to
3509
# yield it later so we get it annotated
3511
parent_keys = self._parent_map[key]
3512
for parent_key in parent_keys:
3513
if parent_key in self._num_needed_children:
3514
self._num_needed_children[parent_key] += 1
3516
self._num_needed_children[parent_key] = 1
3517
pending.update([p for p in parent_keys
3518
if p not in self._all_build_details])
3520
raise errors.RevisionNotPresent(key, self._vf)
2801
self._ghosts.update(missing_versions)
2802
for missing_version in missing_versions:
2803
# add a key, no parents
2804
self._revision_id_graph[missing_version] = ()
2805
pending.discard(missing_version) # don't look for it
2806
if self._ghosts.intersection(self._compression_children):
2808
"We cannot have nodes which have a ghost compression parent:\n"
2810
"compression children: %r"
2811
% (self._ghosts, self._compression_children))
2812
# Cleanout anything that depends on a ghost so that we don't wait for
2813
# the ghost to show up
2814
for node in self._ghosts:
2815
if node in self._annotate_children:
2816
# We won't be building this node
2817
del self._annotate_children[node]
3521
2818
# Generally we will want to read the records in reverse order, because
3522
2819
# we find the parent nodes after the children
3523
2820
records.reverse()
3524
return records, ann_keys
3526
def _get_needed_texts(self, key, pb=None):
3527
# if True or len(self._vf._fallback_vfs) > 0:
3528
if len(self._vf._fallback_vfs) > 0:
3529
# If we have fallbacks, go to the generic path
3530
for v in annotate.Annotator._get_needed_texts(self, key, pb=pb):
3535
records, ann_keys = self._get_build_graph(key)
3536
for idx, (sub_key, text, num_lines) in enumerate(
3537
self._extract_texts(records)):
3539
pb.update('annotating', idx, len(records))
3540
yield sub_key, text, num_lines
3541
for sub_key in ann_keys:
3542
text = self._text_cache[sub_key]
3543
num_lines = len(text) # bad assumption
3544
yield sub_key, text, num_lines
3546
except errors.RetryWithNewPacks, e:
3547
self._vf._access.reload_or_raise(e)
3548
# The cached build_details are no longer valid
3549
self._all_build_details.clear()
3551
def _cache_delta_blocks(self, key, compression_parent, delta, lines):
3552
parent_lines = self._text_cache[compression_parent]
3553
blocks = list(KnitContent.get_line_delta_blocks(delta, parent_lines, lines))
3554
self._matching_blocks[(key, compression_parent)] = blocks
3556
def _expand_record(self, key, parent_keys, compression_parent, record,
3559
if compression_parent:
3560
if compression_parent not in self._content_objects:
3561
# Waiting for the parent
3562
self._pending_deltas.setdefault(compression_parent, []).append(
3563
(key, parent_keys, record, record_details))
3565
# We have the basis parent, so expand the delta
3566
num = self._num_compression_children[compression_parent]
3569
base_content = self._content_objects.pop(compression_parent)
3570
self._num_compression_children.pop(compression_parent)
3572
self._num_compression_children[compression_parent] = num
3573
base_content = self._content_objects[compression_parent]
3574
# It is tempting to want to copy_base_content=False for the last
3575
# child object. However, whenever noeol=False,
3576
# self._text_cache[parent_key] is content._lines. So mutating it
3577
# gives very bad results.
3578
# The alternative is to copy the lines into text cache, but then we
3579
# are copying anyway, so just do it here.
3580
content, delta = self._vf._factory.parse_record(
3581
key, record, record_details, base_content,
3582
copy_base_content=True)
3585
content, _ = self._vf._factory.parse_record(
3586
key, record, record_details, None)
3587
if self._num_compression_children.get(key, 0) > 0:
3588
self._content_objects[key] = content
3589
lines = content.text()
3590
self._text_cache[key] = lines
3591
if delta is not None:
3592
self._cache_delta_blocks(key, compression_parent, delta, lines)
3595
def _get_parent_annotations_and_matches(self, key, text, parent_key):
3596
"""Get the list of annotations for the parent, and the matching lines.
3598
:param text: The opaque value given by _get_needed_texts
3599
:param parent_key: The key for the parent text
3600
:return: (parent_annotations, matching_blocks)
3601
parent_annotations is a list as long as the number of lines in
3603
matching_blocks is a list of (parent_idx, text_idx, len) tuples
3604
indicating which lines match between the two texts
3606
block_key = (key, parent_key)
3607
if block_key in self._matching_blocks:
3608
blocks = self._matching_blocks.pop(block_key)
3609
parent_annotations = self._annotations_cache[parent_key]
3610
return parent_annotations, blocks
3611
return annotate.Annotator._get_parent_annotations_and_matches(self,
3612
key, text, parent_key)
3614
def _process_pending(self, key):
3615
"""The content for 'key' was just processed.
3617
Determine if there is any more pending work to be processed.
3620
if key in self._pending_deltas:
3621
compression_parent = key
3622
children = self._pending_deltas.pop(key)
3623
for child_key, parent_keys, record, record_details in children:
3624
lines = self._expand_record(child_key, parent_keys,
3626
record, record_details)
3627
if self._check_ready_for_annotations(child_key, parent_keys):
3628
to_return.append(child_key)
3629
# Also check any children that are waiting for this parent to be
3631
if key in self._pending_annotation:
3632
children = self._pending_annotation.pop(key)
3633
to_return.extend([c for c, p_keys in children
3634
if self._check_ready_for_annotations(c, p_keys)])
3637
def _check_ready_for_annotations(self, key, parent_keys):
3638
"""return true if this text is ready to be yielded.
3640
Otherwise, this will return False, and queue the text into
3641
self._pending_annotation
3643
for parent_key in parent_keys:
3644
if parent_key not in self._annotations_cache:
3645
# still waiting on at least one parent text, so queue it up
3646
# Note that if there are multiple parents, we need to wait
3648
self._pending_annotation.setdefault(parent_key,
3649
[]).append((key, parent_keys))
3653
def _extract_texts(self, records):
3654
"""Extract the various texts needed based on records"""
2823
def _annotate_records(self, records):
2824
"""Build the annotations for the listed records."""
3655
2825
# We iterate in the order read, rather than a strict order requested
3656
2826
# However, process what we can, and put off to the side things that
3657
2827
# still need parents, cleaning them up when those parents are
3660
# 1) As 'records' are read, see if we can expand these records into
3661
# Content objects (and thus lines)
3662
# 2) If a given line-delta is waiting on its compression parent, it
3663
# gets queued up into self._pending_deltas, otherwise we expand
3664
# it, and put it into self._text_cache and self._content_objects
3665
# 3) If we expanded the text, we will then check to see if all
3666
# parents have also been processed. If so, this text gets yielded,
3667
# else this record gets set aside into pending_annotation
3668
# 4) Further, if we expanded the text in (2), we will then check to
3669
# see if there are any children in self._pending_deltas waiting to
3670
# also be processed. If so, we go back to (2) for those
3671
# 5) Further again, if we yielded the text, we can then check if that
3672
# 'unlocks' any of the texts in pending_annotations, which should
3673
# then get yielded as well
3674
# Note that both steps 4 and 5 are 'recursive' in that unlocking one
3675
# compression child could unlock yet another, and yielding a fulltext
3676
# will also 'unlock' the children that are waiting on that annotation.
3677
# (Though also, unlocking 1 parent's fulltext, does not unlock a child
3678
# if other parents are also waiting.)
3679
# We want to yield content before expanding child content objects, so
3680
# that we know when we can re-use the content lines, and the annotation
3681
# code can know when it can stop caching fulltexts, as well.
3683
# Children that are missing their compression parent
3685
for (key, record, digest) in self._vf._read_records_iter(records):
3687
details = self._all_build_details[key]
3688
(_, compression_parent, parent_keys, record_details) = details
3689
lines = self._expand_record(key, parent_keys, compression_parent,
3690
record, record_details)
3692
# Pending delta should be queued up
2829
for (rev_id, record,
2830
digest) in self._knit._read_records_iter(records):
2831
if rev_id in self._annotated_lines:
3694
# At this point, we may be able to yield this content, if all
3695
# parents are also finished
3696
yield_this_text = self._check_ready_for_annotations(key,
3699
# All parents present
3700
yield key, lines, len(lines)
3701
to_process = self._process_pending(key)
3703
this_process = to_process
3705
for key in this_process:
3706
lines = self._text_cache[key]
3707
yield key, lines, len(lines)
3708
to_process.extend(self._process_pending(key))
2833
parent_ids = self._revision_id_graph[rev_id]
2834
parent_ids = [p for p in parent_ids if p not in self._ghosts]
2835
details = self._all_build_details[rev_id]
2836
(index_memo, compression_parent, parents,
2837
record_details) = details
2838
nodes_to_annotate = []
2839
# TODO: Remove the punning between compression parents, and
2840
# parent_ids, we should be able to do this without assuming
2842
if len(parent_ids) == 0:
2843
# There are no parents for this node, so just add it
2844
# TODO: This probably needs to be decoupled
2845
fulltext_content, delta = self._knit._factory.parse_record(
2846
rev_id, record, record_details, None)
2847
fulltext = self._add_fulltext_content(rev_id, fulltext_content)
2848
nodes_to_annotate.extend(self._add_annotation(rev_id, fulltext,
2849
parent_ids, left_matching_blocks=None))
2851
child = (rev_id, parent_ids, record)
2852
# Check if all the parents are present
2853
self._check_parents(child, nodes_to_annotate)
2854
while nodes_to_annotate:
2855
# Should we use a queue here instead of a stack?
2856
(rev_id, parent_ids, record) = nodes_to_annotate.pop()
2857
(index_memo, compression_parent, parents,
2858
record_details) = self._all_build_details[rev_id]
2860
if compression_parent is not None:
2861
comp_children = self._compression_children[compression_parent]
2862
if rev_id not in comp_children:
2863
raise AssertionError("%r not in compression children %r"
2864
% (rev_id, comp_children))
2865
# If there is only 1 child, it is safe to reuse this
2867
reuse_content = (len(comp_children) == 1
2868
and compression_parent not in
2869
self._nodes_to_keep_annotations)
2871
# Remove it from the cache since it will be changing
2872
parent_fulltext_content = self._fulltext_contents.pop(compression_parent)
2873
# Make sure to copy the fulltext since it might be
2875
parent_fulltext = list(parent_fulltext_content.text())
2877
parent_fulltext_content = self._fulltext_contents[compression_parent]
2878
parent_fulltext = parent_fulltext_content.text()
2879
comp_children.remove(rev_id)
2880
fulltext_content, delta = self._knit._factory.parse_record(
2881
rev_id, record, record_details,
2882
parent_fulltext_content,
2883
copy_base_content=(not reuse_content))
2884
fulltext = self._add_fulltext_content(rev_id,
2886
if compression_parent == parent_ids[0]:
2887
# the compression_parent is the left parent, so we can
2889
blocks = KnitContent.get_line_delta_blocks(delta,
2890
parent_fulltext, fulltext)
2892
fulltext_content = self._knit._factory.parse_fulltext(
2894
fulltext = self._add_fulltext_content(rev_id,
2896
nodes_to_annotate.extend(
2897
self._add_annotation(rev_id, fulltext, parent_ids,
2898
left_matching_blocks=blocks))
2900
def _get_heads_provider(self):
2901
"""Create a heads provider for resolving ancestry issues."""
2902
if self._heads_provider is not None:
2903
return self._heads_provider
2904
parent_provider = _mod_graph.DictParentsProvider(
2905
self._revision_id_graph)
2906
graph_obj = _mod_graph.Graph(parent_provider)
2907
head_cache = _mod_graph.FrozenHeadsCache(graph_obj)
2908
self._heads_provider = head_cache
2911
def annotate(self, key):
2912
"""Return the annotated fulltext at the given key.
2914
:param key: The key to annotate.
2916
if len(self._knit._fallback_vfs) > 0:
2917
# stacked knits can't use the fast path at present.
2918
return self._simple_annotate(key)
2921
records = self._get_build_graph(key)
2922
if key in self._ghosts:
2923
raise errors.RevisionNotPresent(key, self._knit)
2924
self._annotate_records(records)
2925
return self._annotated_lines[key]
2926
except errors.RetryWithNewPacks, e:
2927
self._knit._access.reload_or_raise(e)
2928
# The cached build_details are no longer valid
2929
self._all_build_details.clear()
2931
def _simple_annotate(self, key):
2932
"""Return annotated fulltext, rediffing from the full texts.
2934
This is slow but makes no assumptions about the repository
2935
being able to produce line deltas.
2937
# TODO: this code generates a parent maps of present ancestors; it
2938
# could be split out into a separate method, and probably should use
2939
# iter_ancestry instead. -- mbp and robertc 20080704
2940
graph = _mod_graph.Graph(self._knit)
2941
head_cache = _mod_graph.FrozenHeadsCache(graph)
2942
search = graph._make_breadth_first_searcher([key])
2946
present, ghosts = search.next_with_ghosts()
2947
except StopIteration:
2949
keys.update(present)
2950
parent_map = self._knit.get_parent_map(keys)
2952
reannotate = annotate.reannotate
2953
for record in self._knit.get_record_stream(keys, 'topological', True):
2955
fulltext = split_lines(record.get_bytes_as('fulltext'))
2956
parents = parent_map[key]
2957
if parents is not None:
2958
parent_lines = [parent_cache[parent] for parent in parent_map[key]]
2961
parent_cache[key] = list(
2962
reannotate(parent_lines, fulltext, key, None, head_cache))
2964
return parent_cache[key]
2966
raise errors.RevisionNotPresent(key, self._knit)
3711
from bzrlib._knit_load_data_pyx import _load_data_c as _load_data
3712
except ImportError, e:
3713
osutils.failed_to_load_extension(e)
2970
from bzrlib._knit_load_data_c import _load_data_c as _load_data
3714
2972
from bzrlib._knit_load_data_py import _load_data_py as _load_data
added
removed
bzrlib/knit.py
