« back to all changes in this revision
Viewing changes to bzrlib/knit.py
- browse files at revision 1821.1.1
- compare with another revision
- download diff
- download tarball
- view history from revision 1821.1.1
- Committer: Alexander Belchenko
- Date: 2006-06-29 08:41:31 UTC
- mto: (1860.1.1 win32.installer)
- mto: This revision was merged to the branch mainline in revision 1906.
- Revision ID: bialix@ukr.net-20060629084131-3ea4d44e3204e36f
win32 installer for bzr.dev.0.9
- files added:
- NEWS.developers
- bzrlib/bundle/old
- bzrlib/tests/interversionedfile_implementations
- bzrlib/tests/revisionstore_implementations
- bzrlib/util/configobj/docs
- bzrlib/util/urlgrabber
- files removed:
- COPYING.txt
- bzrlib/benchmarks/tree_creator
- bzrlib/filters
- bzrlib/help_topics
- bzrlib/help_topics/en
- bzrlib/plugins/netrc_credential_store
- bzrlib/plugins/netrc_credential_store/tests
- bzrlib/repofmt
- bzrlib/smart
- bzrlib/tests/commands
- bzrlib/tests/ftp_server
- bzrlib/tests/per_interbranch
- bzrlib/tests/per_intertree
- bzrlib/tests/per_inventory
- bzrlib/tests/per_lock
- bzrlib/tests/per_repository_chk
- bzrlib/tests/per_repository_reference
- bzrlib/tests/per_tree
- bzrlib/tests/ssl_certs
- bzrlib/transport/ftp
- bzrlib/util/tests
- bzrlib/version_info_formats
- doc/_static
- doc/_static/en
- doc/_static/en/quick-reference
- doc/_templates
- doc/developers
- doc/en
- doc/en/admin-guide
- doc/en/developer-guide
- doc/en/mini-tutorial
- doc/en/quick-reference
- doc/en/release-notes
- doc/en/tutorials
- doc/en/upgrade-guide
- doc/en/user-guide
- doc/en/user-guide/images
- doc/en/user-reference
- doc/es
- doc/es/developer-guide
- doc/es/mini-tutorial
- doc/es/quick-reference
- doc/es/release-notes
- doc/es/user-guide
- doc/es/user-reference
- doc/ru
- doc/ru/mini-tutorial
- doc/ru/quick-reference
- doc/ru/tutorials
- doc/ru/user-guide
- doc/ru/user-guide/images
- man1
- tools/packaging
- tools/win32/buildout-templates
- tools/win32/buildout-templates/bin
- files renamed:
- doc/en/developer-guide/HACKING.txt => HACKING
- bzrlib/deprecated_graph.py => bzrlib/graph.py
- bzrlib/_patiencediff_py.py => bzrlib/patiencediff.py
- bzrlib/tests/http_utils.py => bzrlib/tests/HTTPTestUtil.py
- bzrlib/tests/blackbox/test_send.py => bzrlib/tests/blackbox/test_bundle.py
- bzrlib/tests/per_branch/ => bzrlib/tests/branch_implementations/
- bzrlib/tests/per_bzrdir/ => bzrlib/tests/bzrdir_implementations/
- bzrlib/tests/per_interrepository/ => bzrlib/tests/interrepository_implementations/
- bzrlib/tests/per_repository/ => bzrlib/tests/repository_implementations/
- bzrlib/tests/test_commands.py => bzrlib/tests/test_command.py
- bzrlib/tests/test_deprecated_graph.py => bzrlib/tests/test_graph.py
- bzrlib/tests/test_revisionspec.py => bzrlib/tests/test_revisionnamespaces.py
- bzrlib/tests/per_repository/test_revision.py => bzrlib/tests/test_revprops.py
- bzrlib/tests/per_transport.py => bzrlib/tests/test_transport_implementations.py
- bzrlib/tests/per_versionedfile.py => bzrlib/tests/test_versionedfile.py
- bzrlib/tests/per_workingtree/ => bzrlib/tests/workingtree_implementations/
- bzrlib/transport/ftp/__init__.py => bzrlib/transport/ftp.py
- bzrlib/win32utils.py => bzrlib/win32console.py
- bzrlib/xml8.py => bzrlib/xml5.py
- bzrlib/help_topics/en/configuration.txt => doc/configuration.txt
- doc/en/user-guide/index.txt => doc/index.txt
- doc/en/user-guide/plugins.txt => doc/plugins.txt
- doc/en/user-guide/setting_up_email.txt => doc/setting_up_email.txt
- doc/en/user-guide/specifying_revisions.txt => doc/specifying_revisions.txt
- doc/en/tutorials/tutorial.txt => doc/tutorial.txt
- doc/en/user-guide/using_aliases.txt => doc/using_aliases.txt
- tools/generate_docs.py => generate_docs.py
- bzrlib/doc_generate/ => tools/doc_generate/
- files modified:
- .bzrignore
added
removed
bzrlib/knit.py
1
# Copyright (C) 2005, 2006, 2007, 2008 Canonical Ltd
1
# Copyright (C) 2005, 2006 by Canonical Ltd
2
# Written by Martin Pool.
3
# Modified by Johan Rydberg <jrydberg@gnu.org>
4
# Modified by Robert Collins <robert.collins@canonical.com>
5
# Modified by Aaron Bentley <aaron.bentley@utoronto.ca>
2
6
#
3
7
# This program is free software; you can redistribute it and/or modify
4
8
# it under the terms of the GNU General Public License as published by
12
16
#
13
17
# You should have received a copy of the GNU General Public License
14
18
# along with this program; if not, write to the Free Software
15
# Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA
19
# Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307 USA
16
20
17
21
"""Knit versionedfile implementation.
18
22
20
24
updates.
21
25
22
26
Knit file layout:
23
lifeless: the data file is made up of "delta records". each delta record has a delta header
24
that contains; (1) a version id, (2) the size of the delta (in lines), and (3) the digest of
25
the -expanded data- (ie, the delta applied to the parent). the delta also ends with a
27
lifeless: the data file is made up of "delta records". each delta record has a delta header
28
that contains; (1) a version id, (2) the size of the delta (in lines), and (3) the digest of
29
the -expanded data- (ie, the delta applied to the parent). the delta also ends with a
26
30
end-marker; simply "end VERSION"
27
31
28
32
delta can be line or full contents.a
35
39
130,130,2
36
40
8 if elt.get('executable') == 'yes':
37
41
8 ie.executable = True
38
end robertc@robertcollins.net-20051003014215-ee2990904cc4c7ad
42
end robertc@robertcollins.net-20051003014215-ee2990904cc4c7ad
39
43
40
44
41
45
whats in an index:
51
55
52
56
"""
53
57
58
# TODOS:
59
# 10:16 < lifeless> make partial index writes safe
60
# 10:16 < lifeless> implement 'knit.check()' like weave.check()
61
# 10:17 < lifeless> record known ghosts so we can detect when they are filled in rather than the current 'reweave
62
# always' approach.
63
# move sha1 out of the content so that join is faster at verifying parents
64
# record content length ?
65
54
66
67
from copy import copy
55
68
from cStringIO import StringIO
56
from itertools import izip
69
import difflib
70
from itertools import izip, chain
57
71
import operator
58
72
import os
59
73
import sys
74
import warnings
60
75
61
from bzrlib.lazy_import import lazy_import
62
lazy_import(globals(), """
63
from bzrlib import (
64
annotate,
65
debug,
66
diff,
67
graph as _mod_graph,
68
index as _mod_index,
69
lru_cache,
70
pack,
71
progress,
72
trace,
73
tsort,
74
tuned_gzip,
75
)
76
""")
77
from bzrlib import (
78
errors,
79
osutils,
80
patiencediff,
81
)
82
from bzrlib.errors import (
83
FileExists,
84
NoSuchFile,
85
KnitError,
86
InvalidRevisionId,
87
KnitCorrupt,
88
KnitHeaderError,
89
RevisionNotPresent,
90
RevisionAlreadyPresent,
91
SHA1KnitCorrupt,
92
)
93
from bzrlib.osutils import (
94
contains_whitespace,
95
contains_linebreaks,
96
sha_string,
97
sha_strings,
98
split_lines,
99
)
100
from bzrlib.versionedfile import (
101
AbsentContentFactory,
102
adapter_registry,
103
ConstantMapper,
104
ContentFactory,
105
ChunkedContentFactory,
106
sort_groupcompress,
107
VersionedFile,
108
VersionedFiles,
109
)
76
import bzrlib
77
import bzrlib.errors as errors
78
from bzrlib.errors import FileExists, NoSuchFile, KnitError, \
79
InvalidRevisionId, KnitCorrupt, KnitHeaderError, \
80
RevisionNotPresent, RevisionAlreadyPresent
81
from bzrlib.tuned_gzip import GzipFile
82
from bzrlib.trace import mutter
83
from bzrlib.osutils import contains_whitespace, contains_linebreaks, \
84
sha_strings
85
from bzrlib.versionedfile import VersionedFile, InterVersionedFile
86
from bzrlib.symbol_versioning import DEPRECATED_PARAMETER, deprecated_passed
87
from bzrlib.tsort import topo_sort
88
import bzrlib.weave
110
89
111
90
112
91
# TODO: Split out code specific to this format into an associated object.
124
103
125
104
DATA_SUFFIX = '.knit'
126
105
INDEX_SUFFIX = '.kndx'
127
_STREAM_MIN_BUFFER_SIZE = 5*1024*1024
128
129
130
class KnitAdapter(object):
131
"""Base class for knit record adaption."""
132
133
def __init__(self, basis_vf):
134
"""Create an adapter which accesses full texts from basis_vf.
135
136
:param basis_vf: A versioned file to access basis texts of deltas from.
137
May be None for adapters that do not need to access basis texts.
138
"""
139
self._data = KnitVersionedFiles(None, None)
140
self._annotate_factory = KnitAnnotateFactory()
141
self._plain_factory = KnitPlainFactory()
142
self._basis_vf = basis_vf
143
144
145
class FTAnnotatedToUnannotated(KnitAdapter):
146
"""An adapter from FT annotated knits to unannotated ones."""
147
148
def get_bytes(self, factory):
149
annotated_compressed_bytes = factory._raw_record
150
rec, contents = \
151
self._data._parse_record_unchecked(annotated_compressed_bytes)
152
content = self._annotate_factory.parse_fulltext(contents, rec[1])
153
size, bytes = self._data._record_to_data((rec[1],), rec[3], content.text())
154
return bytes
155
156
157
class DeltaAnnotatedToUnannotated(KnitAdapter):
158
"""An adapter for deltas from annotated to unannotated."""
159
160
def get_bytes(self, factory):
161
annotated_compressed_bytes = factory._raw_record
162
rec, contents = \
163
self._data._parse_record_unchecked(annotated_compressed_bytes)
164
delta = self._annotate_factory.parse_line_delta(contents, rec[1],
165
plain=True)
166
contents = self._plain_factory.lower_line_delta(delta)
167
size, bytes = self._data._record_to_data((rec[1],), rec[3], contents)
168
return bytes
169
170
171
class FTAnnotatedToFullText(KnitAdapter):
172
"""An adapter from FT annotated knits to unannotated ones."""
173
174
def get_bytes(self, factory):
175
annotated_compressed_bytes = factory._raw_record
176
rec, contents = \
177
self._data._parse_record_unchecked(annotated_compressed_bytes)
178
content, delta = self._annotate_factory.parse_record(factory.key[-1],
179
contents, factory._build_details, None)
180
return ''.join(content.text())
181
182
183
class DeltaAnnotatedToFullText(KnitAdapter):
184
"""An adapter for deltas from annotated to unannotated."""
185
186
def get_bytes(self, factory):
187
annotated_compressed_bytes = factory._raw_record
188
rec, contents = \
189
self._data._parse_record_unchecked(annotated_compressed_bytes)
190
delta = self._annotate_factory.parse_line_delta(contents, rec[1],
191
plain=True)
192
compression_parent = factory.parents[0]
193
basis_entry = self._basis_vf.get_record_stream(
194
[compression_parent], 'unordered', True).next()
195
if basis_entry.storage_kind == 'absent':
196
raise errors.RevisionNotPresent(compression_parent, self._basis_vf)
197
basis_chunks = basis_entry.get_bytes_as('chunked')
198
basis_lines = osutils.chunks_to_lines(basis_chunks)
199
# Manually apply the delta because we have one annotated content and
200
# one plain.
201
basis_content = PlainKnitContent(basis_lines, compression_parent)
202
basis_content.apply_delta(delta, rec[1])
203
basis_content._should_strip_eol = factory._build_details[1]
204
return ''.join(basis_content.text())
205
206
207
class FTPlainToFullText(KnitAdapter):
208
"""An adapter from FT plain knits to unannotated ones."""
209
210
def get_bytes(self, factory):
211
compressed_bytes = factory._raw_record
212
rec, contents = \
213
self._data._parse_record_unchecked(compressed_bytes)
214
content, delta = self._plain_factory.parse_record(factory.key[-1],
215
contents, factory._build_details, None)
216
return ''.join(content.text())
217
218
219
class DeltaPlainToFullText(KnitAdapter):
220
"""An adapter for deltas from annotated to unannotated."""
221
222
def get_bytes(self, factory):
223
compressed_bytes = factory._raw_record
224
rec, contents = \
225
self._data._parse_record_unchecked(compressed_bytes)
226
delta = self._plain_factory.parse_line_delta(contents, rec[1])
227
compression_parent = factory.parents[0]
228
# XXX: string splitting overhead.
229
basis_entry = self._basis_vf.get_record_stream(
230
[compression_parent], 'unordered', True).next()
231
if basis_entry.storage_kind == 'absent':
232
raise errors.RevisionNotPresent(compression_parent, self._basis_vf)
233
basis_chunks = basis_entry.get_bytes_as('chunked')
234
basis_lines = osutils.chunks_to_lines(basis_chunks)
235
basis_content = PlainKnitContent(basis_lines, compression_parent)
236
# Manually apply the delta because we have one annotated content and
237
# one plain.
238
content, _ = self._plain_factory.parse_record(rec[1], contents,
239
factory._build_details, basis_content)
240
return ''.join(content.text())
241
242
243
class KnitContentFactory(ContentFactory):
244
"""Content factory for streaming from knits.
245
246
:seealso ContentFactory:
247
"""
248
249
def __init__(self, key, parents, build_details, sha1, raw_record,
250
annotated, knit=None, network_bytes=None):
251
"""Create a KnitContentFactory for key.
252
253
:param key: The key.
254
:param parents: The parents.
255
:param build_details: The build details as returned from
256
get_build_details.
257
:param sha1: The sha1 expected from the full text of this object.
258
:param raw_record: The bytes of the knit data from disk.
259
:param annotated: True if the raw data is annotated.
260
:param network_bytes: None to calculate the network bytes on demand,
261
not-none if they are already known.
262
"""
263
ContentFactory.__init__(self)
264
self.sha1 = sha1
265
self.key = key
266
self.parents = parents
267
if build_details[0] == 'line-delta':
268
kind = 'delta'
269
else:
270
kind = 'ft'
271
if annotated:
272
annotated_kind = 'annotated-'
273
else:
274
annotated_kind = ''
275
self.storage_kind = 'knit-%s%s-gz' % (annotated_kind, kind)
276
self._raw_record = raw_record
277
self._network_bytes = network_bytes
278
self._build_details = build_details
279
self._knit = knit
280
281
def _create_network_bytes(self):
282
"""Create a fully serialised network version for transmission."""
283
# storage_kind, key, parents, Noeol, raw_record
284
key_bytes = '\x00'.join(self.key)
285
if self.parents is None:
286
parent_bytes = 'None:'
287
else:
288
parent_bytes = '\t'.join('\x00'.join(key) for key in self.parents)
289
if self._build_details[1]:
290
noeol = 'N'
291
else:
292
noeol = ' '
293
network_bytes = "%s\n%s\n%s\n%s%s" % (self.storage_kind, key_bytes,
294
parent_bytes, noeol, self._raw_record)
295
self._network_bytes = network_bytes
296
297
def get_bytes_as(self, storage_kind):
298
if storage_kind == self.storage_kind:
299
if self._network_bytes is None:
300
self._create_network_bytes()
301
return self._network_bytes
302
if ('-ft-' in self.storage_kind and
303
storage_kind in ('chunked', 'fulltext')):
304
adapter_key = (self.storage_kind, 'fulltext')
305
adapter_factory = adapter_registry.get(adapter_key)
306
adapter = adapter_factory(None)
307
bytes = adapter.get_bytes(self)
308
if storage_kind == 'chunked':
309
return [bytes]
310
else:
311
return bytes
312
if self._knit is not None:
313
# Not redundant with direct conversion above - that only handles
314
# fulltext cases.
315
if storage_kind == 'chunked':
316
return self._knit.get_lines(self.key[0])
317
elif storage_kind == 'fulltext':
318
return self._knit.get_text(self.key[0])
319
raise errors.UnavailableRepresentation(self.key, storage_kind,
320
self.storage_kind)
321
322
323
class LazyKnitContentFactory(ContentFactory):
324
"""A ContentFactory which can either generate full text or a wire form.
325
326
:seealso ContentFactory:
327
"""
328
329
def __init__(self, key, parents, generator, first):
330
"""Create a LazyKnitContentFactory.
331
332
:param key: The key of the record.
333
:param parents: The parents of the record.
334
:param generator: A _ContentMapGenerator containing the record for this
335
key.
336
:param first: Is this the first content object returned from generator?
337
if it is, its storage kind is knit-delta-closure, otherwise it is
338
knit-delta-closure-ref
339
"""
340
self.key = key
341
self.parents = parents
342
self.sha1 = None
343
self._generator = generator
344
self.storage_kind = "knit-delta-closure"
345
if not first:
346
self.storage_kind = self.storage_kind + "-ref"
347
self._first = first
348
349
def get_bytes_as(self, storage_kind):
350
if storage_kind == self.storage_kind:
351
if self._first:
352
return self._generator._wire_bytes()
353
else:
354
# all the keys etc are contained in the bytes returned in the
355
# first record.
356
return ''
357
if storage_kind in ('chunked', 'fulltext'):
358
chunks = self._generator._get_one_work(self.key).text()
359
if storage_kind == 'chunked':
360
return chunks
361
else:
362
return ''.join(chunks)
363
raise errors.UnavailableRepresentation(self.key, storage_kind,
364
self.storage_kind)
365
366
367
def knit_delta_closure_to_records(storage_kind, bytes, line_end):
368
"""Convert a network record to a iterator over stream records.
369
370
:param storage_kind: The storage kind of the record.
371
Must be 'knit-delta-closure'.
372
:param bytes: The bytes of the record on the network.
373
"""
374
generator = _NetworkContentMapGenerator(bytes, line_end)
375
return generator.get_record_stream()
376
377
378
def knit_network_to_record(storage_kind, bytes, line_end):
379
"""Convert a network record to a record object.
380
381
:param storage_kind: The storage kind of the record.
382
:param bytes: The bytes of the record on the network.
383
"""
384
start = line_end
385
line_end = bytes.find('\n', start)
386
key = tuple(bytes[start:line_end].split('\x00'))
387
start = line_end + 1
388
line_end = bytes.find('\n', start)
389
parent_line = bytes[start:line_end]
390
if parent_line == 'None:':
391
parents = None
392
else:
393
parents = tuple(
394
[tuple(segment.split('\x00')) for segment in parent_line.split('\t')
395
if segment])
396
start = line_end + 1
397
noeol = bytes[start] == 'N'
398
if 'ft' in storage_kind:
399
method = 'fulltext'
400
else:
401
method = 'line-delta'
402
build_details = (method, noeol)
403
start = start + 1
404
raw_record = bytes[start:]
405
annotated = 'annotated' in storage_kind
406
return [KnitContentFactory(key, parents, build_details, None, raw_record,
407
annotated, network_bytes=bytes)]
408
106
409
107
410
108
class KnitContent(object):
411
"""Content of a knit version to which deltas can be applied.
412
413
This is always stored in memory as a list of lines with \n at the end,
414
plus a flag saying if the final ending is really there or not, because that
415
corresponds to the on-disk knit representation.
416
"""
417
418
def __init__(self):
419
self._should_strip_eol = False
420
421
def apply_delta(self, delta, new_version_id):
422
"""Apply delta to this object to become new_version_id."""
423
raise NotImplementedError(self.apply_delta)
109
"""Content of a knit version to which deltas can be applied."""
110
111
def __init__(self, lines):
112
self._lines = lines
113
114
def annotate_iter(self):
115
"""Yield tuples of (origin, text) for each content line."""
116
for origin, text in self._lines:
117
yield origin, text
118
119
def annotate(self):
120
"""Return a list of (origin, text) tuples."""
121
return list(self.annotate_iter())
424
122
425
123
def line_delta_iter(self, new_lines):
426
124
"""Generate line-based delta from this content to new_lines."""
427
new_texts = new_lines.text()
428
old_texts = self.text()
429
s = patiencediff.PatienceSequenceMatcher(None, old_texts, new_texts)
430
for tag, i1, i2, j1, j2 in s.get_opcodes():
431
if tag == 'equal':
125
new_texts = [text for origin, text in new_lines._lines]
126
old_texts = [text for origin, text in self._lines]
127
s = KnitSequenceMatcher(None, old_texts, new_texts)
128
for op in s.get_opcodes():
129
if op[0] == 'equal':
432
130
continue
433
# ofrom, oto, length, data
434
yield i1, i2, j2 - j1, new_lines._lines[j1:j2]
131
# ofrom oto length data
132
yield (op[1], op[2], op[4]-op[3], new_lines._lines[op[3]:op[4]])
435
133
436
134
def line_delta(self, new_lines):
437
135
return list(self.line_delta_iter(new_lines))
438
136
439
@staticmethod
440
def get_line_delta_blocks(knit_delta, source, target):
441
"""Extract SequenceMatcher.get_matching_blocks() from a knit delta"""
442
target_len = len(target)
443
s_pos = 0
444
t_pos = 0
445
for s_begin, s_end, t_len, new_text in knit_delta:
446
true_n = s_begin - s_pos
447
n = true_n
448
if n > 0:
449
# knit deltas do not provide reliable info about whether the
450
# last line of a file matches, due to eol handling.
451
if source[s_pos + n -1] != target[t_pos + n -1]:
452
n-=1
453
if n > 0:
454
yield s_pos, t_pos, n
455
t_pos += t_len + true_n
456
s_pos = s_end
457
n = target_len - t_pos
458
if n > 0:
459
if source[s_pos + n -1] != target[t_pos + n -1]:
460
n-=1
461
if n > 0:
462
yield s_pos, t_pos, n
463
yield s_pos + (target_len - t_pos), target_len, 0
464
465
466
class AnnotatedKnitContent(KnitContent):
467
"""Annotated content."""
468
469
def __init__(self, lines):
470
KnitContent.__init__(self)
471
self._lines = lines
472
473
def annotate(self):
474
"""Return a list of (origin, text) for each content line."""
475
lines = self._lines[:]
476
if self._should_strip_eol:
477
origin, last_line = lines[-1]
478
lines[-1] = (origin, last_line.rstrip('\n'))
479
return lines
480
481
def apply_delta(self, delta, new_version_id):
482
"""Apply delta to this object to become new_version_id."""
483
offset = 0
484
lines = self._lines
485
for start, end, count, delta_lines in delta:
486
lines[offset+start:offset+end] = delta_lines
487
offset = offset + (start - end) + count
488
489
def text(self):
490
try:
491
lines = [text for origin, text in self._lines]
492
except ValueError, e:
493
# most commonly (only?) caused by the internal form of the knit
494
# missing annotation information because of a bug - see thread
495
# around 20071015
496
raise KnitCorrupt(self,
497
"line in annotated knit missing annotation information: %s"
498
% (e,))
499
if self._should_strip_eol:
500
lines[-1] = lines[-1].rstrip('\n')
501
return lines
502
503
def copy(self):
504
return AnnotatedKnitContent(self._lines[:])
505
506
507
class PlainKnitContent(KnitContent):
508
"""Unannotated content.
509
510
When annotate[_iter] is called on this content, the same version is reported
511
for all lines. Generally, annotate[_iter] is not useful on PlainKnitContent
512
objects.
513
"""
514
515
def __init__(self, lines, version_id):
516
KnitContent.__init__(self)
517
self._lines = lines
518
self._version_id = version_id
519
520
def annotate(self):
521
"""Return a list of (origin, text) for each content line."""
522
return [(self._version_id, line) for line in self._lines]
523
524
def apply_delta(self, delta, new_version_id):
525
"""Apply delta to this object to become new_version_id."""
526
offset = 0
527
lines = self._lines
528
for start, end, count, delta_lines in delta:
529
lines[offset+start:offset+end] = delta_lines
530
offset = offset + (start - end) + count
531
self._version_id = new_version_id
532
533
def copy(self):
534
return PlainKnitContent(self._lines[:], self._version_id)
535
536
def text(self):
537
lines = self._lines
538
if self._should_strip_eol:
539
lines = lines[:]
540
lines[-1] = lines[-1].rstrip('\n')
541
return lines
137
def text(self):
138
return [text for origin, text in self._lines]
139
140
def copy(self):
141
return KnitContent(self._lines[:])
542
142
543
143
544
144
class _KnitFactory(object):
545
"""Base class for common Factory functions."""
546
547
def parse_record(self, version_id, record, record_details,
548
base_content, copy_base_content=True):
549
"""Parse a record into a full content object.
550
551
:param version_id: The official version id for this content
552
:param record: The data returned by read_records_iter()
553
:param record_details: Details about the record returned by
554
get_build_details
555
:param base_content: If get_build_details returns a compression_parent,
556
you must return a base_content here, else use None
557
:param copy_base_content: When building from the base_content, decide
558
you can either copy it and return a new object, or modify it in
559
place.
560
:return: (content, delta) A Content object and possibly a line-delta,
561
delta may be None
562
"""
563
method, noeol = record_details
564
if method == 'line-delta':
565
if copy_base_content:
566
content = base_content.copy()
567
else:
568
content = base_content
569
delta = self.parse_line_delta(record, version_id)
570
content.apply_delta(delta, version_id)
571
else:
572
content = self.parse_fulltext(record, version_id)
573
delta = None
574
content._should_strip_eol = noeol
575
return (content, delta)
145
"""Base factory for creating content objects."""
146
147
def make(self, lines, version):
148
num_lines = len(lines)
149
return KnitContent(zip([version] * num_lines, lines))
576
150
577
151
578
152
class KnitAnnotateFactory(_KnitFactory):
580
154
581
155
annotated = True
582
156
583
def make(self, lines, version_id):
584
num_lines = len(lines)
585
return AnnotatedKnitContent(zip([version_id] * num_lines, lines))
586
587
def parse_fulltext(self, content, version_id):
157
def parse_fulltext(self, content, version):
588
158
"""Convert fulltext to internal representation
589
159
590
160
fulltext content is of the format
592
162
internal representation is of the format:
593
163
(revid, plaintext)
594
164
"""
595
# TODO: jam 20070209 The tests expect this to be returned as tuples,
596
# but the code itself doesn't really depend on that.
597
# Figure out a way to not require the overhead of turning the
598
# list back into tuples.
599
lines = [tuple(line.split(' ', 1)) for line in content]
600
return AnnotatedKnitContent(lines)
165
lines = []
166
for line in content:
167
origin, text = line.split(' ', 1)
168
lines.append((origin.decode('utf-8'), text))
169
return KnitContent(lines)
601
170
602
171
def parse_line_delta_iter(self, lines):
603
return iter(self.parse_line_delta(lines))
172
for result_item in self.parse_line_delta[lines]:
173
yield result_item
604
174
605
def parse_line_delta(self, lines, version_id, plain=False):
175
def parse_line_delta(self, lines, version):
606
176
"""Convert a line based delta into internal representation.
607
177
608
178
line delta is in the form of:
611
181
revid(utf8) newline\n
612
182
internal representation is
613
183
(start, end, count, [1..count tuples (revid, newline)])
614
615
:param plain: If True, the lines are returned as a plain
616
list without annotations, not as a list of (origin, content) tuples, i.e.
617
(start, end, count, [1..count newline])
618
184
"""
619
185
result = []
620
186
lines = iter(lines)
621
187
next = lines.next
622
623
cache = {}
624
def cache_and_return(line):
625
origin, text = line.split(' ', 1)
626
return cache.setdefault(origin, origin), text
627
628
188
# walk through the lines parsing.
629
# Note that the plain test is explicitly pulled out of the
630
# loop to minimise any performance impact
631
if plain:
632
for header in lines:
633
start, end, count = [int(n) for n in header.split(',')]
634
contents = [next().split(' ', 1)[1] for i in xrange(count)]
635
result.append((start, end, count, contents))
636
else:
637
for header in lines:
638
start, end, count = [int(n) for n in header.split(',')]
639
contents = [tuple(next().split(' ', 1)) for i in xrange(count)]
640
result.append((start, end, count, contents))
641
return result
642
643
def get_fulltext_content(self, lines):
644
"""Extract just the content lines from a fulltext."""
645
return (line.split(' ', 1)[1] for line in lines)
646
647
def get_linedelta_content(self, lines):
648
"""Extract just the content from a line delta.
649
650
This doesn't return all of the extra information stored in a delta.
651
Only the actual content lines.
652
"""
653
lines = iter(lines)
654
next = lines.next
655
189
for header in lines:
656
header = header.split(',')
657
count = int(header[2])
658
for i in xrange(count):
190
start, end, count = [int(n) for n in header.split(',')]
191
contents = []
192
remaining = count
193
while remaining:
659
194
origin, text = next().split(' ', 1)
660
yield text
195
remaining -= 1
196
contents.append((origin.decode('utf-8'), text))
197
result.append((start, end, count, contents))
198
return result
661
199
662
200
def lower_fulltext(self, content):
663
201
"""convert a fulltext content record into a serializable form.
664
202
665
203
see parse_fulltext which this inverts.
666
204
"""
667
return ['%s %s' % (o, t) for o, t in content._lines]
205
return ['%s %s' % (o.encode('utf-8'), t) for o, t in content._lines]
668
206
669
207
def lower_line_delta(self, delta):
670
208
"""convert a delta into a serializable form.
671
209
672
210
See parse_line_delta which this inverts.
673
211
"""
674
# TODO: jam 20070209 We only do the caching thing to make sure that
675
# the origin is a valid utf-8 line, eventually we could remove it
676
212
out = []
677
213
for start, end, c, lines in delta:
678
214
out.append('%d,%d,%d\n' % (start, end, c))
679
out.extend(origin + ' ' + text
680
for origin, text in lines)
215
for origin, text in lines:
216
out.append('%s %s' % (origin.encode('utf-8'), text))
681
217
return out
682
218
683
def annotate(self, knit, key):
684
content = knit._get_content(key)
685
# adjust for the fact that serialised annotations are only key suffixes
686
# for this factory.
687
if type(key) is tuple:
688
prefix = key[:-1]
689
origins = content.annotate()
690
result = []
691
for origin, line in origins:
692
result.append((prefix + (origin,), line))
693
return result
694
else:
695
# XXX: This smells a bit. Why would key ever be a non-tuple here?
696
# Aren't keys defined to be tuples? -- spiv 20080618
697
return content.annotate()
698
699
219
700
220
class KnitPlainFactory(_KnitFactory):
701
221
"""Factory for creating plain Content objects."""
702
222
703
223
annotated = False
704
224
705
def make(self, lines, version_id):
706
return PlainKnitContent(lines, version_id)
707
708
def parse_fulltext(self, content, version_id):
225
def parse_fulltext(self, content, version):
709
226
"""This parses an unannotated fulltext.
710
227
711
228
Note that this is not a noop - the internal representation
712
229
has (versionid, line) - its just a constant versionid.
713
230
"""
714
return self.make(content, version_id)
231
return self.make(content, version)
715
232
716
def parse_line_delta_iter(self, lines, version_id):
717
cur = 0
718
num_lines = len(lines)
719
while cur < num_lines:
720
header = lines[cur]
721
cur += 1
233
def parse_line_delta_iter(self, lines, version):
234
while lines:
235
header = lines.pop(0)
722
236
start, end, c = [int(n) for n in header.split(',')]
723
yield start, end, c, lines[cur:cur+c]
724
cur += c
725
726
def parse_line_delta(self, lines, version_id):
727
return list(self.parse_line_delta_iter(lines, version_id))
728
729
def get_fulltext_content(self, lines):
730
"""Extract just the content lines from a fulltext."""
731
return iter(lines)
732
733
def get_linedelta_content(self, lines):
734
"""Extract just the content from a line delta.
735
736
This doesn't return all of the extra information stored in a delta.
737
Only the actual content lines.
738
"""
739
lines = iter(lines)
740
next = lines.next
741
for header in lines:
742
header = header.split(',')
743
count = int(header[2])
744
for i in xrange(count):
745
yield next()
746
237
yield start, end, c, zip([version] * c, lines[:c])
238
del lines[:c]
239
240
def parse_line_delta(self, lines, version):
241
return list(self.parse_line_delta_iter(lines, version))
242
747
243
def lower_fulltext(self, content):
748
244
return content.text()
749
245
751
247
out = []
752
248
for start, end, c, lines in delta:
753
249
out.append('%d,%d,%d\n' % (start, end, c))
754
out.extend(lines)
250
out.extend([text for origin, text in lines])
755
251
return out
756
252
757
def annotate(self, knit, key):
758
annotator = _KnitAnnotator(knit)
759
return annotator.annotate_flat(key)
760
761
762
763
def make_file_factory(annotated, mapper):
764
"""Create a factory for creating a file based KnitVersionedFiles.
765
766
This is only functional enough to run interface tests, it doesn't try to
767
provide a full pack environment.
768
769
:param annotated: knit annotations are wanted.
770
:param mapper: The mapper from keys to paths.
771
"""
772
def factory(transport):
773
index = _KndxIndex(transport, mapper, lambda:None, lambda:True, lambda:True)
774
access = _KnitKeyAccess(transport, mapper)
775
return KnitVersionedFiles(index, access, annotated=annotated)
776
return factory
777
778
779
def make_pack_factory(graph, delta, keylength):
780
"""Create a factory for creating a pack based VersionedFiles.
781
782
This is only functional enough to run interface tests, it doesn't try to
783
provide a full pack environment.
784
785
:param graph: Store a graph.
786
:param delta: Delta compress contents.
787
:param keylength: How long should keys be.
788
"""
789
def factory(transport):
790
parents = graph or delta
791
ref_length = 0
792
if graph:
793
ref_length += 1
794
if delta:
795
ref_length += 1
796
max_delta_chain = 200
797
else:
798
max_delta_chain = 0
799
graph_index = _mod_index.InMemoryGraphIndex(reference_lists=ref_length,
800
key_elements=keylength)
801
stream = transport.open_write_stream('newpack')
802
writer = pack.ContainerWriter(stream.write)
803
writer.begin()
804
index = _KnitGraphIndex(graph_index, lambda:True, parents=parents,
805
deltas=delta, add_callback=graph_index.add_nodes)
806
access = _DirectPackAccess({})
807
access.set_writer(writer, graph_index, (transport, 'newpack'))
808
result = KnitVersionedFiles(index, access,
809
max_delta_chain=max_delta_chain)
810
result.stream = stream
811
result.writer = writer
812
return result
813
return factory
814
815
816
def cleanup_pack_knit(versioned_files):
817
versioned_files.stream.close()
818
versioned_files.writer.end()
819
820
821
def _get_total_build_size(self, keys, positions):
822
"""Determine the total bytes to build these keys.
823
824
(helper function because _KnitGraphIndex and _KndxIndex work the same, but
825
don't inherit from a common base.)
826
827
:param keys: Keys that we want to build
828
:param positions: dict of {key, (info, index_memo, comp_parent)} (such
829
as returned by _get_components_positions)
830
:return: Number of bytes to build those keys
831
"""
832
all_build_index_memos = {}
833
build_keys = keys
834
while build_keys:
835
next_keys = set()
836
for key in build_keys:
837
# This is mostly for the 'stacked' case
838
# Where we will be getting the data from a fallback
839
if key not in positions:
840
continue
841
_, index_memo, compression_parent = positions[key]
842
all_build_index_memos[key] = index_memo
843
if compression_parent not in all_build_index_memos:
844
next_keys.add(compression_parent)
845
build_keys = next_keys
846
return sum([index_memo[2] for index_memo
847
in all_build_index_memos.itervalues()])
848
849
850
class KnitVersionedFiles(VersionedFiles):
851
"""Storage for many versioned files using knit compression.
852
853
Backend storage is managed by indices and data objects.
854
855
:ivar _index: A _KnitGraphIndex or similar that can describe the
856
parents, graph, compression and data location of entries in this
857
KnitVersionedFiles. Note that this is only the index for
858
*this* vfs; if there are fallbacks they must be queried separately.
859
"""
860
861
def __init__(self, index, data_access, max_delta_chain=200,
862
annotated=False, reload_func=None):
863
"""Create a KnitVersionedFiles with index and data_access.
864
865
:param index: The index for the knit data.
866
:param data_access: The access object to store and retrieve knit
867
records.
868
:param max_delta_chain: The maximum number of deltas to permit during
869
insertion. Set to 0 to prohibit the use of deltas.
870
:param annotated: Set to True to cause annotations to be calculated and
871
stored during insertion.
872
:param reload_func: An function that can be called if we think we need
873
to reload the pack listing and try again. See
874
'bzrlib.repofmt.pack_repo.AggregateIndex' for the signature.
253
254
def make_empty_knit(transport, relpath):
255
"""Construct a empty knit at the specified location."""
256
k = KnitVersionedFile(transport, relpath, 'w', KnitPlainFactory)
257
k._data._open_file()
258
259
260
class KnitVersionedFile(VersionedFile):
261
"""Weave-like structure with faster random access.
262
263
A knit stores a number of texts and a summary of the relationships
264
between them. Texts are identified by a string version-id. Texts
265
are normally stored and retrieved as a series of lines, but can
266
also be passed as single strings.
267
268
Lines are stored with the trailing newline (if any) included, to
269
avoid special cases for files with no final newline. Lines are
270
composed of 8-bit characters, not unicode. The combination of
271
these approaches should mean any 'binary' file can be safely
272
stored and retrieved.
273
"""
274
275
def __init__(self, relpath, transport, file_mode=None, access_mode=None,
276
factory=None, basis_knit=DEPRECATED_PARAMETER, delta=True,
277
create=False):
278
"""Construct a knit at location specified by relpath.
279
280
:param create: If not True, only open an existing knit.
875
281
"""
876
self._index = index
877
self._access = data_access
878
self._max_delta_chain = max_delta_chain
879
if annotated:
880
self._factory = KnitAnnotateFactory()
881
else:
882
self._factory = KnitPlainFactory()
883
self._fallback_vfs = []
884
self._reload_func = reload_func
282
if deprecated_passed(basis_knit):
283
warnings.warn("KnitVersionedFile.__(): The basis_knit parameter is"
284
" deprecated as of bzr 0.9.",
285
DeprecationWarning, stacklevel=2)
286
if access_mode is None:
287
access_mode = 'w'
288
super(KnitVersionedFile, self).__init__(access_mode)
289
assert access_mode in ('r', 'w'), "invalid mode specified %r" % access_mode
290
self.transport = transport
291
self.filename = relpath
292
self.factory = factory or KnitAnnotateFactory()
293
self.writable = (access_mode == 'w')
294
self.delta = delta
295
296
self._index = _KnitIndex(transport, relpath + INDEX_SUFFIX,
297
access_mode, create=create, file_mode=file_mode)
298
self._data = _KnitData(transport, relpath + DATA_SUFFIX,
299
access_mode, create=create and not len(self), file_mode=file_mode)
885
300
886
301
def __repr__(self):
887
return "%s(%r, %r)" % (
888
self.__class__.__name__,
889
self._index,
890
self._access)
891
892
def add_fallback_versioned_files(self, a_versioned_files):
893
"""Add a source of texts for texts not present in this knit.
894
895
:param a_versioned_files: A VersionedFiles object.
896
"""
897
self._fallback_vfs.append(a_versioned_files)
898
899
def add_lines(self, key, parents, lines, parent_texts=None,
900
left_matching_blocks=None, nostore_sha=None, random_id=False,
901
check_content=True):
902
"""See VersionedFiles.add_lines()."""
903
self._index._check_write_ok()
904
self._check_add(key, lines, random_id, check_content)
905
if parents is None:
906
# The caller might pass None if there is no graph data, but kndx
907
# indexes can't directly store that, so we give them
908
# an empty tuple instead.
909
parents = ()
910
line_bytes = ''.join(lines)
911
return self._add(key, lines, parents,
912
parent_texts, left_matching_blocks, nostore_sha, random_id,
913
line_bytes=line_bytes)
914
915
def _add_text(self, key, parents, text, nostore_sha=None, random_id=False):
916
"""See VersionedFiles._add_text()."""
917
self._index._check_write_ok()
918
self._check_add(key, None, random_id, check_content=False)
919
if text.__class__ is not str:
920
raise errors.BzrBadParameterUnicode("text")
921
if parents is None:
922
# The caller might pass None if there is no graph data, but kndx
923
# indexes can't directly store that, so we give them
924
# an empty tuple instead.
925
parents = ()
926
return self._add(key, None, parents,
927
None, None, nostore_sha, random_id,
928
line_bytes=text)
929
930
def _add(self, key, lines, parents, parent_texts,
931
left_matching_blocks, nostore_sha, random_id,
932
line_bytes):
933
"""Add a set of lines on top of version specified by parents.
934
935
Any versions not present will be converted into ghosts.
936
937
:param lines: A list of strings where each one is a single line (has a
938
single newline at the end of the string) This is now optional
939
(callers can pass None). It is left in its location for backwards
940
compatibility. It should ''.join(lines) must == line_bytes
941
:param line_bytes: A single string containing the content
942
943
We pass both lines and line_bytes because different routes bring the
944
values to this function. And for memory efficiency, we don't want to
945
have to split/join on-demand.
946
"""
947
# first thing, if the content is something we don't need to store, find
948
# that out.
949
digest = sha_string(line_bytes)
950
if nostore_sha == digest:
951
raise errors.ExistingContent
952
302
return '%s(%s)' % (self.__class__.__name__,
303
self.transport.abspath(self.filename))
304
305
def _add_delta(self, version_id, parents, delta_parent, sha1, noeol, delta):
306
"""See VersionedFile._add_delta()."""
307
self._check_add(version_id, []) # should we check the lines ?
308
self._check_versions_present(parents)
953
309
present_parents = []
954
if parent_texts is None:
955
parent_texts = {}
956
# Do a single query to ascertain parent presence; we only compress
957
# against parents in the same kvf.
958
present_parent_map = self._index.get_parent_map(parents)
310
ghosts = []
311
parent_texts = {}
959
312
for parent in parents:
960
if parent in present_parent_map:
313
if not self.has_version(parent):
314
ghosts.append(parent)
315
else:
961
316
present_parents.append(parent)
962
317
963
# Currently we can only compress against the left most present parent.
964
if (len(present_parents) == 0 or
965
present_parents[0] != parents[0]):
966
delta = False
967
else:
318
if delta_parent is None:
319
# reconstitute as full text.
320
assert len(delta) == 1 or len(delta) == 0
321
if len(delta):
322
assert delta[0][0] == 0
323
assert delta[0][1] == 0, delta[0][1]
324
return super(KnitVersionedFile, self)._add_delta(version_id,
325
parents,
326
delta_parent,
327
sha1,
328
noeol,
329
delta)
330
331
digest = sha1
332
333
options = []
334
if noeol:
335
options.append('no-eol')
336
337
if delta_parent is not None:
338
# determine the current delta chain length.
968
339
# To speed the extract of texts the delta chain is limited
969
340
# to a fixed number of deltas. This should minimize both
970
341
# I/O and the time spend applying deltas.
971
delta = self._check_should_delta(present_parents[0])
972
973
text_length = len(line_bytes)
974
options = []
975
no_eol = False
976
# Note: line_bytes is not modified to add a newline, that is tracked
977
# via the no_eol flag. 'lines' *is* modified, because that is the
978
# general values needed by the Content code.
979
if line_bytes and line_bytes[-1] != '\n':
980
options.append('no-eol')
981
no_eol = True
982
# Copy the existing list, or create a new one
983
if lines is None:
984
lines = osutils.split_lines(line_bytes)
985
else:
986
lines = lines[:]
987
# Replace the last line with one that ends in a final newline
988
lines[-1] = lines[-1] + '\n'
989
if lines is None:
990
lines = osutils.split_lines(line_bytes)
991
992
for element in key[:-1]:
993
if type(element) is not str:
994
raise TypeError("key contains non-strings: %r" % (key,))
995
if key[-1] is None:
996
key = key[:-1] + ('sha1:' + digest,)
997
elif type(key[-1]) is not str:
998
raise TypeError("key contains non-strings: %r" % (key,))
999
# Knit hunks are still last-element only
1000
version_id = key[-1]
1001
content = self._factory.make(lines, version_id)
1002
if no_eol:
1003
# Hint to the content object that its text() call should strip the
1004
# EOL.
1005
content._should_strip_eol = True
1006
if delta or (self._factory.annotated and len(present_parents) > 0):
1007
# Merge annotations from parent texts if needed.
1008
delta_hunks = self._merge_annotations(content, present_parents,
1009
parent_texts, delta, self._factory.annotated,
1010
left_matching_blocks)
1011
1012
if delta:
1013
options.append('line-delta')
1014
store_lines = self._factory.lower_line_delta(delta_hunks)
1015
size, bytes = self._record_to_data(key, digest,
1016
store_lines)
1017
else:
1018
options.append('fulltext')
1019
# isinstance is slower and we have no hierarchy.
1020
if self._factory.__class__ is KnitPlainFactory:
1021
# Use the already joined bytes saving iteration time in
1022
# _record_to_data.
1023
dense_lines = [line_bytes]
1024
if no_eol:
1025
dense_lines.append('\n')
1026
size, bytes = self._record_to_data(key, digest,
1027
lines, dense_lines)
1028
else:
1029
# get mixed annotation + content and feed it into the
1030
# serialiser.
1031
store_lines = self._factory.lower_fulltext(content)
1032
size, bytes = self._record_to_data(key, digest,
1033
store_lines)
1034
1035
access_memo = self._access.add_raw_records([(key, size)], bytes)[0]
1036
self._index.add_records(
1037
((key, options, access_memo, parents),),
1038
random_id=random_id)
1039
return digest, text_length, content
1040
1041
def annotate(self, key):
1042
"""See VersionedFiles.annotate."""
1043
return self._factory.annotate(self, key)
1044
1045
def get_annotator(self):
1046
return _KnitAnnotator(self)
1047
1048
def check(self, progress_bar=None, keys=None):
1049
"""See VersionedFiles.check()."""
1050
if keys is None:
1051
return self._logical_check()
1052
else:
1053
# At the moment, check does not extra work over get_record_stream
1054
return self.get_record_stream(keys, 'unordered', True)
1055
1056
def _logical_check(self):
1057
# This doesn't actually test extraction of everything, but that will
1058
# impact 'bzr check' substantially, and needs to be integrated with
1059
# care. However, it does check for the obvious problem of a delta with
1060
# no basis.
1061
keys = self._index.keys()
1062
parent_map = self.get_parent_map(keys)
1063
for key in keys:
1064
if self._index.get_method(key) != 'fulltext':
1065
compression_parent = parent_map[key][0]
1066
if compression_parent not in parent_map:
1067
raise errors.KnitCorrupt(self,
1068
"Missing basis parent %s for %s" % (
1069
compression_parent, key))
1070
for fallback_vfs in self._fallback_vfs:
1071
fallback_vfs.check()
1072
1073
def _check_add(self, key, lines, random_id, check_content):
1074
"""check that version_id and lines are safe to add."""
1075
version_id = key[-1]
1076
if version_id is not None:
1077
if contains_whitespace(version_id):
1078
raise InvalidRevisionId(version_id, self)
1079
self.check_not_reserved_id(version_id)
1080
# TODO: If random_id==False and the key is already present, we should
1081
# probably check that the existing content is identical to what is
1082
# being inserted, and otherwise raise an exception. This would make
1083
# the bundle code simpler.
1084
if check_content:
1085
self._check_lines_not_unicode(lines)
1086
self._check_lines_are_lines(lines)
1087
1088
def _check_header(self, key, line):
1089
rec = self._split_header(line)
1090
self._check_header_version(rec, key[-1])
1091
return rec
1092
1093
def _check_header_version(self, rec, version_id):
1094
"""Checks the header version on original format knit records.
1095
1096
These have the last component of the key embedded in the record.
1097
"""
1098
if rec[1] != version_id:
1099
raise KnitCorrupt(self,
1100
'unexpected version, wanted %r, got %r' % (version_id, rec[1]))
1101
1102
def _check_should_delta(self, parent):
1103
"""Iterate back through the parent listing, looking for a fulltext.
1104
1105
This is used when we want to decide whether to add a delta or a new
1106
fulltext. It searches for _max_delta_chain parents. When it finds a
1107
fulltext parent, it sees if the total size of the deltas leading up to
1108
it is large enough to indicate that we want a new full text anyway.
1109
1110
Return True if we should create a new delta, False if we should use a
1111
full text.
1112
"""
1113
delta_size = 0
1114
fulltext_size = None
1115
for count in xrange(self._max_delta_chain):
1116
try:
1117
# Note that this only looks in the index of this particular
1118
# KnitVersionedFiles, not in the fallbacks. This ensures that
1119
# we won't store a delta spanning physical repository
1120
# boundaries.
1121
build_details = self._index.get_build_details([parent])
1122
parent_details = build_details[parent]
1123
except (RevisionNotPresent, KeyError), e:
1124
# Some basis is not locally present: always fulltext
1125
return False
1126
index_memo, compression_parent, _, _ = parent_details
1127
_, _, size = index_memo
1128
if compression_parent is None:
1129
fulltext_size = size
1130
break
1131
delta_size += size
1132
# We don't explicitly check for presence because this is in an
1133
# inner loop, and if it's missing it'll fail anyhow.
1134
parent = compression_parent
1135
else:
1136
# We couldn't find a fulltext, so we must create a new one
342
count = 0
343
delta_parents = [delta_parent]
344
while count < 25:
345
parent = delta_parents[0]
346
method = self._index.get_method(parent)
347
if method == 'fulltext':
348
break
349
delta_parents = self._index.get_parents(parent)
350
count = count + 1
351
if method == 'line-delta':
352
# did not find a fulltext in the delta limit.
353
# just do a normal insertion.
354
return super(KnitVersionedFile, self)._add_delta(version_id,
355
parents,
356
delta_parent,
357
sha1,
358
noeol,
359
delta)
360
361
options.append('line-delta')
362
store_lines = self.factory.lower_line_delta(delta)
363
364
where, size = self._data.add_record(version_id, digest, store_lines)
365
self._index.add_version(version_id, options, where, size, parents)
366
367
def _add_raw_records(self, records, data):
368
"""Add all the records 'records' with data pre-joined in 'data'.
369
370
:param records: A list of tuples(version_id, options, parents, size).
371
:param data: The data for the records. When it is written, the records
372
are adjusted to have pos pointing into data by the sum of
373
the preceding records sizes.
374
"""
375
# write all the data
376
pos = self._data.add_raw_record(data)
377
index_entries = []
378
for (version_id, options, parents, size) in records:
379
index_entries.append((version_id, options, pos, size, parents))
380
pos += size
381
self._index.add_versions(index_entries)
382
383
def clear_cache(self):
384
"""Clear the data cache only."""
385
self._data.clear_cache()
386
387
def copy_to(self, name, transport):
388
"""See VersionedFile.copy_to()."""
389
# copy the current index to a temp index to avoid racing with local
390
# writes
391
transport.put(name + INDEX_SUFFIX + '.tmp', self.transport.get(self._index._filename),)
392
# copy the data file
393
transport.put(name + DATA_SUFFIX, self._data._open_file())
394
# rename the copied index into place
395
transport.rename(name + INDEX_SUFFIX + '.tmp', name + INDEX_SUFFIX)
396
397
def create_empty(self, name, transport, mode=None):
398
return KnitVersionedFile(name, transport, factory=self.factory, delta=self.delta, create=True)
399
400
def _fix_parents(self, version, new_parents):
401
"""Fix the parents list for version.
402
403
This is done by appending a new version to the index
404
with identical data except for the parents list.
405
the parents list must be a superset of the current
406
list.
407
"""
408
current_values = self._index._cache[version]
409
assert set(current_values[4]).difference(set(new_parents)) == set()
410
self._index.add_version(version,
411
current_values[1],
412
current_values[2],
413
current_values[3],
414
new_parents)
415
416
def get_delta(self, version_id):
417
"""Get a delta for constructing version from some other version."""
418
if not self.has_version(version_id):
419
raise RevisionNotPresent(version_id, self.filename)
420
421
parents = self.get_parents(version_id)
422
if len(parents):
423
parent = parents[0]
424
else:
425
parent = None
426
data_pos, data_size = self._index.get_position(version_id)
427
data, sha1 = self._data.read_records(((version_id, data_pos, data_size),))[version_id]
428
version_idx = self._index.lookup(version_id)
429
noeol = 'no-eol' in self._index.get_options(version_id)
430
if 'fulltext' == self._index.get_method(version_id):
431
new_content = self.factory.parse_fulltext(data, version_idx)
432
if parent is not None:
433
reference_content = self._get_content(parent)
434
old_texts = reference_content.text()
435
else:
436
old_texts = []
437
new_texts = new_content.text()
438
delta_seq = KnitSequenceMatcher(None, old_texts, new_texts)
439
return parent, sha1, noeol, self._make_line_delta(delta_seq, new_content)
440
else:
441
delta = self.factory.parse_line_delta(data, version_idx)
442
return parent, sha1, noeol, delta
443
444
def get_graph_with_ghosts(self):
445
"""See VersionedFile.get_graph_with_ghosts()."""
446
graph_items = self._index.get_graph()
447
return dict(graph_items)
448
449
def get_sha1(self, version_id):
450
"""See VersionedFile.get_sha1()."""
451
record_map = self._get_record_map([version_id])
452
method, content, digest, next = record_map[version_id]
453
return digest
454
455
@staticmethod
456
def get_suffixes():
457
"""See VersionedFile.get_suffixes()."""
458
return [DATA_SUFFIX, INDEX_SUFFIX]
459
460
def has_ghost(self, version_id):
461
"""True if there is a ghost reference in the file to version_id."""
462
# maybe we have it
463
if self.has_version(version_id):
1137
464
return False
1138
# Simple heuristic - if the total I/O wold be greater as a delta than
1139
# the originally installed fulltext, we create a new fulltext.
1140
return fulltext_size > delta_size
1141
1142
def _build_details_to_components(self, build_details):
1143
"""Convert a build_details tuple to a position tuple."""
1144
# record_details, access_memo, compression_parent
1145
return build_details[3], build_details[0], build_details[1]
1146
1147
def _get_components_positions(self, keys, allow_missing=False):
1148
"""Produce a map of position data for the components of keys.
465
# optimisable if needed by memoising the _ghosts set.
466
items = self._index.get_graph()
467
for node, parents in items:
468
for parent in parents:
469
if parent not in self._index._cache:
470
if parent == version_id:
471
return True
472
return False
473
474
def versions(self):
475
"""See VersionedFile.versions."""
476
return self._index.get_versions()
477
478
def has_version(self, version_id):
479
"""See VersionedFile.has_version."""
480
return self._index.has_version(version_id)
481
482
__contains__ = has_version
483
484
def _merge_annotations(self, content, parents, parent_texts={},
485
delta=None, annotated=None):
486
"""Merge annotations for content. This is done by comparing
487
the annotations based on changed to the text.
488
"""
489
if annotated:
490
delta_seq = None
491
for parent_id in parents:
492
merge_content = self._get_content(parent_id, parent_texts)
493
seq = KnitSequenceMatcher(None, merge_content.text(), content.text())
494
if delta_seq is None:
495
# setup a delta seq to reuse.
496
delta_seq = seq
497
for i, j, n in seq.get_matching_blocks():
498
if n == 0:
499
continue
500
# this appears to copy (origin, text) pairs across to the new
501
# content for any line that matches the last-checked parent.
502
# FIXME: save the sequence control data for delta compression
503
# against the most relevant parent rather than rediffing.
504
content._lines[j:j+n] = merge_content._lines[i:i+n]
505
if delta:
506
if not annotated:
507
reference_content = self._get_content(parents[0], parent_texts)
508
new_texts = content.text()
509
old_texts = reference_content.text()
510
delta_seq = KnitSequenceMatcher(None, old_texts, new_texts)
511
return self._make_line_delta(delta_seq, content)
512
513
def _make_line_delta(self, delta_seq, new_content):
514
"""Generate a line delta from delta_seq and new_content."""
515
diff_hunks = []
516
for op in delta_seq.get_opcodes():
517
if op[0] == 'equal':
518
continue
519
diff_hunks.append((op[1], op[2], op[4]-op[3], new_content._lines[op[3]:op[4]]))
520
return diff_hunks
521
522
def _get_components_positions(self, version_ids):
523
"""Produce a map of position data for the components of versions.
1149
524
1150
525
This data is intended to be used for retrieving the knit records.
1151
526
1152
A dict of key to (record_details, index_memo, next, parents) is
527
A dict of version_id to (method, data_pos, data_size, next) is
1153
528
returned.
1154
529
method is the way referenced data should be applied.
1155
index_memo is the handle to pass to the data access to actually get the
1156
data
530
data_pos is the position of the data in the knit.
531
data_size is the size of the data in the knit.
1157
532
next is the build-parent of the version, or None for fulltexts.
1158
parents is the version_ids of the parents of this version
1159
1160
:param allow_missing: If True do not raise an error on a missing component,
1161
just ignore it.
1162
533
"""
1163
534
component_data = {}
1164
pending_components = keys
1165
while pending_components:
1166
build_details = self._index.get_build_details(pending_components)
1167
current_components = set(pending_components)
1168
pending_components = set()
1169
for key, details in build_details.iteritems():
1170
(index_memo, compression_parent, parents,
1171
record_details) = details
1172
method = record_details[0]
1173
if compression_parent is not None:
1174
pending_components.add(compression_parent)
1175
component_data[key] = self._build_details_to_components(details)
1176
missing = current_components.difference(build_details)
1177
if missing and not allow_missing:
1178
raise errors.RevisionNotPresent(missing.pop(), self)
535
for version_id in version_ids:
536
cursor = version_id
537
538
while cursor is not None and cursor not in component_data:
539
method = self._index.get_method(cursor)
540
if method == 'fulltext':
541
next = None
542
else:
543
next = self.get_parents(cursor)[0]
544
data_pos, data_size = self._index.get_position(cursor)
545
component_data[cursor] = (method, data_pos, data_size, next)
546
cursor = next
1179
547
return component_data
1180
1181
def _get_content(self, key, parent_texts={}):
548
549
def _get_content(self, version_id, parent_texts={}):
1182
550
"""Returns a content object that makes up the specified
1183
551
version."""
1184
cached_version = parent_texts.get(key, None)
552
if not self.has_version(version_id):
553
raise RevisionNotPresent(version_id, self.filename)
554
555
cached_version = parent_texts.get(version_id, None)
1185
556
if cached_version is not None:
1186
# Ensure the cache dict is valid.
1187
if not self.get_parent_map([key]):
1188
raise RevisionNotPresent(key, self)
1189
557
return cached_version
1190
generator = _VFContentMapGenerator(self, [key])
1191
return generator._get_content(key)
1192
1193
def get_known_graph_ancestry(self, keys):
1194
"""Get a KnownGraph instance with the ancestry of keys."""
1195
parent_map, missing_keys = self._index.find_ancestry(keys)
1196
for fallback in self._fallback_vfs:
1197
if not missing_keys:
1198
break
1199
(f_parent_map, f_missing_keys) = fallback._index.find_ancestry(
1200
missing_keys)
1201
parent_map.update(f_parent_map)
1202
missing_keys = f_missing_keys
1203
kg = _mod_graph.KnownGraph(parent_map)
1204
return kg
1205
1206
def get_parent_map(self, keys):
1207
"""Get a map of the graph parents of keys.
1208
1209
:param keys: The keys to look up parents for.
1210
:return: A mapping from keys to parents. Absent keys are absent from
1211
the mapping.
1212
"""
1213
return self._get_parent_map_with_sources(keys)[0]
1214
1215
def _get_parent_map_with_sources(self, keys):
1216
"""Get a map of the parents of keys.
1217
1218
:param keys: The keys to look up parents for.
1219
:return: A tuple. The first element is a mapping from keys to parents.
1220
Absent keys are absent from the mapping. The second element is a
1221
list with the locations each key was found in. The first element
1222
is the in-this-knit parents, the second the first fallback source,
1223
and so on.
1224
"""
1225
result = {}
1226
sources = [self._index] + self._fallback_vfs
1227
source_results = []
1228
missing = set(keys)
1229
for source in sources:
1230
if not missing:
1231
break
1232
new_result = source.get_parent_map(missing)
1233
source_results.append(new_result)
1234
result.update(new_result)
1235
missing.difference_update(set(new_result))
1236
return result, source_results
1237
1238
def _get_record_map(self, keys, allow_missing=False):
558
559
text_map, contents_map = self._get_content_maps([version_id])
560
return contents_map[version_id]
561
562
def _check_versions_present(self, version_ids):
563
"""Check that all specified versions are present."""
564
version_ids = set(version_ids)
565
for r in list(version_ids):
566
if self._index.has_version(r):
567
version_ids.remove(r)
568
if version_ids:
569
raise RevisionNotPresent(list(version_ids)[0], self.filename)
570
571
def _add_lines_with_ghosts(self, version_id, parents, lines, parent_texts):
572
"""See VersionedFile.add_lines_with_ghosts()."""
573
self._check_add(version_id, lines)
574
return self._add(version_id, lines[:], parents, self.delta, parent_texts)
575
576
def _add_lines(self, version_id, parents, lines, parent_texts):
577
"""See VersionedFile.add_lines."""
578
self._check_add(version_id, lines)
579
self._check_versions_present(parents)
580
return self._add(version_id, lines[:], parents, self.delta, parent_texts)
581
582
def _check_add(self, version_id, lines):
583
"""check that version_id and lines are safe to add."""
584
assert self.writable, "knit is not opened for write"
585
### FIXME escape. RBC 20060228
586
if contains_whitespace(version_id):
587
raise InvalidRevisionId(version_id, self.filename)
588
if self.has_version(version_id):
589
raise RevisionAlreadyPresent(version_id, self.filename)
590
self._check_lines_not_unicode(lines)
591
self._check_lines_are_lines(lines)
592
593
def _add(self, version_id, lines, parents, delta, parent_texts):
594
"""Add a set of lines on top of version specified by parents.
595
596
If delta is true, compress the text as a line-delta against
597
the first parent.
598
599
Any versions not present will be converted into ghosts.
600
"""
601
# 461 0 6546.0390 43.9100 bzrlib.knit:489(_add)
602
# +400 0 889.4890 418.9790 +bzrlib.knit:192(lower_fulltext)
603
# +461 0 1364.8070 108.8030 +bzrlib.knit:996(add_record)
604
# +461 0 193.3940 41.5720 +bzrlib.knit:898(add_version)
605
# +461 0 134.0590 18.3810 +bzrlib.osutils:361(sha_strings)
606
# +461 0 36.3420 15.4540 +bzrlib.knit:146(make)
607
# +1383 0 8.0370 8.0370 +<len>
608
# +61 0 13.5770 7.9190 +bzrlib.knit:199(lower_line_delta)
609
# +61 0 963.3470 7.8740 +bzrlib.knit:427(_get_content)
610
# +61 0 973.9950 5.2950 +bzrlib.knit:136(line_delta)
611
# +61 0 1918.1800 5.2640 +bzrlib.knit:359(_merge_annotations)
612
613
present_parents = []
614
ghosts = []
615
if parent_texts is None:
616
parent_texts = {}
617
for parent in parents:
618
if not self.has_version(parent):
619
ghosts.append(parent)
620
else:
621
present_parents.append(parent)
622
623
if delta and not len(present_parents):
624
delta = False
625
626
digest = sha_strings(lines)
627
options = []
628
if lines:
629
if lines[-1][-1] != '\n':
630
options.append('no-eol')
631
lines[-1] = lines[-1] + '\n'
632
633
if len(present_parents) and delta:
634
# To speed the extract of texts the delta chain is limited
635
# to a fixed number of deltas. This should minimize both
636
# I/O and the time spend applying deltas.
637
count = 0
638
delta_parents = present_parents
639
while count < 25:
640
parent = delta_parents[0]
641
method = self._index.get_method(parent)
642
if method == 'fulltext':
643
break
644
delta_parents = self._index.get_parents(parent)
645
count = count + 1
646
if method == 'line-delta':
647
delta = False
648
649
lines = self.factory.make(lines, version_id)
650
if delta or (self.factory.annotated and len(present_parents) > 0):
651
# Merge annotations from parent texts if so is needed.
652
delta_hunks = self._merge_annotations(lines, present_parents, parent_texts,
653
delta, self.factory.annotated)
654
655
if delta:
656
options.append('line-delta')
657
store_lines = self.factory.lower_line_delta(delta_hunks)
658
else:
659
options.append('fulltext')
660
store_lines = self.factory.lower_fulltext(lines)
661
662
where, size = self._data.add_record(version_id, digest, store_lines)
663
self._index.add_version(version_id, options, where, size, parents)
664
return lines
665
666
def check(self, progress_bar=None):
667
"""See VersionedFile.check()."""
668
669
def _clone_text(self, new_version_id, old_version_id, parents):
670
"""See VersionedFile.clone_text()."""
671
# FIXME RBC 20060228 make fast by only inserting an index with null
672
# delta.
673
self.add_lines(new_version_id, parents, self.get_lines(old_version_id))
674
675
def get_lines(self, version_id):
676
"""See VersionedFile.get_lines()."""
677
return self.get_line_list([version_id])[0]
678
679
def _get_record_map(self, version_ids):
1239
680
"""Produce a dictionary of knit records.
1240
1241
:return: {key:(record, record_details, digest, next)}
1242
record
1243
data returned from read_records (a KnitContentobject)
1244
record_details
1245
opaque information to pass to parse_record
1246
digest
1247
SHA1 digest of the full text after all steps are done
1248
next
1249
build-parent of the version, i.e. the leftmost ancestor.
1250
Will be None if the record is not a delta.
1251
:param keys: The keys to build a map for
1252
:param allow_missing: If some records are missing, rather than
1253
error, just return the data that could be generated.
1254
"""
1255
raw_map = self._get_record_map_unparsed(keys,
1256
allow_missing=allow_missing)
1257
return self._raw_map_to_record_map(raw_map)
1258
1259
def _raw_map_to_record_map(self, raw_map):
1260
"""Parse the contents of _get_record_map_unparsed.
1261
1262
:return: see _get_record_map.
1263
"""
1264
result = {}
1265
for key in raw_map:
1266
data, record_details, next = raw_map[key]
1267
content, digest = self._parse_record(key[-1], data)
1268
result[key] = content, record_details, digest, next
1269
return result
1270
1271
def _get_record_map_unparsed(self, keys, allow_missing=False):
1272
"""Get the raw data for reconstructing keys without parsing it.
1273
1274
:return: A dict suitable for parsing via _raw_map_to_record_map.
1275
key-> raw_bytes, (method, noeol), compression_parent
1276
"""
1277
# This retries the whole request if anything fails. Potentially we
1278
# could be a bit more selective. We could track the keys whose records
1279
# we have successfully found, and then only request the new records
1280
# from there. However, _get_components_positions grabs the whole build
1281
# chain, which means we'll likely try to grab the same records again
1282
# anyway. Also, can the build chains change as part of a pack
1283
# operation? We wouldn't want to end up with a broken chain.
1284
while True:
1285
try:
1286
position_map = self._get_components_positions(keys,
1287
allow_missing=allow_missing)
1288
# key = component_id, r = record_details, i_m = index_memo,
1289
# n = next
1290
records = [(key, i_m) for key, (r, i_m, n)
1291
in position_map.iteritems()]
1292
# Sort by the index memo, so that we request records from the
1293
# same pack file together, and in forward-sorted order
1294
records.sort(key=operator.itemgetter(1))
1295
raw_record_map = {}
1296
for key, data in self._read_records_iter_unchecked(records):
1297
(record_details, index_memo, next) = position_map[key]
1298
raw_record_map[key] = data, record_details, next
1299
return raw_record_map
1300
except errors.RetryWithNewPacks, e:
1301
self._access.reload_or_raise(e)
1302
1303
@classmethod
1304
def _split_by_prefix(cls, keys):
1305
"""For the given keys, split them up based on their prefix.
1306
1307
To keep memory pressure somewhat under control, split the
1308
requests back into per-file-id requests, otherwise "bzr co"
1309
extracts the full tree into memory before writing it to disk.
1310
This should be revisited if _get_content_maps() can ever cross
1311
file-id boundaries.
1312
1313
The keys for a given file_id are kept in the same relative order.
1314
Ordering between file_ids is not, though prefix_order will return the
1315
order that the key was first seen.
1316
1317
:param keys: An iterable of key tuples
1318
:return: (split_map, prefix_order)
1319
split_map A dictionary mapping prefix => keys
1320
prefix_order The order that we saw the various prefixes
1321
"""
1322
split_by_prefix = {}
1323
prefix_order = []
1324
for key in keys:
1325
if len(key) == 1:
1326
prefix = ''
1327
else:
1328
prefix = key[0]
1329
1330
if prefix in split_by_prefix:
1331
split_by_prefix[prefix].append(key)
1332
else:
1333
split_by_prefix[prefix] = [key]
1334
prefix_order.append(prefix)
1335
return split_by_prefix, prefix_order
1336
1337
def _group_keys_for_io(self, keys, non_local_keys, positions,
1338
_min_buffer_size=_STREAM_MIN_BUFFER_SIZE):
1339
"""For the given keys, group them into 'best-sized' requests.
1340
1341
The idea is to avoid making 1 request per file, but to never try to
1342
unpack an entire 1.5GB source tree in a single pass. Also when
1343
possible, we should try to group requests to the same pack file
1344
together.
1345
1346
:return: list of (keys, non_local) tuples that indicate what keys
1347
should be fetched next.
1348
"""
1349
# TODO: Ideally we would group on 2 factors. We want to extract texts
1350
# from the same pack file together, and we want to extract all
1351
# the texts for a given build-chain together. Ultimately it
1352
# probably needs a better global view.
1353
total_keys = len(keys)
1354
prefix_split_keys, prefix_order = self._split_by_prefix(keys)
1355
prefix_split_non_local_keys, _ = self._split_by_prefix(non_local_keys)
1356
cur_keys = []
1357
cur_non_local = set()
1358
cur_size = 0
1359
result = []
1360
sizes = []
1361
for prefix in prefix_order:
1362
keys = prefix_split_keys[prefix]
1363
non_local = prefix_split_non_local_keys.get(prefix, [])
1364
1365
this_size = self._index._get_total_build_size(keys, positions)
1366
cur_size += this_size
1367
cur_keys.extend(keys)
1368
cur_non_local.update(non_local)
1369
if cur_size > _min_buffer_size:
1370
result.append((cur_keys, cur_non_local))
1371
sizes.append(cur_size)
1372
cur_keys = []
1373
cur_non_local = set()
1374
cur_size = 0
1375
if cur_keys:
1376
result.append((cur_keys, cur_non_local))
1377
sizes.append(cur_size)
1378
return result
1379
1380
def get_record_stream(self, keys, ordering, include_delta_closure):
1381
"""Get a stream of records for keys.
1382
1383
:param keys: The keys to include.
1384
:param ordering: Either 'unordered' or 'topological'. A topologically
1385
sorted stream has compression parents strictly before their
1386
children.
1387
:param include_delta_closure: If True then the closure across any
1388
compression parents will be included (in the opaque data).
1389
:return: An iterator of ContentFactory objects, each of which is only
1390
valid until the iterator is advanced.
1391
"""
1392
# keys might be a generator
1393
keys = set(keys)
1394
if not keys:
1395
return
1396
if not self._index.has_graph:
1397
# Cannot sort when no graph has been stored.
1398
ordering = 'unordered'
1399
1400
remaining_keys = keys
1401
while True:
1402
try:
1403
keys = set(remaining_keys)
1404
for content_factory in self._get_remaining_record_stream(keys,
1405
ordering, include_delta_closure):
1406
remaining_keys.discard(content_factory.key)
1407
yield content_factory
1408
return
1409
except errors.RetryWithNewPacks, e:
1410
self._access.reload_or_raise(e)
1411
1412
def _get_remaining_record_stream(self, keys, ordering,
1413
include_delta_closure):
1414
"""This function is the 'retry' portion for get_record_stream."""
1415
if include_delta_closure:
1416
positions = self._get_components_positions(keys, allow_missing=True)
1417
else:
1418
build_details = self._index.get_build_details(keys)
1419
# map from key to
1420
# (record_details, access_memo, compression_parent_key)
1421
positions = dict((key, self._build_details_to_components(details))
1422
for key, details in build_details.iteritems())
1423
absent_keys = keys.difference(set(positions))
1424
# There may be more absent keys : if we're missing the basis component
1425
# and are trying to include the delta closure.
1426
# XXX: We should not ever need to examine remote sources because we do
1427
# not permit deltas across versioned files boundaries.
1428
if include_delta_closure:
1429
needed_from_fallback = set()
1430
# Build up reconstructable_keys dict. key:True in this dict means
1431
# the key can be reconstructed.
1432
reconstructable_keys = {}
1433
for key in keys:
1434
# the delta chain
1435
try:
1436
chain = [key, positions[key][2]]
1437
except KeyError:
1438
needed_from_fallback.add(key)
1439
continue
1440
result = True
1441
while chain[-1] is not None:
1442
if chain[-1] in reconstructable_keys:
1443
result = reconstructable_keys[chain[-1]]
1444
break
1445
else:
1446
try:
1447
chain.append(positions[chain[-1]][2])
1448
except KeyError:
1449
# missing basis component
1450
needed_from_fallback.add(chain[-1])
1451
result = True
1452
break
1453
for chain_key in chain[:-1]:
1454
reconstructable_keys[chain_key] = result
1455
if not result:
1456
needed_from_fallback.add(key)
1457
# Double index lookups here : need a unified api ?
1458
global_map, parent_maps = self._get_parent_map_with_sources(keys)
1459
if ordering in ('topological', 'groupcompress'):
1460
if ordering == 'topological':
1461
# Global topological sort
1462
present_keys = tsort.topo_sort(global_map)
1463
else:
1464
present_keys = sort_groupcompress(global_map)
1465
# Now group by source:
1466
source_keys = []
1467
current_source = None
1468
for key in present_keys:
1469
for parent_map in parent_maps:
1470
if key in parent_map:
1471
key_source = parent_map
1472
break
1473
if current_source is not key_source:
1474
source_keys.append((key_source, []))
1475
current_source = key_source
1476
source_keys[-1][1].append(key)
1477
else:
1478
if ordering != 'unordered':
1479
raise AssertionError('valid values for ordering are:'
1480
' "unordered", "groupcompress" or "topological" not: %r'
1481
% (ordering,))
1482
# Just group by source; remote sources first.
1483
present_keys = []
1484
source_keys = []
1485
for parent_map in reversed(parent_maps):
1486
source_keys.append((parent_map, []))
1487
for key in parent_map:
1488
present_keys.append(key)
1489
source_keys[-1][1].append(key)
1490
# We have been requested to return these records in an order that
1491
# suits us. So we ask the index to give us an optimally sorted
1492
# order.
1493
for source, sub_keys in source_keys:
1494
if source is parent_maps[0]:
1495
# Only sort the keys for this VF
1496
self._index._sort_keys_by_io(sub_keys, positions)
1497
absent_keys = keys - set(global_map)
1498
for key in absent_keys:
1499
yield AbsentContentFactory(key)
1500
# restrict our view to the keys we can answer.
1501
# XXX: Memory: TODO: batch data here to cap buffered data at (say) 1MB.
1502
# XXX: At that point we need to consider the impact of double reads by
1503
# utilising components multiple times.
1504
if include_delta_closure:
1505
# XXX: get_content_maps performs its own index queries; allow state
1506
# to be passed in.
1507
non_local_keys = needed_from_fallback - absent_keys
1508
for keys, non_local_keys in self._group_keys_for_io(present_keys,
1509
non_local_keys,
1510
positions):
1511
generator = _VFContentMapGenerator(self, keys, non_local_keys,
1512
global_map,
1513
ordering=ordering)
1514
for record in generator.get_record_stream():
1515
yield record
1516
else:
1517
for source, keys in source_keys:
1518
if source is parent_maps[0]:
1519
# this KnitVersionedFiles
1520
records = [(key, positions[key][1]) for key in keys]
1521
for key, raw_data, sha1 in self._read_records_iter_raw(records):
1522
(record_details, index_memo, _) = positions[key]
1523
yield KnitContentFactory(key, global_map[key],
1524
record_details, sha1, raw_data, self._factory.annotated, None)
1525
else:
1526
vf = self._fallback_vfs[parent_maps.index(source) - 1]
1527
for record in vf.get_record_stream(keys, ordering,
1528
include_delta_closure):
1529
yield record
1530
1531
def get_sha1s(self, keys):
1532
"""See VersionedFiles.get_sha1s()."""
1533
missing = set(keys)
1534
record_map = self._get_record_map(missing, allow_missing=True)
1535
result = {}
1536
for key, details in record_map.iteritems():
1537
if key not in missing:
1538
continue
1539
# record entry 2 is the 'digest'.
1540
result[key] = details[2]
1541
missing.difference_update(set(result))
1542
for source in self._fallback_vfs:
1543
if not missing:
1544
break
1545
new_result = source.get_sha1s(missing)
1546
result.update(new_result)
1547
missing.difference_update(set(new_result))
1548
return result
1549
1550
def insert_record_stream(self, stream):
1551
"""Insert a record stream into this container.
1552
1553
:param stream: A stream of records to insert.
1554
:return: None
1555
:seealso VersionedFiles.get_record_stream:
1556
"""
1557
def get_adapter(adapter_key):
1558
try:
1559
return adapters[adapter_key]
1560
except KeyError:
1561
adapter_factory = adapter_registry.get(adapter_key)
1562
adapter = adapter_factory(self)
1563
adapters[adapter_key] = adapter
1564
return adapter
1565
delta_types = set()
1566
if self._factory.annotated:
1567
# self is annotated, we need annotated knits to use directly.
1568
annotated = "annotated-"
1569
convertibles = []
1570
else:
1571
# self is not annotated, but we can strip annotations cheaply.
1572
annotated = ""
1573
convertibles = set(["knit-annotated-ft-gz"])
1574
if self._max_delta_chain:
1575
delta_types.add("knit-annotated-delta-gz")
1576
convertibles.add("knit-annotated-delta-gz")
1577
# The set of types we can cheaply adapt without needing basis texts.
1578
native_types = set()
1579
if self._max_delta_chain:
1580
native_types.add("knit-%sdelta-gz" % annotated)
1581
delta_types.add("knit-%sdelta-gz" % annotated)
1582
native_types.add("knit-%sft-gz" % annotated)
1583
knit_types = native_types.union(convertibles)
1584
adapters = {}
1585
# Buffer all index entries that we can't add immediately because their
1586
# basis parent is missing. We don't buffer all because generating
1587
# annotations may require access to some of the new records. However we
1588
# can't generate annotations from new deltas until their basis parent
1589
# is present anyway, so we get away with not needing an index that
1590
# includes the new keys.
1591
#
1592
# See <http://launchpad.net/bugs/300177> about ordering of compression
1593
# parents in the records - to be conservative, we insist that all
1594
# parents must be present to avoid expanding to a fulltext.
1595
#
1596
# key = basis_parent, value = index entry to add
1597
buffered_index_entries = {}
1598
for record in stream:
1599
buffered = False
1600
parents = record.parents
1601
if record.storage_kind in delta_types:
1602
# TODO: eventually the record itself should track
1603
# compression_parent
1604
compression_parent = parents[0]
1605
else:
1606
compression_parent = None
1607
# Raise an error when a record is missing.
1608
if record.storage_kind == 'absent':
1609
raise RevisionNotPresent([record.key], self)
1610
elif ((record.storage_kind in knit_types)
1611
and (compression_parent is None
1612
or not self._fallback_vfs
1613
or self._index.has_key(compression_parent)
1614
or not self.has_key(compression_parent))):
1615
# we can insert the knit record literally if either it has no
1616
# compression parent OR we already have its basis in this kvf
1617
# OR the basis is not present even in the fallbacks. In the
1618
# last case it will either turn up later in the stream and all
1619
# will be well, or it won't turn up at all and we'll raise an
1620
# error at the end.
1621
#
1622
# TODO: self.has_key is somewhat redundant with
1623
# self._index.has_key; we really want something that directly
1624
# asks if it's only present in the fallbacks. -- mbp 20081119
1625
if record.storage_kind not in native_types:
1626
try:
1627
adapter_key = (record.storage_kind, "knit-delta-gz")
1628
adapter = get_adapter(adapter_key)
1629
except KeyError:
1630
adapter_key = (record.storage_kind, "knit-ft-gz")
1631
adapter = get_adapter(adapter_key)
1632
bytes = adapter.get_bytes(record)
1633
else:
1634
# It's a knit record, it has a _raw_record field (even if
1635
# it was reconstituted from a network stream).
1636
bytes = record._raw_record
1637
options = [record._build_details[0]]
1638
if record._build_details[1]:
1639
options.append('no-eol')
1640
# Just blat it across.
1641
# Note: This does end up adding data on duplicate keys. As
1642
# modern repositories use atomic insertions this should not
1643
# lead to excessive growth in the event of interrupted fetches.
1644
# 'knit' repositories may suffer excessive growth, but as a
1645
# deprecated format this is tolerable. It can be fixed if
1646
# needed by in the kndx index support raising on a duplicate
1647
# add with identical parents and options.
1648
access_memo = self._access.add_raw_records(
1649
[(record.key, len(bytes))], bytes)[0]
1650
index_entry = (record.key, options, access_memo, parents)
1651
if 'fulltext' not in options:
1652
# Not a fulltext, so we need to make sure the compression
1653
# parent will also be present.
1654
# Note that pack backed knits don't need to buffer here
1655
# because they buffer all writes to the transaction level,
1656
# but we don't expose that difference at the index level. If
1657
# the query here has sufficient cost to show up in
1658
# profiling we should do that.
1659
#
1660
# They're required to be physically in this
1661
# KnitVersionedFiles, not in a fallback.
1662
if not self._index.has_key(compression_parent):
1663
pending = buffered_index_entries.setdefault(
1664
compression_parent, [])
1665
pending.append(index_entry)
1666
buffered = True
1667
if not buffered:
1668
self._index.add_records([index_entry])
1669
elif record.storage_kind == 'chunked':
1670
self.add_lines(record.key, parents,
1671
osutils.chunks_to_lines(record.get_bytes_as('chunked')))
1672
else:
1673
# Not suitable for direct insertion as a
1674
# delta, either because it's not the right format, or this
1675
# KnitVersionedFiles doesn't permit deltas (_max_delta_chain ==
1676
# 0) or because it depends on a base only present in the
1677
# fallback kvfs.
1678
self._access.flush()
1679
try:
1680
# Try getting a fulltext directly from the record.
1681
bytes = record.get_bytes_as('fulltext')
1682
except errors.UnavailableRepresentation:
1683
adapter_key = record.storage_kind, 'fulltext'
1684
adapter = get_adapter(adapter_key)
1685
bytes = adapter.get_bytes(record)
1686
lines = split_lines(bytes)
1687
try:
1688
self.add_lines(record.key, parents, lines)
1689
except errors.RevisionAlreadyPresent:
1690
pass
1691
# Add any records whose basis parent is now available.
1692
if not buffered:
1693
added_keys = [record.key]
1694
while added_keys:
1695
key = added_keys.pop(0)
1696
if key in buffered_index_entries:
1697
index_entries = buffered_index_entries[key]
1698
self._index.add_records(index_entries)
1699
added_keys.extend(
1700
[index_entry[0] for index_entry in index_entries])
1701
del buffered_index_entries[key]
1702
if buffered_index_entries:
1703
# There were index entries buffered at the end of the stream,
1704
# So these need to be added (if the index supports holding such
1705
# entries for later insertion)
1706
for key in buffered_index_entries:
1707
index_entries = buffered_index_entries[key]
1708
self._index.add_records(index_entries,
1709
missing_compression_parents=True)
1710
1711
def get_missing_compression_parent_keys(self):
1712
"""Return an iterable of keys of missing compression parents.
1713
1714
Check this after calling insert_record_stream to find out if there are
1715
any missing compression parents. If there are, the records that
1716
depend on them are not able to be inserted safely. For atomic
1717
KnitVersionedFiles built on packs, the transaction should be aborted or
1718
suspended - commit will fail at this point. Nonatomic knits will error
1719
earlier because they have no staging area to put pending entries into.
1720
"""
1721
return self._index.get_missing_compression_parents()
1722
1723
def iter_lines_added_or_present_in_keys(self, keys, pb=None):
1724
"""Iterate over the lines in the versioned files from keys.
1725
1726
This may return lines from other keys. Each item the returned
1727
iterator yields is a tuple of a line and a text version that that line
1728
is present in (not introduced in).
1729
1730
Ordering of results is in whatever order is most suitable for the
1731
underlying storage format.
1732
1733
If a progress bar is supplied, it may be used to indicate progress.
1734
The caller is responsible for cleaning up progress bars (because this
1735
is an iterator).
1736
1737
NOTES:
1738
* Lines are normalised by the underlying store: they will all have \\n
1739
terminators.
1740
* Lines are returned in arbitrary order.
1741
* If a requested key did not change any lines (or didn't have any
1742
lines), it may not be mentioned at all in the result.
1743
1744
:param pb: Progress bar supplied by caller.
1745
:return: An iterator over (line, key).
1746
"""
1747
if pb is None:
1748
pb = progress.DummyProgress()
1749
keys = set(keys)
1750
total = len(keys)
1751
done = False
1752
while not done:
1753
try:
1754
# we don't care about inclusions, the caller cares.
1755
# but we need to setup a list of records to visit.
1756
# we need key, position, length
1757
key_records = []
1758
build_details = self._index.get_build_details(keys)
1759
for key, details in build_details.iteritems():
1760
if key in keys:
1761
key_records.append((key, details[0]))
1762
records_iter = enumerate(self._read_records_iter(key_records))
1763
for (key_idx, (key, data, sha_value)) in records_iter:
1764
pb.update('Walking content', key_idx, total)
1765
compression_parent = build_details[key][1]
1766
if compression_parent is None:
1767
# fulltext
1768
line_iterator = self._factory.get_fulltext_content(data)
1769
else:
1770
# Delta
1771
line_iterator = self._factory.get_linedelta_content(data)
1772
# Now that we are yielding the data for this key, remove it
1773
# from the list
1774
keys.remove(key)
1775
# XXX: It might be more efficient to yield (key,
1776
# line_iterator) in the future. However for now, this is a
1777
# simpler change to integrate into the rest of the
1778
# codebase. RBC 20071110
1779
for line in line_iterator:
1780
yield line, key
1781
done = True
1782
except errors.RetryWithNewPacks, e:
1783
self._access.reload_or_raise(e)
1784
# If there are still keys we've not yet found, we look in the fallback
1785
# vfs, and hope to find them there. Note that if the keys are found
1786
# but had no changes or no content, the fallback may not return
1787
# anything.
1788
if keys and not self._fallback_vfs:
1789
# XXX: strictly the second parameter is meant to be the file id
1790
# but it's not easily accessible here.
1791
raise RevisionNotPresent(keys, repr(self))
1792
for source in self._fallback_vfs:
1793
if not keys:
1794
break
1795
source_keys = set()
1796
for line, key in source.iter_lines_added_or_present_in_keys(keys):
1797
source_keys.add(key)
1798
yield line, key
1799
keys.difference_update(source_keys)
1800
pb.update('Walking content', total, total)
1801
1802
def _make_line_delta(self, delta_seq, new_content):
1803
"""Generate a line delta from delta_seq and new_content."""
1804
diff_hunks = []
1805
for op in delta_seq.get_opcodes():
1806
if op[0] == 'equal':
1807
continue
1808
diff_hunks.append((op[1], op[2], op[4]-op[3], new_content._lines[op[3]:op[4]]))
1809
return diff_hunks
1810
1811
def _merge_annotations(self, content, parents, parent_texts={},
1812
delta=None, annotated=None,
1813
left_matching_blocks=None):
1814
"""Merge annotations for content and generate deltas.
1815
1816
This is done by comparing the annotations based on changes to the text
1817
and generating a delta on the resulting full texts. If annotations are
1818
not being created then a simple delta is created.
1819
"""
1820
if left_matching_blocks is not None:
1821
delta_seq = diff._PrematchedMatcher(left_matching_blocks)
1822
else:
1823
delta_seq = None
1824
if annotated:
1825
for parent_key in parents:
1826
merge_content = self._get_content(parent_key, parent_texts)
1827
if (parent_key == parents[0] and delta_seq is not None):
1828
seq = delta_seq
1829
else:
1830
seq = patiencediff.PatienceSequenceMatcher(
1831
None, merge_content.text(), content.text())
1832
for i, j, n in seq.get_matching_blocks():
1833
if n == 0:
1834
continue
1835
# this copies (origin, text) pairs across to the new
1836
# content for any line that matches the last-checked
1837
# parent.
1838
content._lines[j:j+n] = merge_content._lines[i:i+n]
1839
# XXX: Robert says the following block is a workaround for a
1840
# now-fixed bug and it can probably be deleted. -- mbp 20080618
1841
if content._lines and content._lines[-1][1][-1] != '\n':
1842
# The copied annotation was from a line without a trailing EOL,
1843
# reinstate one for the content object, to ensure correct
1844
# serialization.
1845
line = content._lines[-1][1] + '\n'
1846
content._lines[-1] = (content._lines[-1][0], line)
1847
if delta:
1848
if delta_seq is None:
1849
reference_content = self._get_content(parents[0], parent_texts)
1850
new_texts = content.text()
1851
old_texts = reference_content.text()
1852
delta_seq = patiencediff.PatienceSequenceMatcher(
1853
None, old_texts, new_texts)
1854
return self._make_line_delta(delta_seq, content)
1855
1856
def _parse_record(self, version_id, data):
1857
"""Parse an original format knit record.
1858
1859
These have the last element of the key only present in the stored data.
1860
"""
1861
rec, record_contents = self._parse_record_unchecked(data)
1862
self._check_header_version(rec, version_id)
1863
return record_contents, rec[3]
1864
1865
def _parse_record_header(self, key, raw_data):
1866
"""Parse a record header for consistency.
1867
1868
:return: the header and the decompressor stream.
1869
as (stream, header_record)
1870
"""
1871
df = tuned_gzip.GzipFile(mode='rb', fileobj=StringIO(raw_data))
1872
try:
1873
# Current serialise
1874
rec = self._check_header(key, df.readline())
1875
except Exception, e:
1876
raise KnitCorrupt(self,
1877
"While reading {%s} got %s(%s)"
1878
% (key, e.__class__.__name__, str(e)))
1879
return df, rec
1880
1881
def _parse_record_unchecked(self, data):
1882
# profiling notes:
1883
# 4168 calls in 2880 217 internal
1884
# 4168 calls to _parse_record_header in 2121
1885
# 4168 calls to readlines in 330
1886
df = tuned_gzip.GzipFile(mode='rb', fileobj=StringIO(data))
1887
try:
1888
record_contents = df.readlines()
1889
except Exception, e:
1890
raise KnitCorrupt(self, "Corrupt compressed record %r, got %s(%s)" %
1891
(data, e.__class__.__name__, str(e)))
1892
header = record_contents.pop(0)
1893
rec = self._split_header(header)
1894
last_line = record_contents.pop()
1895
if len(record_contents) != int(rec[2]):
1896
raise KnitCorrupt(self,
1897
'incorrect number of lines %s != %s'
1898
' for version {%s} %s'
1899
% (len(record_contents), int(rec[2]),
1900
rec[1], record_contents))
1901
if last_line != 'end %s\n' % rec[1]:
1902
raise KnitCorrupt(self,
1903
'unexpected version end line %r, wanted %r'
1904
% (last_line, rec[1]))
1905
df.close()
1906
return rec, record_contents
1907
1908
def _read_records_iter(self, records):
1909
"""Read text records from data file and yield result.
1910
1911
The result will be returned in whatever is the fastest to read.
1912
Not by the order requested. Also, multiple requests for the same
1913
record will only yield 1 response.
1914
:param records: A list of (key, access_memo) entries
1915
:return: Yields (key, contents, digest) in the order
1916
read, not the order requested
1917
"""
1918
if not records:
1919
return
1920
1921
# XXX: This smells wrong, IO may not be getting ordered right.
1922
needed_records = sorted(set(records), key=operator.itemgetter(1))
1923
if not needed_records:
1924
return
1925
1926
# The transport optimizes the fetching as well
1927
# (ie, reads continuous ranges.)
1928
raw_data = self._access.get_raw_records(
1929
[index_memo for key, index_memo in needed_records])
1930
1931
for (key, index_memo), data in \
1932
izip(iter(needed_records), raw_data):
1933
content, digest = self._parse_record(key[-1], data)
1934
yield key, content, digest
1935
1936
def _read_records_iter_raw(self, records):
1937
"""Read text records from data file and yield raw data.
1938
1939
This unpacks enough of the text record to validate the id is
1940
as expected but thats all.
1941
1942
Each item the iterator yields is (key, bytes,
1943
expected_sha1_of_full_text).
1944
"""
1945
for key, data in self._read_records_iter_unchecked(records):
1946
# validate the header (note that we can only use the suffix in
1947
# current knit records).
1948
df, rec = self._parse_record_header(key, data)
1949
df.close()
1950
yield key, data, rec[3]
1951
1952
def _read_records_iter_unchecked(self, records):
1953
"""Read text records from data file and yield raw data.
1954
1955
No validation is done.
1956
1957
Yields tuples of (key, data).
1958
"""
1959
# setup an iterator of the external records:
1960
# uses readv so nice and fast we hope.
1961
if len(records):
1962
# grab the disk data needed.
1963
needed_offsets = [index_memo for key, index_memo
1964
in records]
1965
raw_records = self._access.get_raw_records(needed_offsets)
1966
1967
for key, index_memo in records:
1968
data = raw_records.next()
1969
yield key, data
1970
1971
def _record_to_data(self, key, digest, lines, dense_lines=None):
1972
"""Convert key, digest, lines into a raw data block.
1973
1974
:param key: The key of the record. Currently keys are always serialised
1975
using just the trailing component.
1976
:param dense_lines: The bytes of lines but in a denser form. For
1977
instance, if lines is a list of 1000 bytestrings each ending in \n,
1978
dense_lines may be a list with one line in it, containing all the
1979
1000's lines and their \n's. Using dense_lines if it is already
1980
known is a win because the string join to create bytes in this
1981
function spends less time resizing the final string.
1982
:return: (len, a StringIO instance with the raw data ready to read.)
1983
"""
1984
chunks = ["version %s %d %s\n" % (key[-1], len(lines), digest)]
1985
chunks.extend(dense_lines or lines)
1986
chunks.append("end %s\n" % key[-1])
1987
for chunk in chunks:
1988
if type(chunk) is not str:
1989
raise AssertionError(
1990
'data must be plain bytes was %s' % type(chunk))
1991
if lines and lines[-1][-1] != '\n':
1992
raise ValueError('corrupt lines value %r' % lines)
1993
compressed_bytes = tuned_gzip.chunks_to_gzip(chunks)
1994
return len(compressed_bytes), compressed_bytes
1995
1996
def _split_header(self, line):
1997
rec = line.split()
1998
if len(rec) != 4:
1999
raise KnitCorrupt(self,
2000
'unexpected number of elements in record header')
2001
return rec
2002
2003
def keys(self):
2004
"""See VersionedFiles.keys."""
2005
if 'evil' in debug.debug_flags:
2006
trace.mutter_callsite(2, "keys scales with size of history")
2007
sources = [self._index] + self._fallback_vfs
2008
result = set()
2009
for source in sources:
2010
result.update(source.keys())
2011
return result
2012
2013
2014
class _ContentMapGenerator(object):
2015
"""Generate texts or expose raw deltas for a set of texts."""
2016
2017
def __init__(self, ordering='unordered'):
2018
self._ordering = ordering
2019
2020
def _get_content(self, key):
2021
"""Get the content object for key."""
2022
# Note that _get_content is only called when the _ContentMapGenerator
2023
# has been constructed with just one key requested for reconstruction.
2024
if key in self.nonlocal_keys:
2025
record = self.get_record_stream().next()
2026
# Create a content object on the fly
2027
lines = osutils.chunks_to_lines(record.get_bytes_as('chunked'))
2028
return PlainKnitContent(lines, record.key)
2029
else:
2030
# local keys we can ask for directly
2031
return self._get_one_work(key)
2032
2033
def get_record_stream(self):
2034
"""Get a record stream for the keys requested during __init__."""
2035
for record in self._work():
2036
yield record
2037
2038
def _work(self):
2039
"""Produce maps of text and KnitContents as dicts.
2040
681
682
The keys are version_ids, the values are tuples of (method, content,
683
digest, next).
684
method is the way the content should be applied.
685
content is a KnitContent object.
686
digest is the SHA1 digest of this version id after all steps are done
687
next is the build-parent of the version, i.e. the leftmost ancestor.
688
If the method is fulltext, next will be None.
689
"""
690
position_map = self._get_components_positions(version_ids)
691
# c = component_id, m = method, p = position, s = size, n = next
692
records = [(c, p, s) for c, (m, p, s, n) in position_map.iteritems()]
693
record_map = {}
694
for component_id, content, digest in\
695
self._data.read_records_iter(records):
696
method, position, size, next = position_map[component_id]
697
record_map[component_id] = method, content, digest, next
698
699
return record_map
700
701
def get_text(self, version_id):
702
"""See VersionedFile.get_text"""
703
return self.get_texts([version_id])[0]
704
705
def get_texts(self, version_ids):
706
return [''.join(l) for l in self.get_line_list(version_ids)]
707
708
def get_line_list(self, version_ids):
709
"""Return the texts of listed versions as a list of strings."""
710
text_map, content_map = self._get_content_maps(version_ids)
711
return [text_map[v] for v in version_ids]
712
713
def _get_content_maps(self, version_ids):
714
"""Produce maps of text and KnitContents
715
2041
716
:return: (text_map, content_map) where text_map contains the texts for
2042
the requested versions and content_map contains the KnitContents.
717
the requested versions and content_map contains the KnitContents.
718
Both dicts take version_ids as their keys.
2043
719
"""
2044
# NB: By definition we never need to read remote sources unless texts
2045
# are requested from them: we don't delta across stores - and we
2046
# explicitly do not want to to prevent data loss situations.
2047
if self.global_map is None:
2048
self.global_map = self.vf.get_parent_map(self.keys)
2049
nonlocal_keys = self.nonlocal_keys
2050
2051
missing_keys = set(nonlocal_keys)
2052
# Read from remote versioned file instances and provide to our caller.
2053
for source in self.vf._fallback_vfs:
2054
if not missing_keys:
2055
break
2056
# Loop over fallback repositories asking them for texts - ignore
2057
# any missing from a particular fallback.
2058
for record in source.get_record_stream(missing_keys,
2059
self._ordering, True):
2060
if record.storage_kind == 'absent':
2061
# Not in thie particular stream, may be in one of the
2062
# other fallback vfs objects.
2063
continue
2064
missing_keys.remove(record.key)
2065
yield record
2066
2067
if self._raw_record_map is None:
2068
raise AssertionError('_raw_record_map should have been filled')
2069
first = True
2070
for key in self.keys:
2071
if key in self.nonlocal_keys:
2072
continue
2073
yield LazyKnitContentFactory(key, self.global_map[key], self, first)
2074
first = False
2075
2076
def _get_one_work(self, requested_key):
2077
# Now, if we have calculated everything already, just return the
2078
# desired text.
2079
if requested_key in self._contents_map:
2080
return self._contents_map[requested_key]
2081
# To simplify things, parse everything at once - code that wants one text
2082
# probably wants them all.
2083
# FUTURE: This function could be improved for the 'extract many' case
2084
# by tracking each component and only doing the copy when the number of
2085
# children than need to apply delta's to it is > 1 or it is part of the
2086
# final output.
2087
multiple_versions = len(self.keys) != 1
2088
if self._record_map is None:
2089
self._record_map = self.vf._raw_map_to_record_map(
2090
self._raw_record_map)
2091
record_map = self._record_map
2092
# raw_record_map is key:
2093
# Have read and parsed records at this point.
2094
for key in self.keys:
2095
if key in self.nonlocal_keys:
2096
# already handled
2097
continue
720
for version_id in version_ids:
721
if not self.has_version(version_id):
722
raise RevisionNotPresent(version_id, self.filename)
723
record_map = self._get_record_map(version_ids)
724
725
text_map = {}
726
content_map = {}
727
final_content = {}
728
for version_id in version_ids:
2098
729
components = []
2099
cursor = key
730
cursor = version_id
2100
731
while cursor is not None:
2101
try:
2102
record, record_details, digest, next = record_map[cursor]
2103
except KeyError:
2104
raise RevisionNotPresent(cursor, self)
2105
components.append((cursor, record, record_details, digest))
732
method, data, digest, next = record_map[cursor]
733
components.append((cursor, method, data, digest))
734
if cursor in content_map:
735
break
2106
736
cursor = next
2107
if cursor in self._contents_map:
2108
# no need to plan further back
2109
components.append((cursor, None, None, None))
2110
break
2111
737
2112
738
content = None
2113
for (component_id, record, record_details,
2114
digest) in reversed(components):
2115
if component_id in self._contents_map:
2116
content = self._contents_map[component_id]
739
for component_id, method, data, digest in reversed(components):
740
if component_id in content_map:
741
content = content_map[component_id]
2117
742
else:
2118
content, delta = self._factory.parse_record(key[-1],
2119
record, record_details, content,
2120
copy_base_content=multiple_versions)
2121
if multiple_versions:
2122
self._contents_map[component_id] = content
743
version_idx = self._index.lookup(component_id)
744
if method == 'fulltext':
745
assert content is None
746
content = self.factory.parse_fulltext(data, version_idx)
747
elif method == 'line-delta':
748
delta = self.factory.parse_line_delta(data[:],
749
version_idx)
750
content = content.copy()
751
content._lines = self._apply_delta(content._lines,
752
delta)
753
content_map[component_id] = content
754
755
if 'no-eol' in self._index.get_options(version_id):
756
content = content.copy()
757
line = content._lines[-1][1].rstrip('\n')
758
content._lines[-1] = (content._lines[-1][0], line)
759
final_content[version_id] = content
2123
760
2124
761
# digest here is the digest from the last applied component.
2125
762
text = content.text()
2126
actual_sha = sha_strings(text)
2127
if actual_sha != digest:
2128
raise SHA1KnitCorrupt(self, actual_sha, digest, key, text)
2129
if multiple_versions:
2130
return self._contents_map[requested_key]
2131
else:
2132
return content
2133
2134
def _wire_bytes(self):
2135
"""Get the bytes to put on the wire for 'key'.
2136
2137
The first collection of bytes asked for returns the serialised
2138
raw_record_map and the additional details (key, parent) for key.
2139
Subsequent calls return just the additional details (key, parent).
2140
The wire storage_kind given for the first key is 'knit-delta-closure',
2141
For subsequent keys it is 'knit-delta-closure-ref'.
2142
2143
:param key: A key from the content generator.
2144
:return: Bytes to put on the wire.
2145
"""
2146
lines = []
2147
# kind marker for dispatch on the far side,
2148
lines.append('knit-delta-closure')
2149
# Annotated or not
2150
if self.vf._factory.annotated:
2151
lines.append('annotated')
2152
else:
2153
lines.append('')
2154
# then the list of keys
2155
lines.append('\t'.join(['\x00'.join(key) for key in self.keys
2156
if key not in self.nonlocal_keys]))
2157
# then the _raw_record_map in serialised form:
2158
map_byte_list = []
2159
# for each item in the map:
2160
# 1 line with key
2161
# 1 line with parents if the key is to be yielded (None: for None, '' for ())
2162
# one line with method
2163
# one line with noeol
2164
# one line with next ('' for None)
2165
# one line with byte count of the record bytes
2166
# the record bytes
2167
for key, (record_bytes, (method, noeol), next) in \
2168
self._raw_record_map.iteritems():
2169
key_bytes = '\x00'.join(key)
2170
parents = self.global_map.get(key, None)
2171
if parents is None:
2172
parent_bytes = 'None:'
2173
else:
2174
parent_bytes = '\t'.join('\x00'.join(key) for key in parents)
2175
method_bytes = method
2176
if noeol:
2177
noeol_bytes = "T"
2178
else:
2179
noeol_bytes = "F"
2180
if next:
2181
next_bytes = '\x00'.join(next)
2182
else:
2183
next_bytes = ''
2184
map_byte_list.append('%s\n%s\n%s\n%s\n%s\n%d\n%s' % (
2185
key_bytes, parent_bytes, method_bytes, noeol_bytes, next_bytes,
2186
len(record_bytes), record_bytes))
2187
map_bytes = ''.join(map_byte_list)
2188
lines.append(map_bytes)
2189
bytes = '\n'.join(lines)
2190
return bytes
2191
2192
2193
class _VFContentMapGenerator(_ContentMapGenerator):
2194
"""Content map generator reading from a VersionedFiles object."""
2195
2196
def __init__(self, versioned_files, keys, nonlocal_keys=None,
2197
global_map=None, raw_record_map=None, ordering='unordered'):
2198
"""Create a _ContentMapGenerator.
2199
2200
:param versioned_files: The versioned files that the texts are being
2201
extracted from.
2202
:param keys: The keys to produce content maps for.
2203
:param nonlocal_keys: An iterable of keys(possibly intersecting keys)
2204
which are known to not be in this knit, but rather in one of the
2205
fallback knits.
2206
:param global_map: The result of get_parent_map(keys) (or a supermap).
2207
This is required if get_record_stream() is to be used.
2208
:param raw_record_map: A unparsed raw record map to use for answering
2209
contents.
2210
"""
2211
_ContentMapGenerator.__init__(self, ordering=ordering)
2212
# The vf to source data from
2213
self.vf = versioned_files
2214
# The keys desired
2215
self.keys = list(keys)
2216
# Keys known to be in fallback vfs objects
2217
if nonlocal_keys is None:
2218
self.nonlocal_keys = set()
2219
else:
2220
self.nonlocal_keys = frozenset(nonlocal_keys)
2221
# Parents data for keys to be returned in get_record_stream
2222
self.global_map = global_map
2223
# The chunked lists for self.keys in text form
2224
self._text_map = {}
2225
# A cache of KnitContent objects used in extracting texts.
2226
self._contents_map = {}
2227
# All the knit records needed to assemble the requested keys as full
2228
# texts.
2229
self._record_map = None
2230
if raw_record_map is None:
2231
self._raw_record_map = self.vf._get_record_map_unparsed(keys,
2232
allow_missing=True)
2233
else:
2234
self._raw_record_map = raw_record_map
2235
# the factory for parsing records
2236
self._factory = self.vf._factory
2237
2238
2239
class _NetworkContentMapGenerator(_ContentMapGenerator):
2240
"""Content map generator sourced from a network stream."""
2241
2242
def __init__(self, bytes, line_end):
2243
"""Construct a _NetworkContentMapGenerator from a bytes block."""
2244
self._bytes = bytes
2245
self.global_map = {}
2246
self._raw_record_map = {}
2247
self._contents_map = {}
2248
self._record_map = None
2249
self.nonlocal_keys = []
2250
# Get access to record parsing facilities
2251
self.vf = KnitVersionedFiles(None, None)
2252
start = line_end
2253
# Annotated or not
2254
line_end = bytes.find('\n', start)
2255
line = bytes[start:line_end]
2256
start = line_end + 1
2257
if line == 'annotated':
2258
self._factory = KnitAnnotateFactory()
2259
else:
2260
self._factory = KnitPlainFactory()
2261
# list of keys to emit in get_record_stream
2262
line_end = bytes.find('\n', start)
2263
line = bytes[start:line_end]
2264
start = line_end + 1
2265
self.keys = [
2266
tuple(segment.split('\x00')) for segment in line.split('\t')
2267
if segment]
2268
# now a loop until the end. XXX: It would be nice if this was just a
2269
# bunch of the same records as get_record_stream(..., False) gives, but
2270
# there is a decent sized gap stopping that at the moment.
2271
end = len(bytes)
2272
while start < end:
2273
# 1 line with key
2274
line_end = bytes.find('\n', start)
2275
key = tuple(bytes[start:line_end].split('\x00'))
2276
start = line_end + 1
2277
# 1 line with parents (None: for None, '' for ())
2278
line_end = bytes.find('\n', start)
2279
line = bytes[start:line_end]
2280
if line == 'None:':
2281
parents = None
2282
else:
2283
parents = tuple(
2284
[tuple(segment.split('\x00')) for segment in line.split('\t')
2285
if segment])
2286
self.global_map[key] = parents
2287
start = line_end + 1
2288
# one line with method
2289
line_end = bytes.find('\n', start)
2290
line = bytes[start:line_end]
2291
method = line
2292
start = line_end + 1
2293
# one line with noeol
2294
line_end = bytes.find('\n', start)
2295
line = bytes[start:line_end]
2296
noeol = line == "T"
2297
start = line_end + 1
2298
# one line with next ('' for None)
2299
line_end = bytes.find('\n', start)
2300
line = bytes[start:line_end]
2301
if not line:
2302
next = None
2303
else:
2304
next = tuple(bytes[start:line_end].split('\x00'))
2305
start = line_end + 1
2306
# one line with byte count of the record bytes
2307
line_end = bytes.find('\n', start)
2308
line = bytes[start:line_end]
2309
count = int(line)
2310
start = line_end + 1
2311
# the record bytes
2312
record_bytes = bytes[start:start+count]
2313
start = start + count
2314
# put it in the map
2315
self._raw_record_map[key] = (record_bytes, (method, noeol), next)
2316
2317
def get_record_stream(self):
2318
"""Get a record stream for for keys requested by the bytestream."""
2319
first = True
2320
for key in self.keys:
2321
yield LazyKnitContentFactory(key, self.global_map[key], self, first)
2322
first = False
2323
2324
def _wire_bytes(self):
2325
return self._bytes
2326
2327
2328
class _KndxIndex(object):
2329
"""Manages knit index files
2330
2331
The index is kept in memory and read on startup, to enable
763
if sha_strings(text) != digest:
764
raise KnitCorrupt(self.filename,
765
'sha-1 does not match %s' % version_id)
766
767
text_map[version_id] = text
768
return text_map, final_content
769
770
def iter_lines_added_or_present_in_versions(self, version_ids=None):
771
"""See VersionedFile.iter_lines_added_or_present_in_versions()."""
772
if version_ids is None:
773
version_ids = self.versions()
774
# we don't care about inclusions, the caller cares.
775
# but we need to setup a list of records to visit.
776
# we need version_id, position, length
777
version_id_records = []
778
requested_versions = list(version_ids)
779
# filter for available versions
780
for version_id in requested_versions:
781
if not self.has_version(version_id):
782
raise RevisionNotPresent(version_id, self.filename)
783
# get a in-component-order queue:
784
version_ids = []
785
for version_id in self.versions():
786
if version_id in requested_versions:
787
version_ids.append(version_id)
788
data_pos, length = self._index.get_position(version_id)
789
version_id_records.append((version_id, data_pos, length))
790
791
pb = bzrlib.ui.ui_factory.nested_progress_bar()
792
count = 0
793
total = len(version_id_records)
794
try:
795
pb.update('Walking content.', count, total)
796
for version_id, data, sha_value in \
797
self._data.read_records_iter(version_id_records):
798
pb.update('Walking content.', count, total)
799
method = self._index.get_method(version_id)
800
version_idx = self._index.lookup(version_id)
801
assert method in ('fulltext', 'line-delta')
802
if method == 'fulltext':
803
content = self.factory.parse_fulltext(data, version_idx)
804
for line in content.text():
805
yield line
806
else:
807
delta = self.factory.parse_line_delta(data, version_idx)
808
for start, end, count, lines in delta:
809
for origin, line in lines:
810
yield line
811
count +=1
812
pb.update('Walking content.', total, total)
813
pb.finished()
814
except:
815
pb.update('Walking content.', total, total)
816
pb.finished()
817
raise
818
819
def num_versions(self):
820
"""See VersionedFile.num_versions()."""
821
return self._index.num_versions()
822
823
__len__ = num_versions
824
825
def annotate_iter(self, version_id):
826
"""See VersionedFile.annotate_iter."""
827
content = self._get_content(version_id)
828
for origin, text in content.annotate_iter():
829
yield origin, text
830
831
def get_parents(self, version_id):
832
"""See VersionedFile.get_parents."""
833
# perf notes:
834
# optimism counts!
835
# 52554 calls in 1264 872 internal down from 3674
836
try:
837
return self._index.get_parents(version_id)
838
except KeyError:
839
raise RevisionNotPresent(version_id, self.filename)
840
841
def get_parents_with_ghosts(self, version_id):
842
"""See VersionedFile.get_parents."""
843
try:
844
return self._index.get_parents_with_ghosts(version_id)
845
except KeyError:
846
raise RevisionNotPresent(version_id, self.filename)
847
848
def get_ancestry(self, versions):
849
"""See VersionedFile.get_ancestry."""
850
if isinstance(versions, basestring):
851
versions = [versions]
852
if not versions:
853
return []
854
self._check_versions_present(versions)
855
return self._index.get_ancestry(versions)
856
857
def get_ancestry_with_ghosts(self, versions):
858
"""See VersionedFile.get_ancestry_with_ghosts."""
859
if isinstance(versions, basestring):
860
versions = [versions]
861
if not versions:
862
return []
863
self._check_versions_present(versions)
864
return self._index.get_ancestry_with_ghosts(versions)
865
866
#@deprecated_method(zero_eight)
867
def walk(self, version_ids):
868
"""See VersionedFile.walk."""
869
# We take the short path here, and extract all relevant texts
870
# and put them in a weave and let that do all the work. Far
871
# from optimal, but is much simpler.
872
# FIXME RB 20060228 this really is inefficient!
873
from bzrlib.weave import Weave
874
875
w = Weave(self.filename)
876
ancestry = self.get_ancestry(version_ids)
877
sorted_graph = topo_sort(self._index.get_graph())
878
version_list = [vid for vid in sorted_graph if vid in ancestry]
879
880
for version_id in version_list:
881
lines = self.get_lines(version_id)
882
w.add_lines(version_id, self.get_parents(version_id), lines)
883
884
for lineno, insert_id, dset, line in w.walk(version_ids):
885
yield lineno, insert_id, dset, line
886
887
def plan_merge(self, ver_a, ver_b):
888
"""See VersionedFile.plan_merge."""
889
ancestors_b = set(self.get_ancestry(ver_b))
890
def status_a(revision, text):
891
if revision in ancestors_b:
892
return 'killed-b', text
893
else:
894
return 'new-a', text
895
896
ancestors_a = set(self.get_ancestry(ver_a))
897
def status_b(revision, text):
898
if revision in ancestors_a:
899
return 'killed-a', text
900
else:
901
return 'new-b', text
902
903
annotated_a = self.annotate(ver_a)
904
annotated_b = self.annotate(ver_b)
905
plain_a = [t for (a, t) in annotated_a]
906
plain_b = [t for (a, t) in annotated_b]
907
blocks = KnitSequenceMatcher(None, plain_a, plain_b).get_matching_blocks()
908
a_cur = 0
909
b_cur = 0
910
for ai, bi, l in blocks:
911
# process all mismatched sections
912
# (last mismatched section is handled because blocks always
913
# includes a 0-length last block)
914
for revision, text in annotated_a[a_cur:ai]:
915
yield status_a(revision, text)
916
for revision, text in annotated_b[b_cur:bi]:
917
yield status_b(revision, text)
918
919
# and now the matched section
920
a_cur = ai + l
921
b_cur = bi + l
922
for text_a, text_b in zip(plain_a[ai:a_cur], plain_b[bi:b_cur]):
923
assert text_a == text_b
924
yield "unchanged", text_a
925
926
927
class _KnitComponentFile(object):
928
"""One of the files used to implement a knit database"""
929
930
def __init__(self, transport, filename, mode, file_mode=None):
931
self._transport = transport
932
self._filename = filename
933
self._mode = mode
934
self._file_mode=file_mode
935
936
def write_header(self):
937
if self._transport.append(self._filename, StringIO(self.HEADER),
938
mode=self._file_mode):
939
raise KnitCorrupt(self._filename, 'misaligned after writing header')
940
941
def check_header(self, fp):
942
line = fp.readline()
943
if line != self.HEADER:
944
raise KnitHeaderError(badline=line)
945
946
def commit(self):
947
"""Commit is a nop."""
948
949
def __repr__(self):
950
return '%s(%s)' % (self.__class__.__name__, self._filename)
951
952
953
class _KnitIndex(_KnitComponentFile):
954
"""Manages knit index file.
955
956
The index is already kept in memory and read on startup, to enable
2332
957
fast lookups of revision information. The cursor of the index
2333
958
file is always pointing to the end, making it easy to append
2334
959
entries.
2344
969
2345
970
Duplicate entries may be written to the index for a single version id
2346
971
if this is done then the latter one completely replaces the former:
2347
this allows updates to correct version and parent information.
972
this allows updates to correct version and parent information.
2348
973
Note that the two entries may share the delta, and that successive
2349
974
annotations and references MUST point to the first entry.
2350
975
2351
976
The index file on disc contains a header, followed by one line per knit
2352
977
record. The same revision can be present in an index file more than once.
2353
The first occurrence gets assigned a sequence number starting from 0.
2354
978
The first occurrence gets assigned a sequence number starting from 0.
979
2355
980
The format of a single line is
2356
981
REVISION_ID FLAGS BYTE_OFFSET LENGTH( PARENT_ID|PARENT_SEQUENCE_ID)* :\n
2357
982
REVISION_ID is a utf8-encoded revision id
2358
FLAGS is a comma separated list of flags about the record. Values include
983
FLAGS is a comma separated list of flags about the record. Values include
2359
984
no-eol, line-delta, fulltext.
2360
985
BYTE_OFFSET is the ascii representation of the byte offset in the data file
2361
986
that the the compressed data starts at.
2365
990
PARENT_SEQUENCE_ID the ascii representation of the sequence number of a
2366
991
revision id already in the knit that is a parent of REVISION_ID.
2367
992
The ' :' marker is the end of record marker.
2368
993
2369
994
partial writes:
2370
when a write is interrupted to the index file, it will result in a line
2371
that does not end in ' :'. If the ' :' is not present at the end of a line,
2372
or at the end of the file, then the record that is missing it will be
2373
ignored by the parser.
995
when a write is interrupted to the index file, it will result in a line that
996
does not end in ' :'. If the ' :' is not present at the end of a line, or at
997
the end of the file, then the record that is missing it will be ignored by
998
the parser.
2374
999
2375
1000
When writing new records to the index file, the data is preceded by '\n'
2376
1001
to ensure that records always start on new lines even if the last write was
2377
1002
interrupted. As a result its normal for the last line in the index to be
2378
1003
missing a trailing newline. One can be added with no harmful effects.
2379
2380
:ivar _kndx_cache: dict from prefix to the old state of KnitIndex objects,
2381
where prefix is e.g. the (fileid,) for .texts instances or () for
2382
constant-mapped things like .revisions, and the old state is
2383
tuple(history_vector, cache_dict). This is used to prevent having an
2384
ABI change with the C extension that reads .kndx files.
2385
1004
"""
2386
1005
2387
1006
HEADER = "# bzr knit index 8\n"
2388
1007
2389
def __init__(self, transport, mapper, get_scope, allow_writes, is_locked):
2390
"""Create a _KndxIndex on transport using mapper."""
2391
self._transport = transport
2392
self._mapper = mapper
2393
self._get_scope = get_scope
2394
self._allow_writes = allow_writes
2395
self._is_locked = is_locked
2396
self._reset_cache()
2397
self.has_graph = True
2398
2399
def add_records(self, records, random_id=False, missing_compression_parents=False):
2400
"""Add multiple records to the index.
2401
2402
:param records: a list of tuples:
2403
(key, options, access_memo, parents).
2404
:param random_id: If True the ids being added were randomly generated
2405
and no check for existence will be performed.
2406
:param missing_compression_parents: If True the records being added are
2407
only compressed against texts already in the index (or inside
2408
records). If False the records all refer to unavailable texts (or
2409
texts inside records) as compression parents.
2410
"""
2411
if missing_compression_parents:
2412
# It might be nice to get the edge of the records. But keys isn't
2413
# _wrong_.
2414
keys = sorted(record[0] for record in records)
2415
raise errors.RevisionNotPresent(keys, self)
2416
paths = {}
2417
for record in records:
2418
key = record[0]
2419
prefix = key[:-1]
2420
path = self._mapper.map(key) + '.kndx'
2421
path_keys = paths.setdefault(path, (prefix, []))
2422
path_keys[1].append(record)
2423
for path in sorted(paths):
2424
prefix, path_keys = paths[path]
2425
self._load_prefixes([prefix])
2426
lines = []
2427
orig_history = self._kndx_cache[prefix][1][:]
2428
orig_cache = self._kndx_cache[prefix][0].copy()
2429
2430
try:
2431
for key, options, (_, pos, size), parents in path_keys:
2432
if parents is None:
2433
# kndx indices cannot be parentless.
2434
parents = ()
2435
line = "\n%s %s %s %s %s :" % (
2436
key[-1], ','.join(options), pos, size,
2437
self._dictionary_compress(parents))
2438
if type(line) is not str:
2439
raise AssertionError(
2440
'data must be utf8 was %s' % type(line))
2441
lines.append(line)
2442
self._cache_key(key, options, pos, size, parents)
2443
if len(orig_history):
2444
self._transport.append_bytes(path, ''.join(lines))
2445
else:
2446
self._init_index(path, lines)
2447
except:
2448
# If any problems happen, restore the original values and re-raise
2449
self._kndx_cache[prefix] = (orig_cache, orig_history)
2450
raise
2451
2452
def scan_unvalidated_index(self, graph_index):
2453
"""See _KnitGraphIndex.scan_unvalidated_index."""
2454
# Because kndx files do not support atomic insertion via separate index
2455
# files, they do not support this method.
2456
raise NotImplementedError(self.scan_unvalidated_index)
2457
2458
def get_missing_compression_parents(self):
2459
"""See _KnitGraphIndex.get_missing_compression_parents."""
2460
# Because kndx files do not support atomic insertion via separate index
2461
# files, they do not support this method.
2462
raise NotImplementedError(self.get_missing_compression_parents)
2463
2464
def _cache_key(self, key, options, pos, size, parent_keys):
1008
# speed of knit parsing went from 280 ms to 280 ms with slots addition.
1009
# __slots__ = ['_cache', '_history', '_transport', '_filename']
1010
1011
def _cache_version(self, version_id, options, pos, size, parents):
2465
1012
"""Cache a version record in the history array and index cache.
2466
2467
This is inlined into _load_data for performance. KEEP IN SYNC.
1013
1014
This is inlined into __init__ for performance. KEEP IN SYNC.
2468
1015
(It saves 60ms, 25% of the __init__ overhead on local 4000 record
2469
1016
indexes).
2470
1017
"""
2471
prefix = key[:-1]
2472
version_id = key[-1]
2473
# last-element only for compatibilty with the C load_data.
2474
parents = tuple(parent[-1] for parent in parent_keys)
2475
for parent in parent_keys:
2476
if parent[:-1] != prefix:
2477
raise ValueError("mismatched prefixes for %r, %r" % (
2478
key, parent_keys))
2479
cache, history = self._kndx_cache[prefix]
2480
1018
# only want the _history index to reference the 1st index entry
2481
1019
# for version_id
2482
if version_id not in cache:
2483
index = len(history)
2484
history.append(version_id)
1020
if version_id not in self._cache:
1021
index = len(self._history)
1022
self._history.append(version_id)
2485
1023
else:
2486
index = cache[version_id][5]
2487
cache[version_id] = (version_id,
1024
index = self._cache[version_id][5]
1025
self._cache[version_id] = (version_id,
2488
1026
options,
2489
1027
pos,
2490
1028
size,
2491
1029
parents,
2492
1030
index)
2493
1031
2494
def check_header(self, fp):
2495
line = fp.readline()
2496
if line == '':
2497
# An empty file can actually be treated as though the file doesn't
2498
# exist yet.
2499
raise errors.NoSuchFile(self)
2500
if line != self.HEADER:
2501
raise KnitHeaderError(badline=line, filename=self)
2502
2503
def _check_read(self):
2504
if not self._is_locked():
2505
raise errors.ObjectNotLocked(self)
2506
if self._get_scope() != self._scope:
2507
self._reset_cache()
2508
2509
def _check_write_ok(self):
2510
"""Assert if not writes are permitted."""
2511
if not self._is_locked():
2512
raise errors.ObjectNotLocked(self)
2513
if self._get_scope() != self._scope:
2514
self._reset_cache()
2515
if self._mode != 'w':
2516
raise errors.ReadOnlyObjectDirtiedError(self)
2517
2518
def get_build_details(self, keys):
2519
"""Get the method, index_memo and compression parent for keys.
2520
2521
Ghosts are omitted from the result.
2522
2523
:param keys: An iterable of keys.
2524
:return: A dict of key:(index_memo, compression_parent, parents,
2525
record_details).
2526
index_memo
2527
opaque structure to pass to read_records to extract the raw
2528
data
2529
compression_parent
2530
Content that this record is built upon, may be None
2531
parents
2532
Logical parents of this node
2533
record_details
2534
extra information about the content which needs to be passed to
2535
Factory.parse_record
2536
"""
2537
parent_map = self.get_parent_map(keys)
2538
result = {}
2539
for key in keys:
2540
if key not in parent_map:
2541
continue # Ghost
2542
method = self.get_method(key)
2543
parents = parent_map[key]
2544
if method == 'fulltext':
2545
compression_parent = None
2546
else:
2547
compression_parent = parents[0]
2548
noeol = 'no-eol' in self.get_options(key)
2549
index_memo = self.get_position(key)
2550
result[key] = (index_memo, compression_parent,
2551
parents, (method, noeol))
2552
return result
2553
2554
def get_method(self, key):
2555
"""Return compression method of specified key."""
2556
options = self.get_options(key)
2557
if 'fulltext' in options:
2558
return 'fulltext'
2559
elif 'line-delta' in options:
2560
return 'line-delta'
2561
else:
2562
raise errors.KnitIndexUnknownMethod(self, options)
2563
2564
def get_options(self, key):
2565
"""Return a list representing options.
2566
2567
e.g. ['foo', 'bar']
2568
"""
2569
prefix, suffix = self._split_key(key)
2570
self._load_prefixes([prefix])
1032
def __init__(self, transport, filename, mode, create=False, file_mode=None):
1033
_KnitComponentFile.__init__(self, transport, filename, mode, file_mode)
1034
self._cache = {}
1035
# position in _history is the 'official' index for a revision
1036
# but the values may have come from a newer entry.
1037
# so - wc -l of a knit index is != the number of unique names
1038
# in the knit.
1039
self._history = []
1040
pb = bzrlib.ui.ui_factory.nested_progress_bar()
2571
1041
try:
2572
return self._kndx_cache[prefix][0][suffix][1]
2573
except KeyError:
2574
raise RevisionNotPresent(key, self)
2575
2576
def find_ancestry(self, keys):
2577
"""See CombinedGraphIndex.find_ancestry()"""
2578
prefixes = set(key[:-1] for key in keys)
2579
self._load_prefixes(prefixes)
2580
result = {}
2581
parent_map = {}
2582
missing_keys = set()
2583
pending_keys = list(keys)
2584
# This assumes that keys will not reference parents in a different
2585
# prefix, which is accurate so far.
2586
while pending_keys:
2587
key = pending_keys.pop()
2588
if key in parent_map:
2589
continue
2590
prefix = key[:-1]
1042
count = 0
1043
total = 1
2591
1044
try:
2592
suffix_parents = self._kndx_cache[prefix][0][key[-1]][4]
2593
except KeyError:
2594
missing_keys.add(key)
1045
pb.update('read knit index', count, total)
1046
fp = self._transport.get(self._filename)
1047
self.check_header(fp)
1048
# readlines reads the whole file at once:
1049
# bad for transports like http, good for local disk
1050
# we save 60 ms doing this one change (
1051
# from calling readline each time to calling
1052
# readlines once.
1053
# probably what we want for nice behaviour on
1054
# http is a incremental readlines that yields, or
1055
# a check for local vs non local indexes,
1056
for l in fp.readlines():
1057
rec = l.split()
1058
if len(rec) < 5 or rec[-1] != ':':
1059
# corrupt line.
1060
# FIXME: in the future we should determine if its a
1061
# short write - and ignore it
1062
# or a different failure, and raise. RBC 20060407
1063
continue
1064
count += 1
1065
total += 1
1066
#pb.update('read knit index', count, total)
1067
# See self._parse_parents
1068
parents = []
1069
for value in rec[4:-1]:
1070
if '.' == value[0]:
1071
# uncompressed reference
1072
parents.append(value[1:])
1073
else:
1074
# this is 15/4000ms faster than isinstance,
1075
# (in lsprof)
1076
# this function is called thousands of times a
1077
# second so small variations add up.
1078
assert value.__class__ is str
1079
parents.append(self._history[int(value)])
1080
# end self._parse_parents
1081
# self._cache_version(rec[0],
1082
# rec[1].split(','),
1083
# int(rec[2]),
1084
# int(rec[3]),
1085
# parents)
1086
# --- self._cache_version
1087
# only want the _history index to reference the 1st
1088
# index entry for version_id
1089
version_id = rec[0]
1090
if version_id not in self._cache:
1091
index = len(self._history)
1092
self._history.append(version_id)
1093
else:
1094
index = self._cache[version_id][5]
1095
self._cache[version_id] = (version_id,
1096
rec[1].split(','),
1097
int(rec[2]),
1098
int(rec[3]),
1099
parents,
1100
index)
1101
# --- self._cache_version
1102
except NoSuchFile, e:
1103
if mode != 'w' or not create:
1104
raise
1105
self.write_header()
1106
finally:
1107
pb.update('read knit index', total, total)
1108
pb.finished()
1109
1110
def _parse_parents(self, compressed_parents):
1111
"""convert a list of string parent values into version ids.
1112
1113
ints are looked up in the index.
1114
.FOO values are ghosts and converted in to FOO.
1115
1116
NOTE: the function is retained here for clarity, and for possible
1117
use in partial index reads. However bulk processing now has
1118
it inlined in __init__ for inner-loop optimisation.
1119
"""
1120
result = []
1121
for value in compressed_parents:
1122
if value[-1] == '.':
1123
# uncompressed reference
1124
result.append(value[1:])
2595
1125
else:
2596
parent_keys = tuple([prefix + (suffix,)
2597
for suffix in suffix_parents])
2598
parent_map[key] = parent_keys
2599
pending_keys.extend([p for p in parent_keys
2600
if p not in parent_map])
2601
return parent_map, missing_keys
2602
2603
def get_parent_map(self, keys):
2604
"""Get a map of the parents of keys.
2605
2606
:param keys: The keys to look up parents for.
2607
:return: A mapping from keys to parents. Absent keys are absent from
2608
the mapping.
2609
"""
2610
# Parse what we need to up front, this potentially trades off I/O
2611
# locality (.kndx and .knit in the same block group for the same file
2612
# id) for less checking in inner loops.
2613
prefixes = set(key[:-1] for key in keys)
2614
self._load_prefixes(prefixes)
2615
result = {}
2616
for key in keys:
2617
prefix = key[:-1]
1126
# this is 15/4000ms faster than isinstance,
1127
# this function is called thousands of times a
1128
# second so small variations add up.
1129
assert value.__class__ is str
1130
result.append(self._history[int(value)])
1131
return result
1132
1133
def get_graph(self):
1134
graph = []
1135
for version_id, index in self._cache.iteritems():
1136
graph.append((version_id, index[4]))
1137
return graph
1138
1139
def get_ancestry(self, versions):
1140
"""See VersionedFile.get_ancestry."""
1141
# get a graph of all the mentioned versions:
1142
graph = {}
1143
pending = set(versions)
1144
while len(pending):
1145
version = pending.pop()
1146
parents = self._cache[version][4]
1147
# got the parents ok
1148
# trim ghosts
1149
parents = [parent for parent in parents if parent in self._cache]
1150
for parent in parents:
1151
# if not completed and not a ghost
1152
if parent not in graph:
1153
pending.add(parent)
1154
graph[version] = parents
1155
return topo_sort(graph.items())
1156
1157
def get_ancestry_with_ghosts(self, versions):
1158
"""See VersionedFile.get_ancestry_with_ghosts."""
1159
# get a graph of all the mentioned versions:
1160
graph = {}
1161
pending = set(versions)
1162
while len(pending):
1163
version = pending.pop()
2618
1164
try:
2619
suffix_parents = self._kndx_cache[prefix][0][key[-1]][4]
1165
parents = self._cache[version][4]
2620
1166
except KeyError:
1167
# ghost, fake it
1168
graph[version] = []
2621
1169
pass
2622
1170
else:
2623
result[key] = tuple(prefix + (suffix,) for
2624
suffix in suffix_parents)
2625
return result
2626
2627
def get_position(self, key):
2628
"""Return details needed to access the version.
2629
2630
:return: a tuple (key, data position, size) to hand to the access
2631
logic to get the record.
2632
"""
2633
prefix, suffix = self._split_key(key)
2634
self._load_prefixes([prefix])
2635
entry = self._kndx_cache[prefix][0][suffix]
2636
return key, entry[2], entry[3]
2637
2638
has_key = _mod_index._has_key_from_parent_map
2639
2640
def _init_index(self, path, extra_lines=[]):
2641
"""Initialize an index."""
2642
sio = StringIO()
2643
sio.write(self.HEADER)
2644
sio.writelines(extra_lines)
2645
sio.seek(0)
2646
self._transport.put_file_non_atomic(path, sio,
2647
create_parent_dir=True)
2648
# self._create_parent_dir)
2649
# mode=self._file_mode,
2650
# dir_mode=self._dir_mode)
2651
2652
def keys(self):
2653
"""Get all the keys in the collection.
2654
2655
The keys are not ordered.
2656
"""
2657
result = set()
2658
# Identify all key prefixes.
2659
# XXX: A bit hacky, needs polish.
2660
if type(self._mapper) is ConstantMapper:
2661
prefixes = [()]
2662
else:
2663
relpaths = set()
2664
for quoted_relpath in self._transport.iter_files_recursive():
2665
path, ext = os.path.splitext(quoted_relpath)
2666
relpaths.add(path)
2667
prefixes = [self._mapper.unmap(path) for path in relpaths]
2668
self._load_prefixes(prefixes)
2669
for prefix in prefixes:
2670
for suffix in self._kndx_cache[prefix][1]:
2671
result.add(prefix + (suffix,))
2672
return result
2673
2674
def _load_prefixes(self, prefixes):
2675
"""Load the indices for prefixes."""
2676
self._check_read()
2677
for prefix in prefixes:
2678
if prefix not in self._kndx_cache:
2679
# the load_data interface writes to these variables.
2680
self._cache = {}
2681
self._history = []
2682
self._filename = prefix
2683
try:
2684
path = self._mapper.map(prefix) + '.kndx'
2685
fp = self._transport.get(path)
2686
try:
2687
# _load_data may raise NoSuchFile if the target knit is
2688
# completely empty.
2689
_load_data(self, fp)
2690
finally:
2691
fp.close()
2692
self._kndx_cache[prefix] = (self._cache, self._history)
2693
del self._cache
2694
del self._filename
2695
del self._history
2696
except NoSuchFile:
2697
self._kndx_cache[prefix] = ({}, [])
2698
if type(self._mapper) is ConstantMapper:
2699
# preserve behaviour for revisions.kndx etc.
2700
self._init_index(path)
2701
del self._cache
2702
del self._filename
2703
del self._history
2704
2705
missing_keys = _mod_index._missing_keys_from_parent_map
2706
2707
def _partition_keys(self, keys):
2708
"""Turn keys into a dict of prefix:suffix_list."""
2709
result = {}
2710
for key in keys:
2711
prefix_keys = result.setdefault(key[:-1], [])
2712
prefix_keys.append(key[-1])
2713
return result
2714
2715
def _dictionary_compress(self, keys):
2716
"""Dictionary compress keys.
2717
2718
:param keys: The keys to generate references to.
2719
:return: A string representation of keys. keys which are present are
2720
dictionary compressed, and others are emitted as fulltext with a
2721
'.' prefix.
2722
"""
2723
if not keys:
2724
return ''
1171
# got the parents ok
1172
for parent in parents:
1173
if parent not in graph:
1174
pending.add(parent)
1175
graph[version] = parents
1176
return topo_sort(graph.items())
1177
1178
def num_versions(self):
1179
return len(self._history)
1180
1181
__len__ = num_versions
1182
1183
def get_versions(self):
1184
return self._history
1185
1186
def idx_to_name(self, idx):
1187
return self._history[idx]
1188
1189
def lookup(self, version_id):
1190
assert version_id in self._cache
1191
return self._cache[version_id][5]
1192
1193
def _version_list_to_index(self, versions):
2725
1194
result_list = []
2726
prefix = keys[0][:-1]
2727
cache = self._kndx_cache[prefix][0]
2728
for key in keys:
2729
if key[:-1] != prefix:
2730
# kndx indices cannot refer across partitioned storage.
2731
raise ValueError("mismatched prefixes for %r" % keys)
2732
if key[-1] in cache:
1195
for version in versions:
1196
if version in self._cache:
2733
1197
# -- inlined lookup() --
2734
result_list.append(str(cache[key[-1]][5]))
1198
result_list.append(str(self._cache[version][5]))
2735
1199
# -- end lookup () --
2736
1200
else:
2737
result_list.append('.' + key[-1])
1201
result_list.append('.' + version.encode('utf-8'))
2738
1202
return ' '.join(result_list)
2739
1203
2740
def _reset_cache(self):
2741
# Possibly this should be a LRU cache. A dictionary from key_prefix to
2742
# (cache_dict, history_vector) for parsed kndx files.
2743
self._kndx_cache = {}
2744
self._scope = self._get_scope()
2745
allow_writes = self._allow_writes()
2746
if allow_writes:
2747
self._mode = 'w'
2748
else:
2749
self._mode = 'r'
2750
2751
def _sort_keys_by_io(self, keys, positions):
2752
"""Figure out an optimal order to read the records for the given keys.
2753
2754
Sort keys, grouped by index and sorted by position.
2755
2756
:param keys: A list of keys whose records we want to read. This will be
2757
sorted 'in-place'.
2758
:param positions: A dict, such as the one returned by
2759
_get_components_positions()
2760
:return: None
2761
"""
2762
def get_sort_key(key):
2763
index_memo = positions[key][1]
2764
# Group by prefix and position. index_memo[0] is the key, so it is
2765
# (file_id, revision_id) and we don't want to sort on revision_id,
2766
# index_memo[1] is the position, and index_memo[2] is the size,
2767
# which doesn't matter for the sort
2768
return index_memo[0][:-1], index_memo[1]
2769
return keys.sort(key=get_sort_key)
2770
2771
_get_total_build_size = _get_total_build_size
2772
2773
def _split_key(self, key):
2774
"""Split key into a prefix and suffix."""
2775
return key[:-1], key[-1]
2776
2777
2778
class _KeyRefs(object):
2779
2780
def __init__(self, track_new_keys=False):
2781
# dict mapping 'key' to 'set of keys referring to that key'
2782
self.refs = {}
2783
if track_new_keys:
2784
# set remembering all new keys
2785
self.new_keys = set()
2786
else:
2787
self.new_keys = None
2788
2789
def clear(self):
2790
if self.refs:
2791
self.refs.clear()
2792
if self.new_keys:
2793
self.new_keys.clear()
2794
2795
def add_references(self, key, refs):
2796
# Record the new references
2797
for referenced in refs:
2798
try:
2799
needed_by = self.refs[referenced]
2800
except KeyError:
2801
needed_by = self.refs[referenced] = set()
2802
needed_by.add(key)
2803
# Discard references satisfied by the new key
2804
self.add_key(key)
2805
2806
def get_new_keys(self):
2807
return self.new_keys
2808
2809
def get_unsatisfied_refs(self):
2810
return self.refs.iterkeys()
2811
2812
def _satisfy_refs_for_key(self, key):
2813
try:
2814
del self.refs[key]
2815
except KeyError:
2816
# No keys depended on this key. That's ok.
2817
pass
2818
2819
def add_key(self, key):
2820
# satisfy refs for key, and remember that we've seen this key.
2821
self._satisfy_refs_for_key(key)
2822
if self.new_keys is not None:
2823
self.new_keys.add(key)
2824
2825
def satisfy_refs_for_keys(self, keys):
2826
for key in keys:
2827
self._satisfy_refs_for_key(key)
2828
2829
def get_referrers(self):
2830
result = set()
2831
for referrers in self.refs.itervalues():
2832
result.update(referrers)
2833
return result
2834
2835
2836
class _KnitGraphIndex(object):
2837
"""A KnitVersionedFiles index layered on GraphIndex."""
2838
2839
def __init__(self, graph_index, is_locked, deltas=False, parents=True,
2840
add_callback=None, track_external_parent_refs=False):
2841
"""Construct a KnitGraphIndex on a graph_index.
2842
2843
:param graph_index: An implementation of bzrlib.index.GraphIndex.
2844
:param is_locked: A callback to check whether the object should answer
2845
queries.
2846
:param deltas: Allow delta-compressed records.
2847
:param parents: If True, record knits parents, if not do not record
2848
parents.
2849
:param add_callback: If not None, allow additions to the index and call
2850
this callback with a list of added GraphIndex nodes:
2851
[(node, value, node_refs), ...]
2852
:param is_locked: A callback, returns True if the index is locked and
2853
thus usable.
2854
:param track_external_parent_refs: If True, record all external parent
2855
references parents from added records. These can be retrieved
2856
later by calling get_missing_parents().
2857
"""
2858
self._add_callback = add_callback
2859
self._graph_index = graph_index
2860
self._deltas = deltas
2861
self._parents = parents
2862
if deltas and not parents:
2863
# XXX: TODO: Delta tree and parent graph should be conceptually
2864
# separate.
2865
raise KnitCorrupt(self, "Cannot do delta compression without "
2866
"parent tracking.")
2867
self.has_graph = parents
2868
self._is_locked = is_locked
2869
self._missing_compression_parents = set()
2870
if track_external_parent_refs:
2871
self._key_dependencies = _KeyRefs()
2872
else:
2873
self._key_dependencies = None
2874
2875
def __repr__(self):
2876
return "%s(%r)" % (self.__class__.__name__, self._graph_index)
2877
2878
def add_records(self, records, random_id=False,
2879
missing_compression_parents=False):
2880
"""Add multiple records to the index.
2881
2882
This function does not insert data into the Immutable GraphIndex
2883
backing the KnitGraphIndex, instead it prepares data for insertion by
2884
the caller and checks that it is safe to insert then calls
2885
self._add_callback with the prepared GraphIndex nodes.
2886
2887
:param records: a list of tuples:
2888
(key, options, access_memo, parents).
2889
:param random_id: If True the ids being added were randomly generated
2890
and no check for existence will be performed.
2891
:param missing_compression_parents: If True the records being added are
2892
only compressed against texts already in the index (or inside
2893
records). If False the records all refer to unavailable texts (or
2894
texts inside records) as compression parents.
2895
"""
2896
if not self._add_callback:
2897
raise errors.ReadOnlyError(self)
2898
# we hope there are no repositories with inconsistent parentage
2899
# anymore.
2900
2901
keys = {}
2902
compression_parents = set()
2903
key_dependencies = self._key_dependencies
2904
for (key, options, access_memo, parents) in records:
2905
if self._parents:
2906
parents = tuple(parents)
2907
if key_dependencies is not None:
2908
key_dependencies.add_references(key, parents)
2909
index, pos, size = access_memo
2910
if 'no-eol' in options:
2911
value = 'N'
2912
else:
2913
value = ' '
2914
value += "%d %d" % (pos, size)
2915
if not self._deltas:
2916
if 'line-delta' in options:
2917
raise KnitCorrupt(self, "attempt to add line-delta in non-delta knit")
2918
if self._parents:
2919
if self._deltas:
2920
if 'line-delta' in options:
2921
node_refs = (parents, (parents[0],))
2922
if missing_compression_parents:
2923
compression_parents.add(parents[0])
2924
else:
2925
node_refs = (parents, ())
2926
else:
2927
node_refs = (parents, )
2928
else:
2929
if parents:
2930
raise KnitCorrupt(self, "attempt to add node with parents "
2931
"in parentless index.")
2932
node_refs = ()
2933
keys[key] = (value, node_refs)
2934
# check for dups
2935
if not random_id:
2936
present_nodes = self._get_entries(keys)
2937
for (index, key, value, node_refs) in present_nodes:
2938
if (value[0] != keys[key][0][0] or
2939
node_refs[:1] != keys[key][1][:1]):
2940
raise KnitCorrupt(self, "inconsistent details in add_records"
2941
": %s %s" % ((value, node_refs), keys[key]))
2942
del keys[key]
2943
result = []
2944
if self._parents:
2945
for key, (value, node_refs) in keys.iteritems():
2946
result.append((key, value, node_refs))
2947
else:
2948
for key, (value, node_refs) in keys.iteritems():
2949
result.append((key, value))
2950
self._add_callback(result)
2951
if missing_compression_parents:
2952
# This may appear to be incorrect (it does not check for
2953
# compression parents that are in the existing graph index),
2954
# but such records won't have been buffered, so this is
2955
# actually correct: every entry when
2956
# missing_compression_parents==True either has a missing parent, or
2957
# a parent that is one of the keys in records.
2958
compression_parents.difference_update(keys)
2959
self._missing_compression_parents.update(compression_parents)
2960
# Adding records may have satisfied missing compression parents.
2961
self._missing_compression_parents.difference_update(keys)
2962
2963
def scan_unvalidated_index(self, graph_index):
2964
"""Inform this _KnitGraphIndex that there is an unvalidated index.
2965
2966
This allows this _KnitGraphIndex to keep track of any missing
2967
compression parents we may want to have filled in to make those
2968
indices valid.
2969
2970
:param graph_index: A GraphIndex
2971
"""
2972
if self._deltas:
2973
new_missing = graph_index.external_references(ref_list_num=1)
2974
new_missing.difference_update(self.get_parent_map(new_missing))
2975
self._missing_compression_parents.update(new_missing)
2976
if self._key_dependencies is not None:
2977
# Add parent refs from graph_index (and discard parent refs that
2978
# the graph_index has).
2979
for node in graph_index.iter_all_entries():
2980
self._key_dependencies.add_references(node[1], node[3][0])
2981
2982
def get_missing_compression_parents(self):
2983
"""Return the keys of missing compression parents.
2984
2985
Missing compression parents occur when a record stream was missing
2986
basis texts, or a index was scanned that had missing basis texts.
2987
"""
2988
return frozenset(self._missing_compression_parents)
2989
2990
def get_missing_parents(self):
2991
"""Return the keys of missing parents."""
2992
# If updating this, you should also update
2993
# groupcompress._GCGraphIndex.get_missing_parents
2994
# We may have false positives, so filter those out.
2995
self._key_dependencies.satisfy_refs_for_keys(
2996
self.get_parent_map(self._key_dependencies.get_unsatisfied_refs()))
2997
return frozenset(self._key_dependencies.get_unsatisfied_refs())
2998
2999
def _check_read(self):
3000
"""raise if reads are not permitted."""
3001
if not self._is_locked():
3002
raise errors.ObjectNotLocked(self)
3003
3004
def _check_write_ok(self):
3005
"""Assert if writes are not permitted."""
3006
if not self._is_locked():
3007
raise errors.ObjectNotLocked(self)
3008
3009
def _compression_parent(self, an_entry):
3010
# return the key that an_entry is compressed against, or None
3011
# Grab the second parent list (as deltas implies parents currently)
3012
compression_parents = an_entry[3][1]
3013
if not compression_parents:
3014
return None
3015
if len(compression_parents) != 1:
3016
raise AssertionError(
3017
"Too many compression parents: %r" % compression_parents)
3018
return compression_parents[0]
3019
3020
def get_build_details(self, keys):
3021
"""Get the method, index_memo and compression parent for version_ids.
3022
3023
Ghosts are omitted from the result.
3024
3025
:param keys: An iterable of keys.
3026
:return: A dict of key:
3027
(index_memo, compression_parent, parents, record_details).
3028
index_memo
3029
opaque structure to pass to read_records to extract the raw
3030
data
3031
compression_parent
3032
Content that this record is built upon, may be None
3033
parents
3034
Logical parents of this node
3035
record_details
3036
extra information about the content which needs to be passed to
3037
Factory.parse_record
3038
"""
3039
self._check_read()
3040
result = {}
3041
entries = self._get_entries(keys, False)
3042
for entry in entries:
3043
key = entry[1]
3044
if not self._parents:
3045
parents = ()
3046
else:
3047
parents = entry[3][0]
3048
if not self._deltas:
3049
compression_parent_key = None
3050
else:
3051
compression_parent_key = self._compression_parent(entry)
3052
noeol = (entry[2][0] == 'N')
3053
if compression_parent_key:
3054
method = 'line-delta'
3055
else:
3056
method = 'fulltext'
3057
result[key] = (self._node_to_position(entry),
3058
compression_parent_key, parents,
3059
(method, noeol))
3060
return result
3061
3062
def _get_entries(self, keys, check_present=False):
3063
"""Get the entries for keys.
3064
3065
:param keys: An iterable of index key tuples.
3066
"""
3067
keys = set(keys)
3068
found_keys = set()
3069
if self._parents:
3070
for node in self._graph_index.iter_entries(keys):
3071
yield node
3072
found_keys.add(node[1])
3073
else:
3074
# adapt parentless index to the rest of the code.
3075
for node in self._graph_index.iter_entries(keys):
3076
yield node[0], node[1], node[2], ()
3077
found_keys.add(node[1])
3078
if check_present:
3079
missing_keys = keys.difference(found_keys)
3080
if missing_keys:
3081
raise RevisionNotPresent(missing_keys.pop(), self)
3082
3083
def get_method(self, key):
3084
"""Return compression method of specified key."""
3085
return self._get_method(self._get_node(key))
3086
3087
def _get_method(self, node):
3088
if not self._deltas:
1204
def add_version(self, version_id, options, pos, size, parents):
1205
"""Add a version record to the index."""
1206
self.add_versions(((version_id, options, pos, size, parents),))
1207
1208
def add_versions(self, versions):
1209
"""Add multiple versions to the index.
1210
1211
:param versions: a list of tuples:
1212
(version_id, options, pos, size, parents).
1213
"""
1214
lines = []
1215
for version_id, options, pos, size, parents in versions:
1216
line = "\n%s %s %s %s %s :" % (version_id.encode('utf-8'),
1217
','.join(options),
1218
pos,
1219
size,
1220
self._version_list_to_index(parents))
1221
assert isinstance(line, str), \
1222
'content must be utf-8 encoded: %r' % (line,)
1223
lines.append(line)
1224
self._transport.append(self._filename, StringIO(''.join(lines)))
1225
# cache after writing, so that a failed write leads to missing cache
1226
# entries not extra ones. XXX TODO: RBC 20060502 in the event of a
1227
# failure, reload the index or flush it or some such, to prevent
1228
# writing records that did complete twice.
1229
for version_id, options, pos, size, parents in versions:
1230
self._cache_version(version_id, options, pos, size, parents)
1231
1232
def has_version(self, version_id):
1233
"""True if the version is in the index."""
1234
return self._cache.has_key(version_id)
1235
1236
def get_position(self, version_id):
1237
"""Return data position and size of specified version."""
1238
return (self._cache[version_id][2], \
1239
self._cache[version_id][3])
1240
1241
def get_method(self, version_id):
1242
"""Return compression method of specified version."""
1243
options = self._cache[version_id][1]
1244
if 'fulltext' in options:
3089
1245
return 'fulltext'
3090
if self._compression_parent(node):
1246
else:
1247
assert 'line-delta' in options
3091
1248
return 'line-delta'
3092
else:
3093
return 'fulltext'
3094
3095
def _get_node(self, key):
3096
try:
3097
return list(self._get_entries([key]))[0]
3098
except IndexError:
3099
raise RevisionNotPresent(key, self)
3100
3101
def get_options(self, key):
3102
"""Return a list representing options.
3103
3104
e.g. ['foo', 'bar']
3105
"""
3106
node = self._get_node(key)
3107
options = [self._get_method(node)]
3108
if node[2][0] == 'N':
3109
options.append('no-eol')
3110
return options
3111
3112
def find_ancestry(self, keys):
3113
"""See CombinedGraphIndex.find_ancestry()"""
3114
return self._graph_index.find_ancestry(keys, 0)
3115
3116
def get_parent_map(self, keys):
3117
"""Get a map of the parents of keys.
3118
3119
:param keys: The keys to look up parents for.
3120
:return: A mapping from keys to parents. Absent keys are absent from
3121
the mapping.
3122
"""
3123
self._check_read()
3124
nodes = self._get_entries(keys)
3125
result = {}
3126
if self._parents:
3127
for node in nodes:
3128
result[node[1]] = node[3][0]
3129
else:
3130
for node in nodes:
3131
result[node[1]] = None
3132
return result
3133
3134
def get_position(self, key):
3135
"""Return details needed to access the version.
3136
3137
:return: a tuple (index, data position, size) to hand to the access
3138
logic to get the record.
3139
"""
3140
node = self._get_node(key)
3141
return self._node_to_position(node)
3142
3143
has_key = _mod_index._has_key_from_parent_map
3144
3145
def keys(self):
3146
"""Get all the keys in the collection.
3147
3148
The keys are not ordered.
3149
"""
3150
self._check_read()
3151
return [node[1] for node in self._graph_index.iter_all_entries()]
3152
3153
missing_keys = _mod_index._missing_keys_from_parent_map
3154
3155
def _node_to_position(self, node):
3156
"""Convert an index value to position details."""
3157
bits = node[2][1:].split(' ')
3158
return node[0], int(bits[0]), int(bits[1])
3159
3160
def _sort_keys_by_io(self, keys, positions):
3161
"""Figure out an optimal order to read the records for the given keys.
3162
3163
Sort keys, grouped by index and sorted by position.
3164
3165
:param keys: A list of keys whose records we want to read. This will be
3166
sorted 'in-place'.
3167
:param positions: A dict, such as the one returned by
3168
_get_components_positions()
3169
:return: None
3170
"""
3171
def get_index_memo(key):
3172
# index_memo is at offset [1]. It is made up of (GraphIndex,
3173
# position, size). GI is an object, which will be unique for each
3174
# pack file. This causes us to group by pack file, then sort by
3175
# position. Size doesn't matter, but it isn't worth breaking up the
3176
# tuple.
3177
return positions[key][1]
3178
return keys.sort(key=get_index_memo)
3179
3180
_get_total_build_size = _get_total_build_size
3181
3182
3183
class _KnitKeyAccess(object):
3184
"""Access to records in .knit files."""
3185
3186
def __init__(self, transport, mapper):
3187
"""Create a _KnitKeyAccess with transport and mapper.
3188
3189
:param transport: The transport the access object is rooted at.
3190
:param mapper: The mapper used to map keys to .knit files.
3191
"""
3192
self._transport = transport
3193
self._mapper = mapper
3194
3195
def add_raw_records(self, key_sizes, raw_data):
3196
"""Add raw knit bytes to a storage area.
3197
3198
The data is spooled to the container writer in one bytes-record per
3199
raw data item.
3200
3201
:param sizes: An iterable of tuples containing the key and size of each
3202
raw data segment.
3203
:param raw_data: A bytestring containing the data.
3204
:return: A list of memos to retrieve the record later. Each memo is an
3205
opaque index memo. For _KnitKeyAccess the memo is (key, pos,
3206
length), where the key is the record key.
3207
"""
3208
if type(raw_data) is not str:
3209
raise AssertionError(
3210
'data must be plain bytes was %s' % type(raw_data))
3211
result = []
3212
offset = 0
3213
# TODO: This can be tuned for writing to sftp and other servers where
3214
# append() is relatively expensive by grouping the writes to each key
3215
# prefix.
3216
for key, size in key_sizes:
3217
path = self._mapper.map(key)
3218
try:
3219
base = self._transport.append_bytes(path + '.knit',
3220
raw_data[offset:offset+size])
3221
except errors.NoSuchFile:
3222
self._transport.mkdir(osutils.dirname(path))
3223
base = self._transport.append_bytes(path + '.knit',
3224
raw_data[offset:offset+size])
3225
# if base == 0:
3226
# chmod.
3227
offset += size
3228
result.append((key, base, size))
3229
return result
3230
3231
def flush(self):
3232
"""Flush pending writes on this access object.
3233
3234
For .knit files this is a no-op.
3235
"""
1249
1250
def get_options(self, version_id):
1251
return self._cache[version_id][1]
1252
1253
def get_parents(self, version_id):
1254
"""Return parents of specified version ignoring ghosts."""
1255
return [parent for parent in self._cache[version_id][4]
1256
if parent in self._cache]
1257
1258
def get_parents_with_ghosts(self, version_id):
1259
"""Return parents of specified version with ghosts."""
1260
return self._cache[version_id][4]
1261
1262
def check_versions_present(self, version_ids):
1263
"""Check that all specified versions are present."""
1264
version_ids = set(version_ids)
1265
for version_id in list(version_ids):
1266
if version_id in self._cache:
1267
version_ids.remove(version_id)
1268
if version_ids:
1269
raise RevisionNotPresent(list(version_ids)[0], self.filename)
1270
1271
1272
class _KnitData(_KnitComponentFile):
1273
"""Contents of the knit data file"""
1274
1275
HEADER = "# bzr knit data 8\n"
1276
1277
def __init__(self, transport, filename, mode, create=False, file_mode=None):
1278
_KnitComponentFile.__init__(self, transport, filename, mode)
1279
self._file = None
1280
self._checked = False
1281
if create:
1282
self._transport.put(self._filename, StringIO(''), mode=file_mode)
1283
1284
def clear_cache(self):
1285
"""Clear the record cache."""
3236
1286
pass
3237
1287
3238
def get_raw_records(self, memos_for_retrieval):
3239
"""Get the raw bytes for a records.
3240
3241
:param memos_for_retrieval: An iterable containing the access memo for
3242
retrieving the bytes.
3243
:return: An iterator over the bytes of the records.
3244
"""
3245
# first pass, group into same-index request to minimise readv's issued.
3246
request_lists = []
3247
current_prefix = None
3248
for (key, offset, length) in memos_for_retrieval:
3249
if current_prefix == key[:-1]:
3250
current_list.append((offset, length))
3251
else:
3252
if current_prefix is not None:
3253
request_lists.append((current_prefix, current_list))
3254
current_prefix = key[:-1]
3255
current_list = [(offset, length)]
3256
# handle the last entry
3257
if current_prefix is not None:
3258
request_lists.append((current_prefix, current_list))
3259
for prefix, read_vector in request_lists:
3260
path = self._mapper.map(prefix) + '.knit'
3261
for pos, data in self._transport.readv(path, read_vector):
3262
yield data
3263
3264
3265
class _DirectPackAccess(object):
3266
"""Access to data in one or more packs with less translation."""
3267
3268
def __init__(self, index_to_packs, reload_func=None, flush_func=None):
3269
"""Create a _DirectPackAccess object.
3270
3271
:param index_to_packs: A dict mapping index objects to the transport
3272
and file names for obtaining data.
3273
:param reload_func: A function to call if we determine that the pack
3274
files have moved and we need to reload our caches. See
3275
bzrlib.repo_fmt.pack_repo.AggregateIndex for more details.
3276
"""
3277
self._container_writer = None
3278
self._write_index = None
3279
self._indices = index_to_packs
3280
self._reload_func = reload_func
3281
self._flush_func = flush_func
3282
3283
def add_raw_records(self, key_sizes, raw_data):
3284
"""Add raw knit bytes to a storage area.
3285
3286
The data is spooled to the container writer in one bytes-record per
3287
raw data item.
3288
3289
:param sizes: An iterable of tuples containing the key and size of each
3290
raw data segment.
3291
:param raw_data: A bytestring containing the data.
3292
:return: A list of memos to retrieve the record later. Each memo is an
3293
opaque index memo. For _DirectPackAccess the memo is (index, pos,
3294
length), where the index field is the write_index object supplied
3295
to the PackAccess object.
3296
"""
3297
if type(raw_data) is not str:
3298
raise AssertionError(
3299
'data must be plain bytes was %s' % type(raw_data))
3300
result = []
3301
offset = 0
3302
for key, size in key_sizes:
3303
p_offset, p_length = self._container_writer.add_bytes_record(
3304
raw_data[offset:offset+size], [])
3305
offset += size
3306
result.append((self._write_index, p_offset, p_length))
3307
return result
3308
3309
def flush(self):
3310
"""Flush pending writes on this access object.
3311
3312
This will flush any buffered writes to a NewPack.
3313
"""
3314
if self._flush_func is not None:
3315
self._flush_func()
1288
def _open_file(self):
1289
if self._file is None:
1290
try:
1291
self._file = self._transport.get(self._filename)
1292
except NoSuchFile:
1293
pass
1294
return self._file
1295
1296
def _record_to_data(self, version_id, digest, lines):
1297
"""Convert version_id, digest, lines into a raw data block.
1298
1299
:return: (len, a StringIO instance with the raw data ready to read.)
1300
"""
1301
sio = StringIO()
1302
data_file = GzipFile(None, mode='wb', fileobj=sio)
1303
data_file.writelines(chain(
1304
["version %s %d %s\n" % (version_id.encode('utf-8'),
1305
len(lines),
1306
digest)],
1307
lines,
1308
["end %s\n" % version_id.encode('utf-8')]))
1309
data_file.close()
1310
length= sio.tell()
1311
1312
sio.seek(0)
1313
return length, sio
1314
1315
def add_raw_record(self, raw_data):
1316
"""Append a prepared record to the data file.
1317
1318
:return: the offset in the data file raw_data was written.
1319
"""
1320
assert isinstance(raw_data, str), 'data must be plain bytes'
1321
return self._transport.append(self._filename, StringIO(raw_data))
1322
1323
def add_record(self, version_id, digest, lines):
1324
"""Write new text record to disk. Returns the position in the
1325
file where it was written."""
1326
size, sio = self._record_to_data(version_id, digest, lines)
1327
# write to disk
1328
start_pos = self._transport.append(self._filename, sio)
1329
return start_pos, size
1330
1331
def _parse_record_header(self, version_id, raw_data):
1332
"""Parse a record header for consistency.
1333
1334
:return: the header and the decompressor stream.
1335
as (stream, header_record)
1336
"""
1337
df = GzipFile(mode='rb', fileobj=StringIO(raw_data))
1338
rec = df.readline().split()
1339
if len(rec) != 4:
1340
raise KnitCorrupt(self._filename, 'unexpected number of elements in record header')
1341
if rec[1].decode('utf-8')!= version_id:
1342
raise KnitCorrupt(self._filename,
1343
'unexpected version, wanted %r, got %r' % (
1344
version_id, rec[1]))
1345
return df, rec
1346
1347
def _parse_record(self, version_id, data):
1348
# profiling notes:
1349
# 4168 calls in 2880 217 internal
1350
# 4168 calls to _parse_record_header in 2121
1351
# 4168 calls to readlines in 330
1352
df, rec = self._parse_record_header(version_id, data)
1353
record_contents = df.readlines()
1354
l = record_contents.pop()
1355
assert len(record_contents) == int(rec[2])
1356
if l.decode('utf-8') != 'end %s\n' % version_id:
1357
raise KnitCorrupt(self._filename, 'unexpected version end line %r, wanted %r'
1358
% (l, version_id))
1359
df.close()
1360
return record_contents, rec[3]
1361
1362
def read_records_iter_raw(self, records):
1363
"""Read text records from data file and yield raw data.
1364
1365
This unpacks enough of the text record to validate the id is
1366
as expected but thats all.
1367
1368
It will actively recompress currently cached records on the
1369
basis that that is cheaper than I/O activity.
1370
"""
1371
# setup an iterator of the external records:
1372
# uses readv so nice and fast we hope.
1373
if len(records):
1374
# grab the disk data needed.
1375
raw_records = self._transport.readv(self._filename,
1376
[(pos, size) for version_id, pos, size in records])
1377
1378
for version_id, pos, size in records:
1379
pos, data = raw_records.next()
1380
# validate the header
1381
df, rec = self._parse_record_header(version_id, data)
1382
df.close()
1383
yield version_id, data
1384
1385
def read_records_iter(self, records):
1386
"""Read text records from data file and yield result.
1387
1388
Each passed record is a tuple of (version_id, pos, len) and
1389
will be read in the given order. Yields (version_id,
1390
contents, digest).
1391
"""
1392
if len(records) == 0:
1393
return
1394
# profiling notes:
1395
# 60890 calls for 4168 extractions in 5045, 683 internal.
1396
# 4168 calls to readv in 1411
1397
# 4168 calls to parse_record in 2880
1398
1399
# Get unique records, sorted by position
1400
needed_records = sorted(set(records), key=operator.itemgetter(1))
1401
1402
# We take it that the transport optimizes the fetching as good
1403
# as possible (ie, reads continuous ranges.)
1404
response = self._transport.readv(self._filename,
1405
[(pos, size) for version_id, pos, size in needed_records])
1406
1407
record_map = {}
1408
for (record_id, pos, size), (pos, data) in \
1409
izip(iter(needed_records), response):
1410
content, digest = self._parse_record(record_id, data)
1411
record_map[record_id] = (digest, content)
1412
1413
for version_id, pos, size in records:
1414
digest, content = record_map[version_id]
1415
yield version_id, content, digest
1416
1417
def read_records(self, records):
1418
"""Read records into a dictionary."""
1419
components = {}
1420
for record_id, content, digest in self.read_records_iter(records):
1421
components[record_id] = (content, digest)
1422
return components
1423
1424
1425
class InterKnit(InterVersionedFile):
1426
"""Optimised code paths for knit to knit operations."""
1427
1428
_matching_file_from_factory = KnitVersionedFile
1429
_matching_file_to_factory = KnitVersionedFile
1430
1431
@staticmethod
1432
def is_compatible(source, target):
1433
"""Be compatible with knits. """
1434
try:
1435
return (isinstance(source, KnitVersionedFile) and
1436
isinstance(target, KnitVersionedFile))
1437
except AttributeError:
1438
return False
1439
1440
def join(self, pb=None, msg=None, version_ids=None, ignore_missing=False):
1441
"""See InterVersionedFile.join."""
1442
assert isinstance(self.source, KnitVersionedFile)
1443
assert isinstance(self.target, KnitVersionedFile)
1444
1445
version_ids = self._get_source_version_ids(version_ids, ignore_missing)
1446
1447
if not version_ids:
1448
return 0
1449
1450
pb = bzrlib.ui.ui_factory.nested_progress_bar()
1451
try:
1452
version_ids = list(version_ids)
1453
if None in version_ids:
1454
version_ids.remove(None)
1455
1456
self.source_ancestry = set(self.source.get_ancestry(version_ids))
1457
this_versions = set(self.target._index.get_versions())
1458
needed_versions = self.source_ancestry - this_versions
1459
cross_check_versions = self.source_ancestry.intersection(this_versions)
1460
mismatched_versions = set()
1461
for version in cross_check_versions:
1462
# scan to include needed parents.
1463
n1 = set(self.target.get_parents_with_ghosts(version))
1464
n2 = set(self.source.get_parents_with_ghosts(version))
1465
if n1 != n2:
1466
# FIXME TEST this check for cycles being introduced works
1467
# the logic is we have a cycle if in our graph we are an
1468
# ancestor of any of the n2 revisions.
1469
for parent in n2:
1470
if parent in n1:
1471
# safe
1472
continue
1473
else:
1474
parent_ancestors = self.source.get_ancestry(parent)
1475
if version in parent_ancestors:
1476
raise errors.GraphCycleError([parent, version])
1477
# ensure this parent will be available later.
1478
new_parents = n2.difference(n1)
1479
needed_versions.update(new_parents.difference(this_versions))
1480
mismatched_versions.add(version)
1481
1482
if not needed_versions and not mismatched_versions:
1483
return 0
1484
full_list = topo_sort(self.source.get_graph())
1485
1486
version_list = [i for i in full_list if (not self.target.has_version(i)
1487
and i in needed_versions)]
1488
1489
# plan the join:
1490
copy_queue = []
1491
copy_queue_records = []
1492
copy_set = set()
1493
for version_id in version_list:
1494
options = self.source._index.get_options(version_id)
1495
parents = self.source._index.get_parents_with_ghosts(version_id)
1496
# check that its will be a consistent copy:
1497
for parent in parents:
1498
# if source has the parent, we must :
1499
# * already have it or
1500
# * have it scheduled already
1501
# otherwise we don't care
1502
assert (self.target.has_version(parent) or
1503
parent in copy_set or
1504
not self.source.has_version(parent))
1505
data_pos, data_size = self.source._index.get_position(version_id)
1506
copy_queue_records.append((version_id, data_pos, data_size))
1507
copy_queue.append((version_id, options, parents))
1508
copy_set.add(version_id)
1509
1510
# data suck the join:
1511
count = 0
1512
total = len(version_list)
1513
raw_datum = []
1514
raw_records = []
1515
for (version_id, raw_data), \
1516
(version_id2, options, parents) in \
1517
izip(self.source._data.read_records_iter_raw(copy_queue_records),
1518
copy_queue):
1519
assert version_id == version_id2, 'logic error, inconsistent results'
1520
count = count + 1
1521
pb.update("Joining knit", count, total)
1522
raw_records.append((version_id, options, parents, len(raw_data)))
1523
raw_datum.append(raw_data)
1524
self.target._add_raw_records(raw_records, ''.join(raw_datum))
1525
1526
for version in mismatched_versions:
1527
# FIXME RBC 20060309 is this needed?
1528
n1 = set(self.target.get_parents_with_ghosts(version))
1529
n2 = set(self.source.get_parents_with_ghosts(version))
1530
# write a combined record to our history preserving the current
1531
# parents as first in the list
1532
new_parents = self.target.get_parents_with_ghosts(version) + list(n2.difference(n1))
1533
self.target.fix_parents(version, new_parents)
1534
return count
1535
finally:
1536
pb.finished()
1537
1538
1539
InterVersionedFile.register_optimiser(InterKnit)
1540
1541
1542
class WeaveToKnit(InterVersionedFile):
1543
"""Optimised code paths for weave to knit operations."""
1544
1545
_matching_file_from_factory = bzrlib.weave.WeaveFile
1546
_matching_file_to_factory = KnitVersionedFile
1547
1548
@staticmethod
1549
def is_compatible(source, target):
1550
"""Be compatible with weaves to knits."""
1551
try:
1552
return (isinstance(source, bzrlib.weave.Weave) and
1553
isinstance(target, KnitVersionedFile))
1554
except AttributeError:
1555
return False
1556
1557
def join(self, pb=None, msg=None, version_ids=None, ignore_missing=False):
1558
"""See InterVersionedFile.join."""
1559
assert isinstance(self.source, bzrlib.weave.Weave)
1560
assert isinstance(self.target, KnitVersionedFile)
1561
1562
version_ids = self._get_source_version_ids(version_ids, ignore_missing)
1563
1564
if not version_ids:
1565
return 0
1566
1567
pb = bzrlib.ui.ui_factory.nested_progress_bar()
1568
try:
1569
version_ids = list(version_ids)
1570
1571
self.source_ancestry = set(self.source.get_ancestry(version_ids))
1572
this_versions = set(self.target._index.get_versions())
1573
needed_versions = self.source_ancestry - this_versions
1574
cross_check_versions = self.source_ancestry.intersection(this_versions)
1575
mismatched_versions = set()
1576
for version in cross_check_versions:
1577
# scan to include needed parents.
1578
n1 = set(self.target.get_parents_with_ghosts(version))
1579
n2 = set(self.source.get_parents(version))
1580
# if all of n2's parents are in n1, then its fine.
1581
if n2.difference(n1):
1582
# FIXME TEST this check for cycles being introduced works
1583
# the logic is we have a cycle if in our graph we are an
1584
# ancestor of any of the n2 revisions.
1585
for parent in n2:
1586
if parent in n1:
1587
# safe
1588
continue
1589
else:
1590
parent_ancestors = self.source.get_ancestry(parent)
1591
if version in parent_ancestors:
1592
raise errors.GraphCycleError([parent, version])
1593
# ensure this parent will be available later.
1594
new_parents = n2.difference(n1)
1595
needed_versions.update(new_parents.difference(this_versions))
1596
mismatched_versions.add(version)
1597
1598
if not needed_versions and not mismatched_versions:
1599
return 0
1600
full_list = topo_sort(self.source.get_graph())
1601
1602
version_list = [i for i in full_list if (not self.target.has_version(i)
1603
and i in needed_versions)]
1604
1605
# do the join:
1606
count = 0
1607
total = len(version_list)
1608
for version_id in version_list:
1609
pb.update("Converting to knit", count, total)
1610
parents = self.source.get_parents(version_id)
1611
# check that its will be a consistent copy:
1612
for parent in parents:
1613
# if source has the parent, we must already have it
1614
assert (self.target.has_version(parent))
1615
self.target.add_lines(
1616
version_id, parents, self.source.get_lines(version_id))
1617
count = count + 1
1618
1619
for version in mismatched_versions:
1620
# FIXME RBC 20060309 is this needed?
1621
n1 = set(self.target.get_parents_with_ghosts(version))
1622
n2 = set(self.source.get_parents(version))
1623
# write a combined record to our history preserving the current
1624
# parents as first in the list
1625
new_parents = self.target.get_parents_with_ghosts(version) + list(n2.difference(n1))
1626
self.target.fix_parents(version, new_parents)
1627
return count
1628
finally:
1629
pb.finished()
1630
1631
1632
InterVersionedFile.register_optimiser(WeaveToKnit)
1633
1634
1635
class KnitSequenceMatcher(difflib.SequenceMatcher):
1636
"""Knit tuned sequence matcher.
1637
1638
This is based on profiling of difflib which indicated some improvements
1639
for our usage pattern.
1640
"""
1641
1642
def find_longest_match(self, alo, ahi, blo, bhi):
1643
"""Find longest matching block in a[alo:ahi] and b[blo:bhi].
1644
1645
If isjunk is not defined:
1646
1647
Return (i,j,k) such that a[i:i+k] is equal to b[j:j+k], where
1648
alo <= i <= i+k <= ahi
1649
blo <= j <= j+k <= bhi
1650
and for all (i',j',k') meeting those conditions,
1651
k >= k'
1652
i <= i'
1653
and if i == i', j <= j'
1654
1655
In other words, of all maximal matching blocks, return one that
1656
starts earliest in a, and of all those maximal matching blocks that
1657
start earliest in a, return the one that starts earliest in b.
1658
1659
>>> s = SequenceMatcher(None, " abcd", "abcd abcd")
1660
>>> s.find_longest_match(0, 5, 0, 9)
1661
(0, 4, 5)
1662
1663
If isjunk is defined, first the longest matching block is
1664
determined as above, but with the additional restriction that no
1665
junk element appears in the block. Then that block is extended as
1666
far as possible by matching (only) junk elements on both sides. So
1667
the resulting block never matches on junk except as identical junk
1668
happens to be adjacent to an "interesting" match.
1669
1670
Here's the same example as before, but considering blanks to be
1671
junk. That prevents " abcd" from matching the " abcd" at the tail
1672
end of the second sequence directly. Instead only the "abcd" can
1673
match, and matches the leftmost "abcd" in the second sequence:
1674
1675
>>> s = SequenceMatcher(lambda x: x==" ", " abcd", "abcd abcd")
1676
>>> s.find_longest_match(0, 5, 0, 9)
1677
(1, 0, 4)
1678
1679
If no blocks match, return (alo, blo, 0).
1680
1681
>>> s = SequenceMatcher(None, "ab", "c")
1682
>>> s.find_longest_match(0, 2, 0, 1)
1683
(0, 0, 0)
1684
"""
1685
1686
# CAUTION: stripping common prefix or suffix would be incorrect.
1687
# E.g.,
1688
# ab
1689
# acab
1690
# Longest matching block is "ab", but if common prefix is
1691
# stripped, it's "a" (tied with "b"). UNIX(tm) diff does so
1692
# strip, so ends up claiming that ab is changed to acab by
1693
# inserting "ca" in the middle. That's minimal but unintuitive:
1694
# "it's obvious" that someone inserted "ac" at the front.
1695
# Windiff ends up at the same place as diff, but by pairing up
1696
# the unique 'b's and then matching the first two 'a's.
1697
1698
a, b, b2j, isbjunk = self.a, self.b, self.b2j, self.isbjunk
1699
besti, bestj, bestsize = alo, blo, 0
1700
# find longest junk-free match
1701
# during an iteration of the loop, j2len[j] = length of longest
1702
# junk-free match ending with a[i-1] and b[j]
1703
j2len = {}
1704
# nothing = []
1705
b2jget = b2j.get
1706
for i in xrange(alo, ahi):
1707
# look at all instances of a[i] in b; note that because
1708
# b2j has no junk keys, the loop is skipped if a[i] is junk
1709
j2lenget = j2len.get
1710
newj2len = {}
3316
1711
3317
def get_raw_records(self, memos_for_retrieval):
3318
"""Get the raw bytes for a records.
1712
# changing b2j.get(a[i], nothing) to a try:KeyError pair produced the
1713
# following improvement
1714
# 704 0 4650.5320 2620.7410 bzrlib.knit:1336(find_longest_match)
1715
# +326674 0 1655.1210 1655.1210 +<method 'get' of 'dict' objects>
1716
# +76519 0 374.6700 374.6700 +<method 'has_key' of 'dict' objects>
1717
# to
1718
# 704 0 3733.2820 2209.6520 bzrlib.knit:1336(find_longest_match)
1719
# +211400 0 1147.3520 1147.3520 +<method 'get' of 'dict' objects>
1720
# +76519 0 376.2780 376.2780 +<method 'has_key' of 'dict' objects>
3319
1721
3320
:param memos_for_retrieval: An iterable containing the (index, pos,
3321
length) memo for retrieving the bytes. The Pack access method
3322
looks up the pack to use for a given record in its index_to_pack
3323
map.
3324
:return: An iterator over the bytes of the records.
3325
"""
3326
# first pass, group into same-index requests
3327
request_lists = []
3328
current_index = None
3329
for (index, offset, length) in memos_for_retrieval:
3330
if current_index == index:
3331
current_list.append((offset, length))
3332
else:
3333
if current_index is not None:
3334
request_lists.append((current_index, current_list))
3335
current_index = index
3336
current_list = [(offset, length)]
3337
# handle the last entry
3338
if current_index is not None:
3339
request_lists.append((current_index, current_list))
3340
for index, offsets in request_lists:
3341
1722
try:
3342
transport, path = self._indices[index]
1723
js = b2j[a[i]]
3343
1724
except KeyError:
3344
# A KeyError here indicates that someone has triggered an index
3345
# reload, and this index has gone missing, we need to start
3346
# over.
3347
if self._reload_func is None:
3348
# If we don't have a _reload_func there is nothing that can
3349
# be done
3350
raise
3351
raise errors.RetryWithNewPacks(index,
3352
reload_occurred=True,
3353
exc_info=sys.exc_info())
3354
try:
3355
reader = pack.make_readv_reader(transport, path, offsets)
3356
for names, read_func in reader.iter_records():
3357
yield read_func(None)
3358
except errors.NoSuchFile:
3359
# A NoSuchFile error indicates that a pack file has gone
3360
# missing on disk, we need to trigger a reload, and start over.
3361
if self._reload_func is None:
3362
raise
3363
raise errors.RetryWithNewPacks(transport.abspath(path),
3364
reload_occurred=False,
3365
exc_info=sys.exc_info())
3366
3367
def set_writer(self, writer, index, transport_packname):
3368
"""Set a writer to use for adding data."""
3369
if index is not None:
3370
self._indices[index] = transport_packname
3371
self._container_writer = writer
3372
self._write_index = index
3373
3374
def reload_or_raise(self, retry_exc):
3375
"""Try calling the reload function, or re-raise the original exception.
3376
3377
This should be called after _DirectPackAccess raises a
3378
RetryWithNewPacks exception. This function will handle the common logic
3379
of determining when the error is fatal versus being temporary.
3380
It will also make sure that the original exception is raised, rather
3381
than the RetryWithNewPacks exception.
3382
3383
If this function returns, then the calling function should retry
3384
whatever operation was being performed. Otherwise an exception will
3385
be raised.
3386
3387
:param retry_exc: A RetryWithNewPacks exception.
3388
"""
3389
is_error = False
3390
if self._reload_func is None:
3391
is_error = True
3392
elif not self._reload_func():
3393
# The reload claimed that nothing changed
3394
if not retry_exc.reload_occurred:
3395
# If there wasn't an earlier reload, then we really were
3396
# expecting to find changes. We didn't find them, so this is a
3397
# hard error
3398
is_error = True
3399
if is_error:
3400
exc_class, exc_value, exc_traceback = retry_exc.exc_info
3401
raise exc_class, exc_value, exc_traceback
3402
3403
3404
# Deprecated, use PatienceSequenceMatcher instead
3405
KnitSequenceMatcher = patiencediff.PatienceSequenceMatcher
3406
3407
3408
def annotate_knit(knit, revision_id):
3409
"""Annotate a knit with no cached annotations.
3410
3411
This implementation is for knits with no cached annotations.
3412
It will work for knits with cached annotations, but this is not
3413
recommended.
3414
"""
3415
annotator = _KnitAnnotator(knit)
3416
return iter(annotator.annotate_flat(revision_id))
3417
3418
3419
class _KnitAnnotator(annotate.Annotator):
3420
"""Build up the annotations for a text."""
3421
3422
def __init__(self, vf):
3423
annotate.Annotator.__init__(self, vf)
3424
3425
# TODO: handle Nodes which cannot be extracted
3426
# self._ghosts = set()
3427
3428
# Map from (key, parent_key) => matching_blocks, should be 'use once'
3429
self._matching_blocks = {}
3430
3431
# KnitContent objects
3432
self._content_objects = {}
3433
# The number of children that depend on this fulltext content object
3434
self._num_compression_children = {}
3435
# Delta records that need their compression parent before they can be
3436
# expanded
3437
self._pending_deltas = {}
3438
# Fulltext records that are waiting for their parents fulltexts before
3439
# they can be yielded for annotation
3440
self._pending_annotation = {}
3441
3442
self._all_build_details = {}
3443
3444
def _get_build_graph(self, key):
3445
"""Get the graphs for building texts and annotations.
3446
3447
The data you need for creating a full text may be different than the
3448
data you need to annotate that text. (At a minimum, you need both
3449
parents to create an annotation, but only need 1 parent to generate the
3450
fulltext.)
3451
3452
:return: A list of (key, index_memo) records, suitable for
3453
passing to read_records_iter to start reading in the raw data from
3454
the pack file.
3455
"""
3456
pending = set([key])
3457
records = []
3458
ann_keys = set()
3459
self._num_needed_children[key] = 1
3460
while pending:
3461
# get all pending nodes
3462
this_iteration = pending
3463
build_details = self._vf._index.get_build_details(this_iteration)
3464
self._all_build_details.update(build_details)
3465
# new_nodes = self._vf._index._get_entries(this_iteration)
3466
pending = set()
3467
for key, details in build_details.iteritems():
3468
(index_memo, compression_parent, parent_keys,
3469
record_details) = details
3470
self._parent_map[key] = parent_keys
3471
self._heads_provider = None
3472
records.append((key, index_memo))
3473
# Do we actually need to check _annotated_lines?
3474
pending.update([p for p in parent_keys
3475
if p not in self._all_build_details])
3476
if parent_keys:
3477
for parent_key in parent_keys:
3478
if parent_key in self._num_needed_children:
3479
self._num_needed_children[parent_key] += 1
3480
else:
3481
self._num_needed_children[parent_key] = 1
3482
if compression_parent:
3483
if compression_parent in self._num_compression_children:
3484
self._num_compression_children[compression_parent] += 1
3485
else:
3486
self._num_compression_children[compression_parent] = 1
3487
3488
missing_versions = this_iteration.difference(build_details.keys())
3489
if missing_versions:
3490
for key in missing_versions:
3491
if key in self._parent_map and key in self._text_cache:
3492
# We already have this text ready, we just need to
3493
# yield it later so we get it annotated
3494
ann_keys.add(key)
3495
parent_keys = self._parent_map[key]
3496
for parent_key in parent_keys:
3497
if parent_key in self._num_needed_children:
3498
self._num_needed_children[parent_key] += 1
3499
else:
3500
self._num_needed_children[parent_key] = 1
3501
pending.update([p for p in parent_keys
3502
if p not in self._all_build_details])
3503
else:
3504
raise errors.RevisionNotPresent(key, self._vf)
3505
# Generally we will want to read the records in reverse order, because
3506
# we find the parent nodes after the children
3507
records.reverse()
3508
return records, ann_keys
3509
3510
def _get_needed_texts(self, key, pb=None):
3511
# if True or len(self._vf._fallback_vfs) > 0:
3512
if len(self._vf._fallback_vfs) > 0:
3513
# If we have fallbacks, go to the generic path
3514
for v in annotate.Annotator._get_needed_texts(self, key, pb=pb):
3515
yield v
3516
return
3517
while True:
3518
try:
3519
records, ann_keys = self._get_build_graph(key)
3520
for idx, (sub_key, text, num_lines) in enumerate(
3521
self._extract_texts(records)):
3522
if pb is not None:
3523
pb.update('annotating', idx, len(records))
3524
yield sub_key, text, num_lines
3525
for sub_key in ann_keys:
3526
text = self._text_cache[sub_key]
3527
num_lines = len(text) # bad assumption
3528
yield sub_key, text, num_lines
3529
return
3530
except errors.RetryWithNewPacks, e:
3531
self._vf._access.reload_or_raise(e)
3532
# The cached build_details are no longer valid
3533
self._all_build_details.clear()
3534
3535
def _cache_delta_blocks(self, key, compression_parent, delta, lines):
3536
parent_lines = self._text_cache[compression_parent]
3537
blocks = list(KnitContent.get_line_delta_blocks(delta, parent_lines, lines))
3538
self._matching_blocks[(key, compression_parent)] = blocks
3539
3540
def _expand_record(self, key, parent_keys, compression_parent, record,
3541
record_details):
3542
delta = None
3543
if compression_parent:
3544
if compression_parent not in self._content_objects:
3545
# Waiting for the parent
3546
self._pending_deltas.setdefault(compression_parent, []).append(
3547
(key, parent_keys, record, record_details))
3548
return None
3549
# We have the basis parent, so expand the delta
3550
num = self._num_compression_children[compression_parent]
3551
num -= 1
3552
if num == 0:
3553
base_content = self._content_objects.pop(compression_parent)
3554
self._num_compression_children.pop(compression_parent)
1725
pass
3555
1726
else:
3556
self._num_compression_children[compression_parent] = num
3557
base_content = self._content_objects[compression_parent]
3558
# It is tempting to want to copy_base_content=False for the last
3559
# child object. However, whenever noeol=False,
3560
# self._text_cache[parent_key] is content._lines. So mutating it
3561
# gives very bad results.
3562
# The alternative is to copy the lines into text cache, but then we
3563
# are copying anyway, so just do it here.
3564
content, delta = self._vf._factory.parse_record(
3565
key, record, record_details, base_content,
3566
copy_base_content=True)
3567
else:
3568
# Fulltext record
3569
content, _ = self._vf._factory.parse_record(
3570
key, record, record_details, None)
3571
if self._num_compression_children.get(key, 0) > 0:
3572
self._content_objects[key] = content
3573
lines = content.text()
3574
self._text_cache[key] = lines
3575
if delta is not None:
3576
self._cache_delta_blocks(key, compression_parent, delta, lines)
3577
return lines
3578
3579
def _get_parent_annotations_and_matches(self, key, text, parent_key):
3580
"""Get the list of annotations for the parent, and the matching lines.
3581
3582
:param text: The opaque value given by _get_needed_texts
3583
:param parent_key: The key for the parent text
3584
:return: (parent_annotations, matching_blocks)
3585
parent_annotations is a list as long as the number of lines in
3586
parent
3587
matching_blocks is a list of (parent_idx, text_idx, len) tuples
3588
indicating which lines match between the two texts
3589
"""
3590
block_key = (key, parent_key)
3591
if block_key in self._matching_blocks:
3592
blocks = self._matching_blocks.pop(block_key)
3593
parent_annotations = self._annotations_cache[parent_key]
3594
return parent_annotations, blocks
3595
return annotate.Annotator._get_parent_annotations_and_matches(self,
3596
key, text, parent_key)
3597
3598
def _process_pending(self, key):
3599
"""The content for 'key' was just processed.
3600
3601
Determine if there is any more pending work to be processed.
3602
"""
3603
to_return = []
3604
if key in self._pending_deltas:
3605
compression_parent = key
3606
children = self._pending_deltas.pop(key)
3607
for child_key, parent_keys, record, record_details in children:
3608
lines = self._expand_record(child_key, parent_keys,
3609
compression_parent,
3610
record, record_details)
3611
if self._check_ready_for_annotations(child_key, parent_keys):
3612
to_return.append(child_key)
3613
# Also check any children that are waiting for this parent to be
3614
# annotation ready
3615
if key in self._pending_annotation:
3616
children = self._pending_annotation.pop(key)
3617
to_return.extend([c for c, p_keys in children
3618
if self._check_ready_for_annotations(c, p_keys)])
3619
return to_return
3620
3621
def _check_ready_for_annotations(self, key, parent_keys):
3622
"""return true if this text is ready to be yielded.
3623
3624
Otherwise, this will return False, and queue the text into
3625
self._pending_annotation
3626
"""
3627
for parent_key in parent_keys:
3628
if parent_key not in self._annotations_cache:
3629
# still waiting on at least one parent text, so queue it up
3630
# Note that if there are multiple parents, we need to wait
3631
# for all of them.
3632
self._pending_annotation.setdefault(parent_key,
3633
[]).append((key, parent_keys))
3634
return False
3635
return True
3636
3637
def _extract_texts(self, records):
3638
"""Extract the various texts needed based on records"""
3639
# We iterate in the order read, rather than a strict order requested
3640
# However, process what we can, and put off to the side things that
3641
# still need parents, cleaning them up when those parents are
3642
# processed.
3643
# Basic data flow:
3644
# 1) As 'records' are read, see if we can expand these records into
3645
# Content objects (and thus lines)
3646
# 2) If a given line-delta is waiting on its compression parent, it
3647
# gets queued up into self._pending_deltas, otherwise we expand
3648
# it, and put it into self._text_cache and self._content_objects
3649
# 3) If we expanded the text, we will then check to see if all
3650
# parents have also been processed. If so, this text gets yielded,
3651
# else this record gets set aside into pending_annotation
3652
# 4) Further, if we expanded the text in (2), we will then check to
3653
# see if there are any children in self._pending_deltas waiting to
3654
# also be processed. If so, we go back to (2) for those
3655
# 5) Further again, if we yielded the text, we can then check if that
3656
# 'unlocks' any of the texts in pending_annotations, which should
3657
# then get yielded as well
3658
# Note that both steps 4 and 5 are 'recursive' in that unlocking one
3659
# compression child could unlock yet another, and yielding a fulltext
3660
# will also 'unlock' the children that are waiting on that annotation.
3661
# (Though also, unlocking 1 parent's fulltext, does not unlock a child
3662
# if other parents are also waiting.)
3663
# We want to yield content before expanding child content objects, so
3664
# that we know when we can re-use the content lines, and the annotation
3665
# code can know when it can stop caching fulltexts, as well.
3666
3667
# Children that are missing their compression parent
3668
pending_deltas = {}
3669
for (key, record, digest) in self._vf._read_records_iter(records):
3670
# ghosts?
3671
details = self._all_build_details[key]
3672
(_, compression_parent, parent_keys, record_details) = details
3673
lines = self._expand_record(key, parent_keys, compression_parent,
3674
record, record_details)
3675
if lines is None:
3676
# Pending delta should be queued up
3677
continue
3678
# At this point, we may be able to yield this content, if all
3679
# parents are also finished
3680
yield_this_text = self._check_ready_for_annotations(key,
3681
parent_keys)
3682
if yield_this_text:
3683
# All parents present
3684
yield key, lines, len(lines)
3685
to_process = self._process_pending(key)
3686
while to_process:
3687
this_process = to_process
3688
to_process = []
3689
for key in this_process:
3690
lines = self._text_cache[key]
3691
yield key, lines, len(lines)
3692
to_process.extend(self._process_pending(key))
3693
3694
try:
3695
from bzrlib._knit_load_data_pyx import _load_data_c as _load_data
3696
except ImportError:
3697
from bzrlib._knit_load_data_py import _load_data_py as _load_data
1727
for j in js:
1728
# a[i] matches b[j]
1729
if j >= blo:
1730
if j >= bhi:
1731
break
1732
k = newj2len[j] = 1 + j2lenget(-1 + j, 0)
1733
if k > bestsize:
1734
besti, bestj, bestsize = 1 + i-k, 1 + j-k, k
1735
j2len = newj2len
1736
1737
# Extend the best by non-junk elements on each end. In particular,
1738
# "popular" non-junk elements aren't in b2j, which greatly speeds
1739
# the inner loop above, but also means "the best" match so far
1740
# doesn't contain any junk *or* popular non-junk elements.
1741
while besti > alo and bestj > blo and \
1742
not isbjunk(b[bestj-1]) and \
1743
a[besti-1] == b[bestj-1]:
1744
besti, bestj, bestsize = besti-1, bestj-1, bestsize+1
1745
while besti+bestsize < ahi and bestj+bestsize < bhi and \
1746
not isbjunk(b[bestj+bestsize]) and \
1747
a[besti+bestsize] == b[bestj+bestsize]:
1748
bestsize += 1
1749
1750
# Now that we have a wholly interesting match (albeit possibly
1751
# empty!), we may as well suck up the matching junk on each
1752
# side of it too. Can't think of a good reason not to, and it
1753
# saves post-processing the (possibly considerable) expense of
1754
# figuring out what to do with it. In the case of an empty
1755
# interesting match, this is clearly the right thing to do,
1756
# because no other kind of match is possible in the regions.
1757
while besti > alo and bestj > blo and \
1758
isbjunk(b[bestj-1]) and \
1759
a[besti-1] == b[bestj-1]:
1760
besti, bestj, bestsize = besti-1, bestj-1, bestsize+1
1761
while besti+bestsize < ahi and bestj+bestsize < bhi and \
1762
isbjunk(b[bestj+bestsize]) and \
1763
a[besti+bestsize] == b[bestj+bestsize]:
1764
bestsize = bestsize + 1
1765
1766
return besti, bestj, bestsize
1767
Loggerhead is a web-based interface for Breezy
Version: 2.0.1