##// END OF EJS Templates
i18n: don't translate a transaction name
Bryan O'Sullivan -
r27794:9a1f3f9b default
parent child Browse files
Show More
@@ -1,387 +1,387 b''
1 1 # streamclone.py - producing and consuming streaming repository data
2 2 #
3 3 # Copyright 2015 Gregory Szorc <gregory.szorc@gmail.com>
4 4 #
5 5 # This software may be used and distributed according to the terms of the
6 6 # GNU General Public License version 2 or any later version.
7 7
8 8 from __future__ import absolute_import
9 9
10 10 import struct
11 11 import time
12 12
13 13 from .i18n import _
14 14 from . import (
15 15 branchmap,
16 16 error,
17 17 store,
18 18 util,
19 19 )
20 20
21 21 def canperformstreamclone(pullop, bailifbundle2supported=False):
22 22 """Whether it is possible to perform a streaming clone as part of pull.
23 23
24 24 ``bailifbundle2supported`` will cause the function to return False if
25 25 bundle2 stream clones are supported. It should only be called by the
26 26 legacy stream clone code path.
27 27
28 28 Returns a tuple of (supported, requirements). ``supported`` is True if
29 29 streaming clone is supported and False otherwise. ``requirements`` is
30 30 a set of repo requirements from the remote, or ``None`` if stream clone
31 31 isn't supported.
32 32 """
33 33 repo = pullop.repo
34 34 remote = pullop.remote
35 35
36 36 bundle2supported = False
37 37 if pullop.canusebundle2:
38 38 if 'v1' in pullop.remotebundle2caps.get('stream', []):
39 39 bundle2supported = True
40 40 # else
41 41 # Server doesn't support bundle2 stream clone or doesn't support
42 42 # the versions we support. Fall back and possibly allow legacy.
43 43
44 44 # Ensures legacy code path uses available bundle2.
45 45 if bailifbundle2supported and bundle2supported:
46 46 return False, None
47 47 # Ensures bundle2 doesn't try to do a stream clone if it isn't supported.
48 48 #elif not bailifbundle2supported and not bundle2supported:
49 49 # return False, None
50 50
51 51 # Streaming clone only works on empty repositories.
52 52 if len(repo):
53 53 return False, None
54 54
55 55 # Streaming clone only works if all data is being requested.
56 56 if pullop.heads:
57 57 return False, None
58 58
59 59 streamrequested = pullop.streamclonerequested
60 60
61 61 # If we don't have a preference, let the server decide for us. This
62 62 # likely only comes into play in LANs.
63 63 if streamrequested is None:
64 64 # The server can advertise whether to prefer streaming clone.
65 65 streamrequested = remote.capable('stream-preferred')
66 66
67 67 if not streamrequested:
68 68 return False, None
69 69
70 70 # In order for stream clone to work, the client has to support all the
71 71 # requirements advertised by the server.
72 72 #
73 73 # The server advertises its requirements via the "stream" and "streamreqs"
74 74 # capability. "stream" (a value-less capability) is advertised if and only
75 75 # if the only requirement is "revlogv1." Else, the "streamreqs" capability
76 76 # is advertised and contains a comma-delimited list of requirements.
77 77 requirements = set()
78 78 if remote.capable('stream'):
79 79 requirements.add('revlogv1')
80 80 else:
81 81 streamreqs = remote.capable('streamreqs')
82 82 # This is weird and shouldn't happen with modern servers.
83 83 if not streamreqs:
84 84 return False, None
85 85
86 86 streamreqs = set(streamreqs.split(','))
87 87 # Server requires something we don't support. Bail.
88 88 if streamreqs - repo.supportedformats:
89 89 return False, None
90 90 requirements = streamreqs
91 91
92 92 return True, requirements
93 93
94 94 def maybeperformlegacystreamclone(pullop):
95 95 """Possibly perform a legacy stream clone operation.
96 96
97 97 Legacy stream clones are performed as part of pull but before all other
98 98 operations.
99 99
100 100 A legacy stream clone will not be performed if a bundle2 stream clone is
101 101 supported.
102 102 """
103 103 supported, requirements = canperformstreamclone(pullop)
104 104
105 105 if not supported:
106 106 return
107 107
108 108 repo = pullop.repo
109 109 remote = pullop.remote
110 110
111 111 # Save remote branchmap. We will use it later to speed up branchcache
112 112 # creation.
113 113 rbranchmap = None
114 114 if remote.capable('branchmap'):
115 115 rbranchmap = remote.branchmap()
116 116
117 117 repo.ui.status(_('streaming all changes\n'))
118 118
119 119 fp = remote.stream_out()
120 120 l = fp.readline()
121 121 try:
122 122 resp = int(l)
123 123 except ValueError:
124 124 raise error.ResponseError(
125 125 _('unexpected response from remote server:'), l)
126 126 if resp == 1:
127 127 raise error.Abort(_('operation forbidden by server'))
128 128 elif resp == 2:
129 129 raise error.Abort(_('locking the remote repository failed'))
130 130 elif resp != 0:
131 131 raise error.Abort(_('the server sent an unknown error code'))
132 132
133 133 l = fp.readline()
134 134 try:
135 135 filecount, bytecount = map(int, l.split(' ', 1))
136 136 except (ValueError, TypeError):
137 137 raise error.ResponseError(
138 138 _('unexpected response from remote server:'), l)
139 139
140 140 lock = repo.lock()
141 141 try:
142 142 consumev1(repo, fp, filecount, bytecount)
143 143
144 144 # new requirements = old non-format requirements +
145 145 # new format-related remote requirements
146 146 # requirements from the streamed-in repository
147 147 repo.requirements = requirements | (
148 148 repo.requirements - repo.supportedformats)
149 149 repo._applyopenerreqs()
150 150 repo._writerequirements()
151 151
152 152 if rbranchmap:
153 153 branchmap.replacecache(repo, rbranchmap)
154 154
155 155 repo.invalidate()
156 156 finally:
157 157 lock.release()
158 158
159 159 def allowservergeneration(ui):
160 160 """Whether streaming clones are allowed from the server."""
161 161 return ui.configbool('server', 'uncompressed', True, untrusted=True)
162 162
163 163 # This is it's own function so extensions can override it.
164 164 def _walkstreamfiles(repo):
165 165 return repo.store.walk()
166 166
167 167 def generatev1(repo):
168 168 """Emit content for version 1 of a streaming clone.
169 169
170 170 This returns a 3-tuple of (file count, byte size, data iterator).
171 171
172 172 The data iterator consists of N entries for each file being transferred.
173 173 Each file entry starts as a line with the file name and integer size
174 174 delimited by a null byte.
175 175
176 176 The raw file data follows. Following the raw file data is the next file
177 177 entry, or EOF.
178 178
179 179 When used on the wire protocol, an additional line indicating protocol
180 180 success will be prepended to the stream. This function is not responsible
181 181 for adding it.
182 182
183 183 This function will obtain a repository lock to ensure a consistent view of
184 184 the store is captured. It therefore may raise LockError.
185 185 """
186 186 entries = []
187 187 total_bytes = 0
188 188 # Get consistent snapshot of repo, lock during scan.
189 189 lock = repo.lock()
190 190 try:
191 191 repo.ui.debug('scanning\n')
192 192 for name, ename, size in _walkstreamfiles(repo):
193 193 if size:
194 194 entries.append((name, size))
195 195 total_bytes += size
196 196 finally:
197 197 lock.release()
198 198
199 199 repo.ui.debug('%d files, %d bytes to transfer\n' %
200 200 (len(entries), total_bytes))
201 201
202 202 svfs = repo.svfs
203 203 oldaudit = svfs.mustaudit
204 204 debugflag = repo.ui.debugflag
205 205 svfs.mustaudit = False
206 206
207 207 def emitrevlogdata():
208 208 try:
209 209 for name, size in entries:
210 210 if debugflag:
211 211 repo.ui.debug('sending %s (%d bytes)\n' % (name, size))
212 212 # partially encode name over the wire for backwards compat
213 213 yield '%s\0%d\n' % (store.encodedir(name), size)
214 214 if size <= 65536:
215 215 yield svfs.read(name)
216 216 else:
217 217 for chunk in util.filechunkiter(svfs(name), limit=size):
218 218 yield chunk
219 219 finally:
220 220 svfs.mustaudit = oldaudit
221 221
222 222 return len(entries), total_bytes, emitrevlogdata()
223 223
224 224 def generatev1wireproto(repo):
225 225 """Emit content for version 1 of streaming clone suitable for the wire.
226 226
227 227 This is the data output from ``generatev1()`` with a header line
228 228 indicating file count and byte size.
229 229 """
230 230 filecount, bytecount, it = generatev1(repo)
231 231 yield '%d %d\n' % (filecount, bytecount)
232 232 for chunk in it:
233 233 yield chunk
234 234
235 235 def generatebundlev1(repo, compression='UN'):
236 236 """Emit content for version 1 of a stream clone bundle.
237 237
238 238 The first 4 bytes of the output ("HGS1") denote this as stream clone
239 239 bundle version 1.
240 240
241 241 The next 2 bytes indicate the compression type. Only "UN" is currently
242 242 supported.
243 243
244 244 The next 16 bytes are two 64-bit big endian unsigned integers indicating
245 245 file count and byte count, respectively.
246 246
247 247 The next 2 bytes is a 16-bit big endian unsigned short declaring the length
248 248 of the requirements string, including a trailing \0. The following N bytes
249 249 are the requirements string, which is ASCII containing a comma-delimited
250 250 list of repo requirements that are needed to support the data.
251 251
252 252 The remaining content is the output of ``generatev1()`` (which may be
253 253 compressed in the future).
254 254
255 255 Returns a tuple of (requirements, data generator).
256 256 """
257 257 if compression != 'UN':
258 258 raise ValueError('we do not support the compression argument yet')
259 259
260 260 requirements = repo.requirements & repo.supportedformats
261 261 requires = ','.join(sorted(requirements))
262 262
263 263 def gen():
264 264 yield 'HGS1'
265 265 yield compression
266 266
267 267 filecount, bytecount, it = generatev1(repo)
268 268 repo.ui.status(_('writing %d bytes for %d files\n') %
269 269 (bytecount, filecount))
270 270
271 271 yield struct.pack('>QQ', filecount, bytecount)
272 272 yield struct.pack('>H', len(requires) + 1)
273 273 yield requires + '\0'
274 274
275 275 # This is where we'll add compression in the future.
276 276 assert compression == 'UN'
277 277
278 278 seen = 0
279 279 repo.ui.progress(_('bundle'), 0, total=bytecount)
280 280
281 281 for chunk in it:
282 282 seen += len(chunk)
283 283 repo.ui.progress(_('bundle'), seen, total=bytecount)
284 284 yield chunk
285 285
286 286 repo.ui.progress(_('bundle'), None)
287 287
288 288 return requirements, gen()
289 289
290 290 def consumev1(repo, fp, filecount, bytecount):
291 291 """Apply the contents from version 1 of a streaming clone file handle.
292 292
293 293 This takes the output from "streamout" and applies it to the specified
294 294 repository.
295 295
296 296 Like "streamout," the status line added by the wire protocol is not handled
297 297 by this function.
298 298 """
299 299 lock = repo.lock()
300 300 try:
301 301 repo.ui.status(_('%d files to transfer, %s of data\n') %
302 302 (filecount, util.bytecount(bytecount)))
303 303 handled_bytes = 0
304 304 repo.ui.progress(_('clone'), 0, total=bytecount)
305 305 start = time.time()
306 306
307 tr = repo.transaction(_('clone'))
307 tr = repo.transaction('clone')
308 308 try:
309 309 for i in xrange(filecount):
310 310 # XXX doesn't support '\n' or '\r' in filenames
311 311 l = fp.readline()
312 312 try:
313 313 name, size = l.split('\0', 1)
314 314 size = int(size)
315 315 except (ValueError, TypeError):
316 316 raise error.ResponseError(
317 317 _('unexpected response from remote server:'), l)
318 318 if repo.ui.debugflag:
319 319 repo.ui.debug('adding %s (%s)\n' %
320 320 (name, util.bytecount(size)))
321 321 # for backwards compat, name was partially encoded
322 322 with repo.svfs(store.decodedir(name), 'w') as ofp:
323 323 for chunk in util.filechunkiter(fp, limit=size):
324 324 handled_bytes += len(chunk)
325 325 repo.ui.progress(_('clone'), handled_bytes,
326 326 total=bytecount)
327 327 ofp.write(chunk)
328 328 tr.close()
329 329 finally:
330 330 tr.release()
331 331
332 332 # Writing straight to files circumvented the inmemory caches
333 333 repo.invalidate()
334 334
335 335 elapsed = time.time() - start
336 336 if elapsed <= 0:
337 337 elapsed = 0.001
338 338 repo.ui.progress(_('clone'), None)
339 339 repo.ui.status(_('transferred %s in %.1f seconds (%s/sec)\n') %
340 340 (util.bytecount(bytecount), elapsed,
341 341 util.bytecount(bytecount / elapsed)))
342 342 finally:
343 343 lock.release()
344 344
345 345 def applybundlev1(repo, fp):
346 346 """Apply the content from a stream clone bundle version 1.
347 347
348 348 We assume the 4 byte header has been read and validated and the file handle
349 349 is at the 2 byte compression identifier.
350 350 """
351 351 if len(repo):
352 352 raise error.Abort(_('cannot apply stream clone bundle on non-empty '
353 353 'repo'))
354 354
355 355 compression = fp.read(2)
356 356 if compression != 'UN':
357 357 raise error.Abort(_('only uncompressed stream clone bundles are '
358 358 'supported; got %s') % compression)
359 359
360 360 filecount, bytecount = struct.unpack('>QQ', fp.read(16))
361 361 requireslen = struct.unpack('>H', fp.read(2))[0]
362 362 requires = fp.read(requireslen)
363 363
364 364 if not requires.endswith('\0'):
365 365 raise error.Abort(_('malformed stream clone bundle: '
366 366 'requirements not properly encoded'))
367 367
368 368 requirements = set(requires.rstrip('\0').split(','))
369 369 missingreqs = requirements - repo.supportedformats
370 370 if missingreqs:
371 371 raise error.Abort(_('unable to apply stream clone: '
372 372 'unsupported format: %s') %
373 373 ', '.join(sorted(missingreqs)))
374 374
375 375 consumev1(repo, fp, filecount, bytecount)
376 376
377 377 class streamcloneapplier(object):
378 378 """Class to manage applying streaming clone bundles.
379 379
380 380 We need to wrap ``applybundlev1()`` in a dedicated type to enable bundle
381 381 readers to perform bundle type-specific functionality.
382 382 """
383 383 def __init__(self, fh):
384 384 self._fh = fh
385 385
386 386 def apply(self, repo):
387 387 return applybundlev1(repo, self._fh)
General Comments 0
You need to be logged in to leave comments. Login now