source file: /home/buildslave/tahoe/edgy/build/src/allmydata/mutable/servermap.py
file stats: 591 lines, 573 executed: 97.0% covered
1.
2. import sys, time
3. from zope.interface import implements
4. from itertools import count
5. from twisted.internet import defer
6. from twisted.python import failure
7. from foolscap import DeadReferenceError
8. from foolscap.eventual import eventually
9. from allmydata.util import base32, hashutil, idlib, log
10. from allmydata import storage
11. from allmydata.interfaces import IServermapUpdaterStatus
12. from pycryptopp.publickey import rsa
13.
14. from common import MODE_CHECK, MODE_ANYTHING, MODE_WRITE, MODE_READ, \
15. DictOfSets, CorruptShareError, NeedMoreDataError
16. from layout import unpack_prefix_and_signature, unpack_header, unpack_share, \
17. SIGNED_PREFIX_LENGTH
18.
19. class UpdateStatus:
20. implements(IServermapUpdaterStatus)
21. statusid_counter = count(0)
22. def __init__(self):
23. self.timings = {}
24. self.timings["per_server"] = {}
25. self.timings["cumulative_verify"] = 0.0
26. self.privkey_from = None
27. self.problems = {}
28. self.active = True
29. self.storage_index = None
30. self.mode = "?"
31. self.status = "Not started"
32. self.progress = 0.0
33. self.counter = self.statusid_counter.next()
34. self.started = time.time()
35. self.finished = None
36.
37. def add_per_server_time(self, peerid, op, sent, elapsed):
38. assert op in ("query", "late", "privkey")
39. if peerid not in self.timings["per_server"]:
40. self.timings["per_server"][peerid] = []
41. self.timings["per_server"][peerid].append((op,sent,elapsed))
42.
43. def get_started(self):
44. return self.started
45. def get_finished(self):
46. return self.finished
47. def get_storage_index(self):
48. return self.storage_index
49. def get_mode(self):
50. return self.mode
51. def get_servermap(self):
52. return self.servermap
53. def get_privkey_from(self):
54. return self.privkey_from
55. def using_helper(self):
56. return False
57. def get_size(self):
58. return "-NA-"
59. def get_status(self):
60. return self.status
61. def get_progress(self):
62. return self.progress
63. def get_active(self):
64. return self.active
65. def get_counter(self):
66. return self.counter
67.
68. def set_storage_index(self, si):
69. self.storage_index = si
70. def set_mode(self, mode):
71. self.mode = mode
72. def set_privkey_from(self, peerid):
73. self.privkey_from = peerid
74. def set_status(self, status):
75. self.status = status
76. def set_progress(self, value):
77. self.progress = value
78. def set_active(self, value):
79. self.active = value
80. def set_finished(self, when):
81. self.finished = when
82.
83. class ServerMap:
84. """I record the placement of mutable shares.
85.
86. This object records which shares (of various versions) are located on
87. which servers.
88.
89. One purpose I serve is to inform callers about which versions of the
90. mutable file are recoverable and 'current'.
91.
92. A second purpose is to serve as a state marker for test-and-set
93. operations. I am passed out of retrieval operations and back into publish
94. operations, which means 'publish this new version, but only if nothing
95. has changed since I last retrieved this data'. This reduces the chances
96. of clobbering a simultaneous (uncoordinated) write.
97.
98. @ivar servermap: a dictionary, mapping a (peerid, shnum) tuple to a
99. (versionid, timestamp) tuple. Each 'versionid' is a
100. tuple of (seqnum, root_hash, IV, segsize, datalength,
101. k, N, signed_prefix, offsets)
102.
103. @ivar connections: maps peerid to a RemoteReference
104.
105. @ivar bad_shares: dict with keys of (peerid, shnum) tuples, describing
106. shares that I should ignore (because a previous user of
107. the servermap determined that they were invalid). The
108. updater only locates a certain number of shares: if
109. some of these turn out to have integrity problems and
110. are unusable, the caller will need to mark those shares
111. as bad, then re-update the servermap, then try again.
112. The dict maps (peerid, shnum) tuple to old checkstring.
113. """
114.
115. def __init__(self):
116. self.servermap = {}
117. self.connections = {}
118. self.unreachable_peers = set() # peerids that didn't respond to queries
119. self.problems = [] # mostly for debugging
120. self.bad_shares = {} # maps (peerid,shnum) to old checkstring
121. self.last_update_mode = None
122. self.last_update_time = 0
123.
124. def mark_bad_share(self, peerid, shnum, checkstring):
125. """This share was found to be bad, either in the checkstring or
126. signature (detected during mapupdate), or deeper in the share
127. (detected at retrieve time). Remove it from our list of useful
128. shares, and remember that it is bad so we don't add it back again
129. later. We record the share's old checkstring (which might be
130. corrupted or badly signed) so that a repair operation can do the
131. test-and-set using it as a reference.
132. """
133. key = (peerid, shnum) # record checkstring
134. self.bad_shares[key] = checkstring
135. self.servermap.pop(key, None)
136.
137. def add_new_share(self, peerid, shnum, verinfo, timestamp):
138. """We've written a new share out, replacing any that was there
139. before."""
140. key = (peerid, shnum)
141. self.bad_shares.pop(key, None)
142. self.servermap[key] = (verinfo, timestamp)
143.
144. def dump(self, out=sys.stdout):
145. print >>out, "servermap:"
146.
147. for ( (peerid, shnum), (verinfo, timestamp) ) in self.servermap.items():
148. (seqnum, root_hash, IV, segsize, datalength, k, N, prefix,
149. offsets_tuple) = verinfo
150. print >>out, ("[%s]: sh#%d seq%d-%s %d-of-%d len%d" %
151. (idlib.shortnodeid_b2a(peerid), shnum,
152. seqnum, base32.b2a(root_hash)[:4], k, N,
153. datalength))
154. if self.problems:
155. print >>out, "%d PROBLEMS" % len(self.problems)
156. for f in self.problems:
157. print >>out, str(f)
158. return out
159.
160. def all_peers(self):
161. return set([peerid
162. for (peerid, shnum)
163. in self.servermap])
164.
165. def make_sharemap(self):
166. """Return a dict that maps shnum to a set of peerds that hold it."""
167. sharemap = DictOfSets()
168. for (peerid, shnum) in self.servermap:
169. sharemap.add(shnum, peerid)
170. return sharemap
171.
172. def make_versionmap(self):
173. """Return a dict that maps versionid to sets of (shnum, peerid,
174. timestamp) tuples."""
175. versionmap = DictOfSets()
176. for ( (peerid, shnum), (verinfo, timestamp) ) in self.servermap.items():
177. versionmap.add(verinfo, (shnum, peerid, timestamp))
178. return versionmap
179.
180. def shares_on_peer(self, peerid):
181. return set([shnum
182. for (s_peerid, shnum)
183. in self.servermap
184. if s_peerid == peerid])
185.
186. def version_on_peer(self, peerid, shnum):
187. key = (peerid, shnum)
188. if key in self.servermap:
189. (verinfo, timestamp) = self.servermap[key]
190. return verinfo
191. return None
192.
193. def shares_available(self):
194. """Return a dict that maps verinfo to tuples of
195. (num_distinct_shares, k, N) tuples."""
196. versionmap = self.make_versionmap()
197. all_shares = {}
198. for verinfo, shares in versionmap.items():
199. s = set()
200. for (shnum, peerid, timestamp) in shares:
201. s.add(shnum)
202. (seqnum, root_hash, IV, segsize, datalength, k, N, prefix,
203. offsets_tuple) = verinfo
204. all_shares[verinfo] = (len(s), k, N)
205. return all_shares
206.
207. def highest_seqnum(self):
208. available = self.shares_available()
209. seqnums = [verinfo[0]
210. for verinfo in available.keys()]
211. seqnums.append(0)
212. return max(seqnums)
213.
214. def summarize_version(self, verinfo):
215. """Take a versionid, return a string that describes it."""
216. (seqnum, root_hash, IV, segsize, datalength, k, N, prefix,
217. offsets_tuple) = verinfo
218. return "seq%d-%s" % (seqnum, base32.b2a(root_hash)[:4])
219.
220. def summarize_versions(self):
221. """Return a string describing which versions we know about."""
222. versionmap = self.make_versionmap()
223. bits = []
224. for (verinfo, shares) in versionmap.items():
225. vstr = self.summarize_version(verinfo)
226. shnums = set([shnum for (shnum, peerid, timestamp) in shares])
227. bits.append("%d*%s" % (len(shnums), vstr))
228. return "/".join(bits)
229.
230. def recoverable_versions(self):
231. """Return a set of versionids, one for each version that is currently
232. recoverable."""
233. versionmap = self.make_versionmap()
234.
235. recoverable_versions = set()
236. for (verinfo, shares) in versionmap.items():
237. (seqnum, root_hash, IV, segsize, datalength, k, N, prefix,
238. offsets_tuple) = verinfo
239. shnums = set([shnum for (shnum, peerid, timestamp) in shares])
240. if len(shnums) >= k:
241. # this one is recoverable
242. recoverable_versions.add(verinfo)
243.
244. return recoverable_versions
245.
246. def unrecoverable_versions(self):
247. """Return a set of versionids, one for each version that is currently
248. unrecoverable."""
249. versionmap = self.make_versionmap()
250.
251. unrecoverable_versions = set()
252. for (verinfo, shares) in versionmap.items():
253. (seqnum, root_hash, IV, segsize, datalength, k, N, prefix,
254. offsets_tuple) = verinfo
255. shnums = set([shnum for (shnum, peerid, timestamp) in shares])
256. if len(shnums) < k:
257. unrecoverable_versions.add(verinfo)
258.
259. return unrecoverable_versions
260.
261. def best_recoverable_version(self):
262. """Return a single versionid, for the so-called 'best' recoverable
263. version. Sequence number is the primary sort criteria, followed by
264. root hash. Returns None if there are no recoverable versions."""
265. recoverable = list(self.recoverable_versions())
266. recoverable.sort()
267. if recoverable:
268. return recoverable[-1]
269. return None
270.
271. def size_of_version(self, verinfo):
272. """Given a versionid (perhaps returned by best_recoverable_version),
273. return the size of the file in bytes."""
274. (seqnum, root_hash, IV, segsize, datalength, k, N, prefix,
275. offsets_tuple) = verinfo
276. return datalength
277.
278. def unrecoverable_newer_versions(self):
279. # Return a dict of versionid -> health, for versions that are
280. # unrecoverable and have later seqnums than any recoverable versions.
281. # These indicate that a write will lose data.
282. versionmap = self.make_versionmap()
283. healths = {} # maps verinfo to (found,k)
284. unrecoverable = set()
285. highest_recoverable_seqnum = -1
286. for (verinfo, shares) in versionmap.items():
287. (seqnum, root_hash, IV, segsize, datalength, k, N, prefix,
288. offsets_tuple) = verinfo
289. shnums = set([shnum for (shnum, peerid, timestamp) in shares])
290. healths[verinfo] = (len(shnums),k)
291. if len(shnums) < k:
292. unrecoverable.add(verinfo)
293. else:
294. highest_recoverable_seqnum = max(seqnum,
295. highest_recoverable_seqnum)
296.
297. newversions = {}
298. for verinfo in unrecoverable:
299. (seqnum, root_hash, IV, segsize, datalength, k, N, prefix,
300. offsets_tuple) = verinfo
301. if seqnum > highest_recoverable_seqnum:
302. newversions[verinfo] = healths[verinfo]
303.
304. return newversions
305.
306.
307. def needs_merge(self):
308. # return True if there are multiple recoverable versions with the
309. # same seqnum, meaning that MutableFileNode.read_best_version is not
310. # giving you the whole story, and that using its data to do a
311. # subsequent publish will lose information.
312. return bool(len(self.recoverable_versions()) > 1)
313.
314.
315. class ServermapUpdater:
316. def __init__(self, filenode, servermap, mode=MODE_READ):
317. """I update a servermap, locating a sufficient number of useful
318. shares and remembering where they are located.
319.
320. """
321.
322. self._node = filenode
323. self._servermap = servermap
324. self.mode = mode
325. self._running = True
326.
327. self._storage_index = filenode.get_storage_index()
328. self._last_failure = None
329.
330. self._status = UpdateStatus()
331. self._status.set_storage_index(self._storage_index)
332. self._status.set_progress(0.0)
333. self._status.set_mode(mode)
334.
335. # how much data should we read?
336. # * if we only need the checkstring, then [0:75]
337. # * if we need to validate the checkstring sig, then [543ish:799ish]
338. # * if we need the verification key, then [107:436ish]
339. # * the offset table at [75:107] tells us about the 'ish'
340. # * if we need the encrypted private key, we want [-1216ish:]
341. # * but we can't read from negative offsets
342. # * the offset table tells us the 'ish', also the positive offset
343. # A future version of the SMDF slot format should consider using
344. # fixed-size slots so we can retrieve less data. For now, we'll just
345. # read 2000 bytes, which also happens to read enough actual data to
346. # pre-fetch a 9-entry dirnode.
347. self._read_size = 2000
348. if mode == MODE_CHECK:
349. # we use unpack_prefix_and_signature, so we need 1k
350. self._read_size = 1000
351. self._need_privkey = False
352. if mode == MODE_WRITE and not self._node._privkey:
353. self._need_privkey = True
354. # check+repair: repair requires the privkey, so if we didn't happen
355. # to ask for it during the check, we'll have problems doing the
356. # publish.
357.
358. prefix = storage.si_b2a(self._storage_index)[:5]
359. self._log_number = log.msg(format="SharemapUpdater(%(si)s): starting (%(mode)s)",
360. si=prefix, mode=mode)
361.
362. def get_status(self):
363. return self._status
364.
365. def log(self, *args, **kwargs):
366. if "parent" not in kwargs:
367. kwargs["parent"] = self._log_number
368. if "facility" not in kwargs:
369. kwargs["facility"] = "tahoe.mutable.mapupdate"
370. return log.msg(*args, **kwargs)
371.
372. def update(self):
373. """Update the servermap to reflect current conditions. Returns a
374. Deferred that fires with the servermap once the update has finished."""
375. self._started = time.time()
376. self._status.set_active(True)
377.
378. # self._valid_versions is a set of validated verinfo tuples. We just
379. # use it to remember which versions had valid signatures, so we can
380. # avoid re-checking the signatures for each share.
381. self._valid_versions = set()
382.
383. # self.versionmap maps verinfo tuples to sets of (shnum, peerid,
384. # timestamp) tuples. This is used to figure out which versions might
385. # be retrievable, and to make the eventual data download faster.
386. self.versionmap = DictOfSets()
387.
388. self._done_deferred = defer.Deferred()
389.
390. # first, which peers should be talk to? Any that were in our old
391. # servermap, plus "enough" others.
392.
393. self._queries_completed = 0
394.
395. client = self._node._client
396. full_peerlist = client.get_permuted_peers("storage",
397. self._node._storage_index)
398. self.full_peerlist = full_peerlist # for use later, immutable
399. self.extra_peers = full_peerlist[:] # peers are removed as we use them
400. self._good_peers = set() # peers who had some shares
401. self._empty_peers = set() # peers who don't have any shares
402. self._bad_peers = set() # peers to whom our queries failed
403.
404. k = self._node.get_required_shares()
405. if k is None:
406. # make a guess
407. k = 3
408. N = self._node.get_required_shares()
409. if N is None:
410. N = 10
411. self.EPSILON = k
412. # we want to send queries to at least this many peers (although we
413. # might not wait for all of their answers to come back)
414. self.num_peers_to_query = k + self.EPSILON
415.
416. if self.mode == MODE_CHECK:
417. initial_peers_to_query = dict(full_peerlist)
418. must_query = set(initial_peers_to_query.keys())
419. self.extra_peers = []
420. elif self.mode == MODE_WRITE:
421. # we're planning to replace all the shares, so we want a good
422. # chance of finding them all. We will keep searching until we've
423. # seen epsilon that don't have a share.
424. self.num_peers_to_query = N + self.EPSILON
425. initial_peers_to_query, must_query = self._build_initial_querylist()
426. self.required_num_empty_peers = self.EPSILON
427.
428. # TODO: arrange to read lots of data from k-ish servers, to avoid
429. # the extra round trip required to read large directories. This
430. # might also avoid the round trip required to read the encrypted
431. # private key.
432.
433. else:
434. initial_peers_to_query, must_query = self._build_initial_querylist()
435.
436. # this is a set of peers that we are required to get responses from:
437. # they are peers who used to have a share, so we need to know where
438. # they currently stand, even if that means we have to wait for a
439. # silently-lost TCP connection to time out. We remove peers from this
440. # set as we get responses.
441. self._must_query = must_query
442.
443. # now initial_peers_to_query contains the peers that we should ask,
444. # self.must_query contains the peers that we must have heard from
445. # before we can consider ourselves finished, and self.extra_peers
446. # contains the overflow (peers that we should tap if we don't get
447. # enough responses)
448.
449. self._send_initial_requests(initial_peers_to_query)
450. self._status.timings["initial_queries"] = time.time() - self._started
451. return self._done_deferred
452.
453. def _build_initial_querylist(self):
454. initial_peers_to_query = {}
455. must_query = set()
456. for peerid in self._servermap.all_peers():
457. ss = self._servermap.connections[peerid]
458. # we send queries to everyone who was already in the sharemap
459. initial_peers_to_query[peerid] = ss
460. # and we must wait for responses from them
461. must_query.add(peerid)
462.
463. while ((self.num_peers_to_query > len(initial_peers_to_query))
464. and self.extra_peers):
465. (peerid, ss) = self.extra_peers.pop(0)
466. initial_peers_to_query[peerid] = ss
467.
468. return initial_peers_to_query, must_query
469.
470. def _send_initial_requests(self, peerlist):
471. self._status.set_status("Sending %d initial queries" % len(peerlist))
472. self._queries_outstanding = set()
473. self._sharemap = DictOfSets() # shnum -> [(peerid, seqnum, R)..]
474. dl = []
475. for (peerid, ss) in peerlist.items():
476. self._queries_outstanding.add(peerid)
477. self._do_query(ss, peerid, self._storage_index, self._read_size)
478.
479. if not peerlist:
480. # there is nobody to ask, so we need to short-circuit the state
481. # machine.
482. d = defer.maybeDeferred(self._check_for_done, None)
483. d.addErrback(self._fatal_error)
484.
485. # control flow beyond this point: state machine. Receiving responses
486. # from queries is the input. We might send out more queries, or we
487. # might produce a result.
488. return None
489.
490. def _do_query(self, ss, peerid, storage_index, readsize):
491. self.log(format="sending query to [%(peerid)s], readsize=%(readsize)d",
492. peerid=idlib.shortnodeid_b2a(peerid),
493. readsize=readsize,
494. level=log.NOISY)
495. self._servermap.connections[peerid] = ss
496. started = time.time()
497. self._queries_outstanding.add(peerid)
498. d = self._do_read(ss, peerid, storage_index, [], [(0, readsize)])
499. d.addCallback(self._got_results, peerid, readsize, (ss, storage_index),
500. started)
501. d.addErrback(self._query_failed, peerid)
502. # errors that aren't handled by _query_failed (and errors caused by
503. # _query_failed) get logged, but we still want to check for doneness.
504. d.addErrback(log.err)
505. d.addBoth(self._check_for_done)
506. d.addErrback(self._fatal_error)
507. return d
508.
509. def _do_read(self, ss, peerid, storage_index, shnums, readv):
510. d = ss.callRemote("slot_readv", storage_index, shnums, readv)
511. return d
512.
513. def _got_results(self, datavs, peerid, readsize, stuff, started):
514. lp = self.log(format="got result from [%(peerid)s], %(numshares)d shares",
515. peerid=idlib.shortnodeid_b2a(peerid),
516. numshares=len(datavs),
517. level=log.NOISY)
518. now = time.time()
519. elapsed = now - started
520. self._queries_outstanding.discard(peerid)
521. self._must_query.discard(peerid)
522. self._queries_completed += 1
523. if not self._running:
524. self.log("but we're not running, so we'll ignore it", parent=lp,
525. level=log.NOISY)
526. self._status.add_per_server_time(peerid, "late", started, elapsed)
527. return
528. self._status.add_per_server_time(peerid, "query", started, elapsed)
529.
530. if datavs:
531. self._good_peers.add(peerid)
532. else:
533. self._empty_peers.add(peerid)
534.
535. last_verinfo = None
536. last_shnum = None
537. for shnum,datav in datavs.items():
538. data = datav[0]
539. try:
540. verinfo = self._got_results_one_share(shnum, data, peerid, lp)
541. last_verinfo = verinfo
542. last_shnum = shnum
543. self._node._cache.add(verinfo, shnum, 0, data, now)
544. except CorruptShareError, e:
545. # log it and give the other shares a chance to be processed
546. f = failure.Failure()
547. self.log(format="bad share: %(f_value)s", f_value=str(f.value),
548. failure=f, parent=lp, level=log.WEIRD, umid="h5llHg")
549. self._bad_peers.add(peerid)
550. self._last_failure = f
551. checkstring = data[:SIGNED_PREFIX_LENGTH]
552. self._servermap.mark_bad_share(peerid, shnum, checkstring)
553. self._servermap.problems.append(f)
554. pass
555.
556. self._status.timings["cumulative_verify"] += (time.time() - now)
557.
558. if self._need_privkey and last_verinfo:
559. # send them a request for the privkey. We send one request per
560. # server.
561. lp2 = self.log("sending privkey request",
562. parent=lp, level=log.NOISY)
563. (seqnum, root_hash, IV, segsize, datalength, k, N, prefix,
564. offsets_tuple) = last_verinfo
565. o = dict(offsets_tuple)
566.
567. self._queries_outstanding.add(peerid)
568. readv = [ (o['enc_privkey'], (o['EOF'] - o['enc_privkey'])) ]
569. ss = self._servermap.connections[peerid]
570. privkey_started = time.time()
571. d = self._do_read(ss, peerid, self._storage_index,
572. [last_shnum], readv)
573. d.addCallback(self._got_privkey_results, peerid, last_shnum,
574. privkey_started, lp2)
575. d.addErrback(self._privkey_query_failed, peerid, last_shnum, lp2)
576. d.addErrback(log.err)
577. d.addCallback(self._check_for_done)
578. d.addErrback(self._fatal_error)
579.
580. # all done!
581. self.log("_got_results done", parent=lp, level=log.NOISY)
582.
583. def _got_results_one_share(self, shnum, data, peerid, lp):
584. self.log(format="_got_results: got shnum #%(shnum)d from peerid %(peerid)s",
585. shnum=shnum,
586. peerid=idlib.shortnodeid_b2a(peerid),
587. level=log.NOISY,
588. parent=lp)
589.
590. # this might raise NeedMoreDataError, if the pubkey and signature
591. # live at some weird offset. That shouldn't happen, so I'm going to
592. # treat it as a bad share.
593. (seqnum, root_hash, IV, k, N, segsize, datalength,
594. pubkey_s, signature, prefix) = unpack_prefix_and_signature(data)
595.
596. if not self._node.get_pubkey():
597. fingerprint = hashutil.ssk_pubkey_fingerprint_hash(pubkey_s)
598. assert len(fingerprint) == 32
599. if fingerprint != self._node._fingerprint:
600. raise CorruptShareError(peerid, shnum,
601. "pubkey doesn't match fingerprint")
602. self._node._populate_pubkey(self._deserialize_pubkey(pubkey_s))
603.
604. if self._need_privkey:
605. self._try_to_extract_privkey(data, peerid, shnum, lp)
606.
607. (ig_version, ig_seqnum, ig_root_hash, ig_IV, ig_k, ig_N,
608. ig_segsize, ig_datalen, offsets) = unpack_header(data)
609. offsets_tuple = tuple( [(key,value) for key,value in offsets.items()] )
610.
611. verinfo = (seqnum, root_hash, IV, segsize, datalength, k, N, prefix,
612. offsets_tuple)
613.
614. if verinfo not in self._valid_versions:
615. # it's a new pair. Verify the signature.
616. valid = self._node._pubkey.verify(prefix, signature)
617. if not valid:
618. raise CorruptShareError(peerid, shnum, "signature is invalid")
619.
620. # ok, it's a valid verinfo. Add it to the list of validated
621. # versions.
622. self.log(" found valid version %d-%s from %s-sh%d: %d-%d/%d/%d"
623. % (seqnum, base32.b2a(root_hash)[:4],
624. idlib.shortnodeid_b2a(peerid), shnum,
625. k, N, segsize, datalength),
626. parent=lp)
627. self._valid_versions.add(verinfo)
628. # We now know that this is a valid candidate verinfo.
629.
630. if (peerid, shnum) in self._servermap.bad_shares:
631. # we've been told that the rest of the data in this share is
632. # unusable, so don't add it to the servermap.
633. self.log("but we've been told this is a bad share",
634. parent=lp, level=log.UNUSUAL)
635. return verinfo
636.
637. # Add the info to our servermap.
638. timestamp = time.time()
639. self._servermap.add_new_share(peerid, shnum, verinfo, timestamp)
640. # and the versionmap
641. self.versionmap.add(verinfo, (shnum, peerid, timestamp))
642. return verinfo
643.
644. def _deserialize_pubkey(self, pubkey_s):
645. verifier = rsa.create_verifying_key_from_string(pubkey_s)
646. return verifier
647.
648. def _try_to_extract_privkey(self, data, peerid, shnum, lp):
649. try:
650. r = unpack_share(data)
651. except NeedMoreDataError, e:
652. # this share won't help us. oh well.
653. offset = e.encprivkey_offset
654. length = e.encprivkey_length
655. self.log("shnum %d on peerid %s: share was too short (%dB) "
656. "to get the encprivkey; [%d:%d] ought to hold it" %
657. (shnum, idlib.shortnodeid_b2a(peerid), len(data),
658. offset, offset+length),
659. parent=lp)
660. # NOTE: if uncoordinated writes are taking place, someone might
661. # change the share (and most probably move the encprivkey) before
662. # we get a chance to do one of these reads and fetch it. This
663. # will cause us to see a NotEnoughSharesError(unable to fetch
664. # privkey) instead of an UncoordinatedWriteError . This is a
665. # nuisance, but it will go away when we move to DSA-based mutable
666. # files (since the privkey will be small enough to fit in the
667. # write cap).
668.
669. return
670.
671. (seqnum, root_hash, IV, k, N, segsize, datalen,
672. pubkey, signature, share_hash_chain, block_hash_tree,
673. share_data, enc_privkey) = r
674.
675. return self._try_to_validate_privkey(enc_privkey, peerid, shnum, lp)
676.
677. def _try_to_validate_privkey(self, enc_privkey, peerid, shnum, lp):
678.
679. alleged_privkey_s = self._node._decrypt_privkey(enc_privkey)
680. alleged_writekey = hashutil.ssk_writekey_hash(alleged_privkey_s)
681. if alleged_writekey != self._node.get_writekey():
682. self.log("invalid privkey from %s shnum %d" %
683. (idlib.nodeid_b2a(peerid)[:8], shnum),
684. parent=lp, level=log.WEIRD, umid="aJVccw")
685. return
686.
687. # it's good
688. self.log("got valid privkey from shnum %d on peerid %s" %
689. (shnum, idlib.shortnodeid_b2a(peerid)),
690. parent=lp)
691. privkey = rsa.create_signing_key_from_string(alleged_privkey_s)
692. self._node._populate_encprivkey(enc_privkey)
693. self._node._populate_privkey(privkey)
694. self._need_privkey = False
695. self._status.set_privkey_from(peerid)
696.
697.
698. def _query_failed(self, f, peerid):
699. if not self._running:
700. return
701. level = log.WEIRD
702. if f.check(DeadReferenceError):
703. level = log.UNUSUAL
704. self.log(format="error during query: %(f_value)s",
705. f_value=str(f.value), failure=f,
706. level=level, umid="IHXuQg")
707. self._must_query.discard(peerid)
708. self._queries_outstanding.discard(peerid)
709. self._bad_peers.add(peerid)
710. self._servermap.problems.append(f)
711. self._servermap.unreachable_peers.add(peerid) # TODO: overkill?
712. self._queries_completed += 1
713. self._last_failure = f
714.
715. def _got_privkey_results(self, datavs, peerid, shnum, started, lp):
716. now = time.time()
717. elapsed = now - started
718. self._status.add_per_server_time(peerid, "privkey", started, elapsed)
719. self._queries_outstanding.discard(peerid)
720. if not self._need_privkey:
721. return
722. if shnum not in datavs:
723. self.log("privkey wasn't there when we asked it",
724. level=log.WEIRD, umid="VA9uDQ")
725. return
726. datav = datavs[shnum]
727. enc_privkey = datav[0]
728. self._try_to_validate_privkey(enc_privkey, peerid, shnum, lp)
729.
730. def _privkey_query_failed(self, f, peerid, shnum, lp):
731. self._queries_outstanding.discard(peerid)
732. if not self._running:
733. return
734. level = log.WEIRD
735. if f.check(DeadReferenceError):
736. level = log.UNUSUAL
737. self.log(format="error during privkey query: %(f_value)s",
738. f_value=str(f.value), failure=f,
739. parent=lp, level=level, umid="McoJ5w")
740. self._servermap.problems.append(f)
741. self._last_failure = f
742.
743. def _check_for_done(self, res):
744. # exit paths:
745. # return self._send_more_queries(outstanding) : send some more queries
746. # return self._done() : all done
747. # return : keep waiting, no new queries
748.
749. lp = self.log(format=("_check_for_done, mode is '%(mode)s', "
750. "%(outstanding)d queries outstanding, "
751. "%(extra)d extra peers available, "
752. "%(must)d 'must query' peers left, "
753. "need_privkey=%(need_privkey)s"
754. ),
755. mode=self.mode,
756. outstanding=len(self._queries_outstanding),
757. extra=len(self.extra_peers),
758. must=len(self._must_query),
759. need_privkey=self._need_privkey,
760. level=log.NOISY,
761. )
762.
763. if not self._running:
764. self.log("but we're not running", parent=lp, level=log.NOISY)
765. return
766.
767. if self._must_query:
768. # we are still waiting for responses from peers that used to have
769. # a share, so we must continue to wait. No additional queries are
770. # required at this time.
771. self.log("%d 'must query' peers left" % len(self._must_query),
772. level=log.NOISY, parent=lp)
773. return
774.
775. if (not self._queries_outstanding and not self.extra_peers):
776. # all queries have retired, and we have no peers left to ask. No
777. # more progress can be made, therefore we are done.
778. self.log("all queries are retired, no extra peers: done",
779. parent=lp)
780. return self._done()
781.
782. recoverable_versions = self._servermap.recoverable_versions()
783. unrecoverable_versions = self._servermap.unrecoverable_versions()
784.
785. # what is our completion policy? how hard should we work?
786.
787. if self.mode == MODE_ANYTHING:
788. if recoverable_versions:
789. self.log("%d recoverable versions: done"
790. % len(recoverable_versions),
791. parent=lp)
792. return self._done()
793.
794. if self.mode == MODE_CHECK:
795. # we used self._must_query, and we know there aren't any
796. # responses still waiting, so that means we must be done
797. self.log("done", parent=lp)
798. return self._done()
799.
800. MAX_IN_FLIGHT = 5
801. if self.mode == MODE_READ:
802. # if we've queried k+epsilon servers, and we see a recoverable
803. # version, and we haven't seen any unrecoverable higher-seqnum'ed
804. # versions, then we're done.
805.
806. if self._queries_completed < self.num_peers_to_query:
807. self.log(format="%(completed)d completed, %(query)d to query: need more",
808. completed=self._queries_completed,
809. query=self.num_peers_to_query,
810. level=log.NOISY, parent=lp)
811. return self._send_more_queries(MAX_IN_FLIGHT)
812. if not recoverable_versions:
813. self.log("no recoverable versions: need more",
814. level=log.NOISY, parent=lp)
815. return self._send_more_queries(MAX_IN_FLIGHT)
816. highest_recoverable = max(recoverable_versions)
817. highest_recoverable_seqnum = highest_recoverable[0]
818. for unrec_verinfo in unrecoverable_versions:
819. if unrec_verinfo[0] > highest_recoverable_seqnum:
820. # there is evidence of a higher-seqnum version, but we
821. # don't yet see enough shares to recover it. Try harder.
822. # TODO: consider sending more queries.
823. # TODO: consider limiting the search distance
824. self.log("evidence of higher seqnum: need more",
825. level=log.UNUSUAL, parent=lp)
826. return self._send_more_queries(MAX_IN_FLIGHT)
827. # all the unrecoverable versions were old or concurrent with a
828. # recoverable version. Good enough.
829. self.log("no higher-seqnum: done", parent=lp)
830. return self._done()
831.
832. if self.mode == MODE_WRITE:
833. # we want to keep querying until we've seen a few that don't have
834. # any shares, to be sufficiently confident that we've seen all
835. # the shares. This is still less work than MODE_CHECK, which asks
836. # every server in the world.
837.
838. if not recoverable_versions:
839. self.log("no recoverable versions: need more", parent=lp,
840. level=log.NOISY)
841. return self._send_more_queries(MAX_IN_FLIGHT)
842.
843. last_found = -1
844. last_not_responded = -1
845. num_not_responded = 0
846. num_not_found = 0
847. states = []
848. found_boundary = False
849.
850. for i,(peerid,ss) in enumerate(self.full_peerlist):
851. if peerid in self._bad_peers:
852. # query failed
853. states.append("x")
854. #self.log("loop [%s]: x" % idlib.shortnodeid_b2a(peerid))
855. elif peerid in self._empty_peers:
856. # no shares
857. states.append("0")
858. #self.log("loop [%s]: 0" % idlib.shortnodeid_b2a(peerid))
859. if last_found != -1:
860. num_not_found += 1
861. if num_not_found >= self.EPSILON:
862. self.log("found our boundary, %s" %
863. "".join(states),
864. parent=lp, level=log.NOISY)
865. found_boundary = True
866. break
867.
868. elif peerid in self._good_peers:
869. # yes shares
870. states.append("1")
871. #self.log("loop [%s]: 1" % idlib.shortnodeid_b2a(peerid))
872. last_found = i
873. num_not_found = 0
874. else:
875. # not responded yet
876. states.append("?")
877. #self.log("loop [%s]: ?" % idlib.shortnodeid_b2a(peerid))
878. last_not_responded = i
879. num_not_responded += 1
880.
881. if found_boundary:
882. # we need to know that we've gotten answers from
883. # everybody to the left of here
884. if last_not_responded == -1:
885. # we're done
886. self.log("have all our answers",
887. parent=lp, level=log.NOISY)
888. # .. unless we're still waiting on the privkey
889. if self._need_privkey:
890. self.log("but we're still waiting for the privkey",
891. parent=lp, level=log.NOISY)
892. # if we found the boundary but we haven't yet found
893. # the privkey, we may need to look further. If
894. # somehow all the privkeys were corrupted (but the
895. # shares were readable), then this is likely to do an
896. # exhaustive search.
897. return self._send_more_queries(MAX_IN_FLIGHT)
898. return self._done()
899. # still waiting for somebody
900. return self._send_more_queries(num_not_responded)
901.
902. # if we hit here, we didn't find our boundary, so we're still
903. # waiting for peers
904. self.log("no boundary yet, %s" % "".join(states), parent=lp,
905. level=log.NOISY)
906. return self._send_more_queries(MAX_IN_FLIGHT)
907.
908. # otherwise, keep up to 5 queries in flight. TODO: this is pretty
909. # arbitrary, really I want this to be something like k -
910. # max(known_version_sharecounts) + some extra
911. self.log("catchall: need more", parent=lp, level=log.NOISY)
912. return self._send_more_queries(MAX_IN_FLIGHT)
913.
914. def _send_more_queries(self, num_outstanding):
915. more_queries = []
916.
917. while True:
918. self.log(format=" there are %(outstanding)d queries outstanding",
919. outstanding=len(self._queries_outstanding),
920. level=log.NOISY)
921. active_queries = len(self._queries_outstanding) + len(more_queries)
922. if active_queries >= num_outstanding:
923. break
924. if not self.extra_peers:
925. break
926. more_queries.append(self.extra_peers.pop(0))
927.
928. self.log(format="sending %(more)d more queries: %(who)s",
929. more=len(more_queries),
930. who=" ".join(["[%s]" % idlib.shortnodeid_b2a(peerid)
931. for (peerid,ss) in more_queries]),
932. level=log.NOISY)
933.
934. for (peerid, ss) in more_queries:
935. self._do_query(ss, peerid, self._storage_index, self._read_size)
936. # we'll retrigger when those queries come back
937.
938. def _done(self):
939. if not self._running:
940. return
941. self._running = False
942. now = time.time()
943. elapsed = now - self._started
944. self._status.set_finished(now)
945. self._status.timings["total"] = elapsed
946. self._status.set_progress(1.0)
947. self._status.set_status("Done")
948. self._status.set_active(False)
949.
950. self._servermap.last_update_mode = self.mode
951. self._servermap.last_update_time = self._started
952. # the servermap will not be touched after this
953. self.log("servermap: %s" % self._servermap.summarize_versions())
954. eventually(self._done_deferred.callback, self._servermap)
955.
956. def _fatal_error(self, f):
957. self.log("fatal error", failure=f, level=log.WEIRD, umid="1cNvlw")
958. self._done_deferred.errback(f)
959.
960.