src/allmydata/mutable/publish.py

   1
   2
   3 import os, struct, time
   4 from itertools import count
   5 from zope.interface import implements
   6 from twisted.internet import defer
   7 from twisted.python import failure
   8 from allmydata.interfaces import IPublishStatus, FileTooLargeError
   9 from allmydata.util import base32, hashutil, mathutil, idlib, log
  10 from allmydata import hashtree, codec
  11 from allmydata.storage.server import si_b2a
  12 from pycryptopp.cipher.aes import AES
  13 from foolscap.api import eventually
  14
  15 from common import MODE_WRITE, MODE_CHECK, DictOfSets, \
  16      UncoordinatedWriteError, NotEnoughServersError
  17 from servermap import ServerMap
  18 from layout import pack_prefix, pack_share, unpack_header, pack_checkstring, \
  19      unpack_checkstring, SIGNED_PREFIX
  20
  21 class PublishStatus:
  22     implements(IPublishStatus)
  23     statusid_counter = count(0)
  24     def __init__(self):
  25         self.timings = {}
  26         self.timings["send_per_server"] = {}
  27         self.servermap = None
  28         self.problems = {}
  29         self.active = True
  30         self.storage_index = None
  31         self.helper = False
  32         self.encoding = ("?", "?")
  33         self.size = None
  34         self.status = "Not started"
  35         self.progress = 0.0
  36         self.counter = self.statusid_counter.next()
  37         self.started = time.time()
  38
  39     def add_per_server_time(self, peerid, elapsed):
  40         if peerid not in self.timings["send_per_server"]:
  41             self.timings["send_per_server"][peerid] = []
  42         self.timings["send_per_server"][peerid].append(elapsed)
  43
  44     def get_started(self):
  45         return self.started
  46     def get_storage_index(self):
  47         return self.storage_index
  48     def get_encoding(self):
  49         return self.encoding
  50     def using_helper(self):
  51         return self.helper
  52     def get_servermap(self):
  53         return self.servermap
  54     def get_size(self):
  55         return self.size
  56     def get_status(self):
  57         return self.status
  58     def get_progress(self):
  59         return self.progress
  60     def get_active(self):
  61         return self.active
  62     def get_counter(self):
  63         return self.counter
  64
  65     def set_storage_index(self, si):
  66         self.storage_index = si
  67     def set_helper(self, helper):
  68         self.helper = helper
  69     def set_servermap(self, servermap):
  70         self.servermap = servermap
  71     def set_encoding(self, k, n):
  72         self.encoding = (k, n)
  73     def set_size(self, size):
  74         self.size = size
  75     def set_status(self, status):
  76         self.status = status
  77     def set_progress(self, value):
  78         self.progress = value
  79     def set_active(self, value):
  80         self.active = value
  81
  82 class LoopLimitExceededError(Exception):
  83     pass
  84
  85 class Publish:
  86     """I represent a single act of publishing the mutable file to the grid. I
  87     will only publish my data if the servermap I am using still represents
  88     the current state of the world.
  89
  90     To make the initial publish, set servermap to None.
  91     """
  92
  93     # we limit the segment size as usual to constrain our memory footprint.
  94     # The max segsize is higher for mutable files, because we want to support
  95     # dirnodes with up to 10k children, and each child uses about 330 bytes.
  96     # If you actually put that much into a directory you'll be using a
  97     # footprint of around 14MB, which is higher than we'd like, but it is
  98     # more important right now to support large directories than to make
  99     # memory usage small when you use them. Once we implement MDMF (with
 100     # multiple segments), we will drop this back down, probably to 128KiB.
 101     MAX_SEGMENT_SIZE = 3500000
 102
 103     def __init__(self, filenode, servermap):
 104         self._node = filenode
 105         self._servermap = servermap
 106         self._storage_index = self._node.get_storage_index()
 107         self._log_prefix = prefix = si_b2a(self._storage_index)[:5]
 108         num = self._node._client.log("Publish(%s): starting" % prefix)
 109         self._log_number = num
 110         self._running = True
 111         self._first_write_error = None
 112
 113         self._status = PublishStatus()
 114         self._status.set_storage_index(self._storage_index)
 115         self._status.set_helper(False)
 116         self._status.set_progress(0.0)
 117         self._status.set_active(True)
 118
 119     def get_status(self):
 120         return self._status
 121
 122     def log(self, *args, **kwargs):
 123         if 'parent' not in kwargs:
 124             kwargs['parent'] = self._log_number
 125         if "facility" not in kwargs:
 126             kwargs["facility"] = "tahoe.mutable.publish"
 127         return log.msg(*args, **kwargs)
 128
 129     def publish(self, newdata):
 130         """Publish the filenode's current contents.  Returns a Deferred that
 131         fires (with None) when the publish has done as much work as it's ever
 132         going to do, or errbacks with ConsistencyError if it detects a
 133         simultaneous write.
 134         """
 135
 136         # 1: generate shares (SDMF: files are small, so we can do it in RAM)
 137         # 2: perform peer selection, get candidate servers
 138         #  2a: send queries to n+epsilon servers, to determine current shares
 139         #  2b: based upon responses, create target map
 140         # 3: send slot_testv_and_readv_and_writev messages
 141         # 4: as responses return, update share-dispatch table
 142         # 4a: may need to run recovery algorithm
 143         # 5: when enough responses are back, we're done
 144
 145         self.log("starting publish, datalen is %s" % len(newdata))
 146         if len(newdata) > self.MAX_SEGMENT_SIZE:
 147             raise FileTooLargeError("SDMF is limited to one segment, and "
 148                                     "%d > %d" % (len(newdata),
 149                                                  self.MAX_SEGMENT_SIZE))
 150         self._status.set_size(len(newdata))
 151         self._status.set_status("Started")
 152         self._started = time.time()
 153
 154         self.done_deferred = defer.Deferred()
 155
 156         self._writekey = self._node.get_writekey()
 157         assert self._writekey, "need write capability to publish"
 158
 159         # first, which servers will we publish to? We require that the
 160         # servermap was updated in MODE_WRITE, so we can depend upon the
 161         # peerlist computed by that process instead of computing our own.
 162         if self._servermap:
 163             assert self._servermap.last_update_mode in (MODE_WRITE, MODE_CHECK)
 164             # we will push a version that is one larger than anything present
 165             # in the grid, according to the servermap.
 166             self._new_seqnum = self._servermap.highest_seqnum() + 1
 167         else:
 168             # If we don't have a servermap, that's because we're doing the
 169             # initial publish
 170             self._new_seqnum = 1
 171             self._servermap = ServerMap()
 172         self._status.set_servermap(self._servermap)
 173
 174         self.log(format="new seqnum will be %(seqnum)d",
 175                  seqnum=self._new_seqnum, level=log.NOISY)
 176
 177         # having an up-to-date servermap (or using a filenode that was just
 178         # created for the first time) also guarantees that the following
 179         # fields are available
 180         self.readkey = self._node.get_readkey()
 181         self.required_shares = self._node.get_required_shares()
 182         assert self.required_shares is not None
 183         self.total_shares = self._node.get_total_shares()
 184         assert self.total_shares is not None
 185         self._status.set_encoding(self.required_shares, self.total_shares)
 186
 187         self._pubkey = self._node.get_pubkey()
 188         assert self._pubkey
 189         self._privkey = self._node.get_privkey()
 190         assert self._privkey
 191         self._encprivkey = self._node.get_encprivkey()
 192
 193         client = self._node._client
 194         full_peerlist = client.get_permuted_peers("storage",
 195                                                   self._storage_index)
 196         self.full_peerlist = full_peerlist # for use later, immutable
 197         self.bad_peers = set() # peerids who have errbacked/refused requests
 198
 199         self.newdata = newdata
 200         self.salt = os.urandom(16)
 201
 202         self.setup_encoding_parameters()
 203
 204         # if we experience any surprises (writes which were rejected because
 205         # our test vector did not match, or shares which we didn't expect to
 206         # see), we set this flag and report an UncoordinatedWriteError at the
 207         # end of the publish process.
 208         self.surprised = False
 209
 210         # as a failsafe, refuse to iterate through self.loop more than a
 211         # thousand times.
 212         self.looplimit = 1000
 213
 214         # we keep track of three tables. The first is our goal: which share
 215         # we want to see on which servers. This is initially populated by the
 216         # existing servermap.
 217         self.goal = set() # pairs of (peerid, shnum) tuples
 218
 219         # the second table is our list of outstanding queries: those which
 220         # are in flight and may or may not be delivered, accepted, or
 221         # acknowledged. Items are added to this table when the request is
 222         # sent, and removed when the response returns (or errbacks).
 223         self.outstanding = set() # (peerid, shnum) tuples
 224
 225         # the third is a table of successes: share which have actually been
 226         # placed. These are populated when responses come back with success.
 227         # When self.placed == self.goal, we're done.
 228         self.placed = set() # (peerid, shnum) tuples
 229
 230         # we also keep a mapping from peerid to RemoteReference. Each time we
 231         # pull a connection out of the full peerlist, we add it to this for
 232         # use later.
 233         self.connections = {}
 234
 235         self.bad_share_checkstrings = {}
 236
 237         # we use the servermap to populate the initial goal: this way we will
 238         # try to update each existing share in place.
 239         for (peerid, shnum) in self._servermap.servermap:
 240             self.goal.add( (peerid, shnum) )
 241             self.connections[peerid] = self._servermap.connections[peerid]
 242         # then we add in all the shares that were bad (corrupted, bad
 243         # signatures, etc). We want to replace these.
 244         for key, old_checkstring in self._servermap.bad_shares.items():
 245             (peerid, shnum) = key
 246             self.goal.add(key)
 247             self.bad_share_checkstrings[key] = old_checkstring
 248             self.connections[peerid] = self._servermap.connections[peerid]
 249
 250         # create the shares. We'll discard these as they are delivered. SDMF:
 251         # we're allowed to hold everything in memory.
 252
 253         self._status.timings["setup"] = time.time() - self._started
 254         d = self._encrypt_and_encode()
 255         d.addCallback(self._generate_shares)
 256         def _start_pushing(res):
 257             self._started_pushing = time.time()
 258             return res
 259         d.addCallback(_start_pushing)
 260         d.addCallback(self.loop) # trigger delivery
 261         d.addErrback(self._fatal_error)
 262
 263         return self.done_deferred
 264
 265     def setup_encoding_parameters(self):
 266         segment_size = min(self.MAX_SEGMENT_SIZE, len(self.newdata))
 267         # this must be a multiple of self.required_shares
 268         segment_size = mathutil.next_multiple(segment_size,
 269                                               self.required_shares)
 270         self.segment_size = segment_size
 271         if segment_size:
 272             self.num_segments = mathutil.div_ceil(len(self.newdata),
 273                                                   segment_size)
 274         else:
 275             self.num_segments = 0
 276         assert self.num_segments in [0, 1,] # SDMF restrictions
 277
 278     def _fatal_error(self, f):
 279         self.log("error during loop", failure=f, level=log.UNUSUAL)
 280         self._done(f)
 281
 282     def _update_status(self):
 283         self._status.set_status("Sending Shares: %d placed out of %d, "
 284                                 "%d messages outstanding" %
 285                                 (len(self.placed),
 286                                  len(self.goal),
 287                                  len(self.outstanding)))
 288         self._status.set_progress(1.0 * len(self.placed) / len(self.goal))
 289
 290     def loop(self, ignored=None):
 291         self.log("entering loop", level=log.NOISY)
 292         if not self._running:
 293             return
 294
 295         self.looplimit -= 1
 296         if self.looplimit <= 0:
 297             raise LoopLimitExceededError("loop limit exceeded")
 298
 299         if self.surprised:
 300             # don't send out any new shares, just wait for the outstanding
 301             # ones to be retired.
 302             self.log("currently surprised, so don't send any new shares",
 303                      level=log.NOISY)
 304         else:
 305             self.update_goal()
 306             # how far are we from our goal?
 307             needed = self.goal - self.placed - self.outstanding
 308             self._update_status()
 309
 310             if needed:
 311                 # we need to send out new shares
 312                 self.log(format="need to send %(needed)d new shares",
 313                          needed=len(needed), level=log.NOISY)
 314                 self._send_shares(needed)
 315                 return
 316
 317         if self.outstanding:
 318             # queries are still pending, keep waiting
 319             self.log(format="%(outstanding)d queries still outstanding",
 320                      outstanding=len(self.outstanding),
 321                      level=log.NOISY)
 322             return
 323
 324         # no queries outstanding, no placements needed: we're done
 325         self.log("no queries outstanding, no placements needed: done",
 326                  level=log.OPERATIONAL)
 327         now = time.time()
 328         elapsed = now - self._started_pushing
 329         self._status.timings["push"] = elapsed
 330         return self._done(None)
 331
 332     def log_goal(self, goal, message=""):
 333         logmsg = [message]
 334         for (shnum, peerid) in sorted([(s,p) for (p,s) in goal]):
 335             logmsg.append("sh%d to [%s]" % (shnum,
 336                                             idlib.shortnodeid_b2a(peerid)))
 337         self.log("current goal: %s" % (", ".join(logmsg)), level=log.NOISY)
 338         self.log("we are planning to push new seqnum=#%d" % self._new_seqnum,
 339                  level=log.NOISY)
 340
 341     def update_goal(self):
 342         # if log.recording_noisy
 343         if True:
 344             self.log_goal(self.goal, "before update: ")
 345
 346         # first, remove any bad peers from our goal
 347         self.goal = set([ (peerid, shnum)
 348                           for (peerid, shnum) in self.goal
 349                           if peerid not in self.bad_peers ])
 350
 351         # find the homeless shares:
 352         homefull_shares = set([shnum for (peerid, shnum) in self.goal])
 353         homeless_shares = set(range(self.total_shares)) - homefull_shares
 354         homeless_shares = sorted(list(homeless_shares))
 355         # place them somewhere. We prefer unused servers at the beginning of
 356         # the available peer list.
 357
 358         if not homeless_shares:
 359             return
 360
 361         # if an old share X is on a node, put the new share X there too.
 362         # TODO: 1: redistribute shares to achieve one-per-peer, by copying
 363         #       shares from existing peers to new (less-crowded) ones. The
 364         #       old shares must still be updated.
 365         # TODO: 2: move those shares instead of copying them, to reduce future
 366         #       update work
 367
 368         # this is a bit CPU intensive but easy to analyze. We create a sort
 369         # order for each peerid. If the peerid is marked as bad, we don't
 370         # even put them in the list. Then we care about the number of shares
 371         # which have already been assigned to them. After that we care about
 372         # their permutation order.
 373         old_assignments = DictOfSets()
 374         for (peerid, shnum) in self.goal:
 375             old_assignments.add(peerid, shnum)
 376
 377         peerlist = []
 378         for i, (peerid, ss) in enumerate(self.full_peerlist):
 379             if peerid in self.bad_peers:
 380                 continue
 381             entry = (len(old_assignments.get(peerid, [])), i, peerid, ss)
 382             peerlist.append(entry)
 383         peerlist.sort()
 384
 385         if not peerlist:
 386             raise NotEnoughServersError("Ran out of non-bad servers, "
 387                                         "first_error=%s" %
 388                                         str(self._first_write_error),
 389                                         self._first_write_error)
 390
 391         new_assignments = []
 392         # we then index this peerlist with an integer, because we may have to
 393         # wrap. We update the goal as we go.
 394         i = 0
 395         for shnum in homeless_shares:
 396             (ignored1, ignored2, peerid, ss) = peerlist[i]
 397             # if we are forced to send a share to a server that already has
 398             # one, we may have two write requests in flight, and the
 399             # servermap (which was computed before either request was sent)
 400             # won't reflect the new shares, so the second response will be
 401             # surprising. There is code in _got_write_answer() to tolerate
 402             # this, otherwise it would cause the publish to fail with an
 403             # UncoordinatedWriteError. See #546 for details of the trouble
 404             # this used to cause.
 405             self.goal.add( (peerid, shnum) )
 406             self.connections[peerid] = ss
 407             i += 1
 408             if i >= len(peerlist):
 409                 i = 0
 410         if True:
 411             self.log_goal(self.goal, "after update: ")
 412
 413
 414
 415     def _encrypt_and_encode(self):
 416         # this returns a Deferred that fires with a list of (sharedata,
 417         # sharenum) tuples. TODO: cache the ciphertext, only produce the
 418         # shares that we care about.
 419         self.log("_encrypt_and_encode")
 420
 421         self._status.set_status("Encrypting")
 422         started = time.time()
 423
 424         key = hashutil.ssk_readkey_data_hash(self.salt, self.readkey)
 425         enc = AES(key)
 426         crypttext = enc.process(self.newdata)
 427         assert len(crypttext) == len(self.newdata)
 428
 429         now = time.time()
 430         self._status.timings["encrypt"] = now - started
 431         started = now
 432
 433         # now apply FEC
 434
 435         self._status.set_status("Encoding")
 436         fec = codec.CRSEncoder()
 437         fec.set_params(self.segment_size,
 438                        self.required_shares, self.total_shares)
 439         piece_size = fec.get_block_size()
 440         crypttext_pieces = [None] * self.required_shares
 441         for i in range(len(crypttext_pieces)):
 442             offset = i * piece_size
 443             piece = crypttext[offset:offset+piece_size]
 444             piece = piece + "\x00"*(piece_size - len(piece)) # padding
 445             crypttext_pieces[i] = piece
 446             assert len(piece) == piece_size
 447
 448         d = fec.encode(crypttext_pieces)
 449         def _done_encoding(res):
 450             elapsed = time.time() - started
 451             self._status.timings["encode"] = elapsed
 452             return res
 453         d.addCallback(_done_encoding)
 454         return d
 455
 456     def _generate_shares(self, shares_and_shareids):
 457         # this sets self.shares and self.root_hash
 458         self.log("_generate_shares")
 459         self._status.set_status("Generating Shares")
 460         started = time.time()
 461
 462         # we should know these by now
 463         privkey = self._privkey
 464         encprivkey = self._encprivkey
 465         pubkey = self._pubkey
 466
 467         (shares, share_ids) = shares_and_shareids
 468
 469         assert len(shares) == len(share_ids)
 470         assert len(shares) == self.total_shares
 471         all_shares = {}
 472         block_hash_trees = {}
 473         share_hash_leaves = [None] * len(shares)
 474         for i in range(len(shares)):
 475             share_data = shares[i]
 476             shnum = share_ids[i]
 477             all_shares[shnum] = share_data
 478
 479             # build the block hash tree. SDMF has only one leaf.
 480             leaves = [hashutil.block_hash(share_data)]
 481             t = hashtree.HashTree(leaves)
 482             block_hash_trees[shnum] = block_hash_tree = list(t)
 483             share_hash_leaves[shnum] = t[0]
 484         for leaf in share_hash_leaves:
 485             assert leaf is not None
 486         share_hash_tree = hashtree.HashTree(share_hash_leaves)
 487         share_hash_chain = {}
 488         for shnum in range(self.total_shares):
 489             needed_hashes = share_hash_tree.needed_hashes(shnum)
 490             share_hash_chain[shnum] = dict( [ (i, share_hash_tree[i])
 491                                               for i in needed_hashes ] )
 492         root_hash = share_hash_tree[0]
 493         assert len(root_hash) == 32
 494         self.log("my new root_hash is %s" % base32.b2a(root_hash))
 495         self._new_version_info = (self._new_seqnum, root_hash, self.salt)
 496
 497         prefix = pack_prefix(self._new_seqnum, root_hash, self.salt,
 498                              self.required_shares, self.total_shares,
 499                              self.segment_size, len(self.newdata))
 500
 501         # now pack the beginning of the share. All shares are the same up
 502         # to the signature, then they have divergent share hash chains,
 503         # then completely different block hash trees + salt + share data,
 504         # then they all share the same encprivkey at the end. The sizes
 505         # of everything are the same for all shares.
 506
 507         sign_started = time.time()
 508         signature = privkey.sign(prefix)
 509         self._status.timings["sign"] = time.time() - sign_started
 510
 511         verification_key = pubkey.serialize()
 512
 513         final_shares = {}
 514         for shnum in range(self.total_shares):
 515             final_share = pack_share(prefix,
 516                                      verification_key,
 517                                      signature,
 518                                      share_hash_chain[shnum],
 519                                      block_hash_trees[shnum],
 520                                      all_shares[shnum],
 521                                      encprivkey)
 522             final_shares[shnum] = final_share
 523         elapsed = time.time() - started
 524         self._status.timings["pack"] = elapsed
 525         self.shares = final_shares
 526         self.root_hash = root_hash
 527
 528         # we also need to build up the version identifier for what we're
 529         # pushing. Extract the offsets from one of our shares.
 530         assert final_shares
 531         offsets = unpack_header(final_shares.values()[0])[-1]
 532         offsets_tuple = tuple( [(key,value) for key,value in offsets.items()] )
 533         verinfo = (self._new_seqnum, root_hash, self.salt,
 534                    self.segment_size, len(self.newdata),
 535                    self.required_shares, self.total_shares,
 536                    prefix, offsets_tuple)
 537         self.versioninfo = verinfo
 538
 539
 540
 541     def _send_shares(self, needed):
 542         self.log("_send_shares")
 543
 544         # we're finally ready to send out our shares. If we encounter any
 545         # surprises here, it's because somebody else is writing at the same
 546         # time. (Note: in the future, when we remove the _query_peers() step
 547         # and instead speculate about [or remember] which shares are where,
 548         # surprises here are *not* indications of UncoordinatedWriteError,
 549         # and we'll need to respond to them more gracefully.)
 550
 551         # needed is a set of (peerid, shnum) tuples. The first thing we do is
 552         # organize it by peerid.
 553
 554         peermap = DictOfSets()
 555         for (peerid, shnum) in needed:
 556             peermap.add(peerid, shnum)
 557
 558         # the next thing is to build up a bunch of test vectors. The
 559         # semantics of Publish are that we perform the operation if the world
 560         # hasn't changed since the ServerMap was constructed (more or less).
 561         # For every share we're trying to place, we create a test vector that
 562         # tests to see if the server*share still corresponds to the
 563         # map.
 564
 565         all_tw_vectors = {} # maps peerid to tw_vectors
 566         sm = self._servermap.servermap
 567
 568         for key in needed:
 569             (peerid, shnum) = key
 570
 571             if key in sm:
 572                 # an old version of that share already exists on the
 573                 # server, according to our servermap. We will create a
 574                 # request that attempts to replace it.
 575                 old_versionid, old_timestamp = sm[key]
 576                 (old_seqnum, old_root_hash, old_salt, old_segsize,
 577                  old_datalength, old_k, old_N, old_prefix,
 578                  old_offsets_tuple) = old_versionid
 579                 old_checkstring = pack_checkstring(old_seqnum,
 580                                                    old_root_hash,
 581                                                    old_salt)
 582                 testv = (0, len(old_checkstring), "eq", old_checkstring)
 583
 584             elif key in self.bad_share_checkstrings:
 585                 old_checkstring = self.bad_share_checkstrings[key]
 586                 testv = (0, len(old_checkstring), "eq", old_checkstring)
 587
 588             else:
 589                 # add a testv that requires the share not exist
 590
 591                 # Unfortunately, foolscap-0.2.5 has a bug in the way inbound
 592                 # constraints are handled. If the same object is referenced
 593                 # multiple times inside the arguments, foolscap emits a
 594                 # 'reference' token instead of a distinct copy of the
 595                 # argument. The bug is that these 'reference' tokens are not
 596                 # accepted by the inbound constraint code. To work around
 597                 # this, we need to prevent python from interning the
 598                 # (constant) tuple, by creating a new copy of this vector
 599                 # each time.
 600
 601                 # This bug is fixed in foolscap-0.2.6, and even though this
 602                 # version of Tahoe requires foolscap-0.3.1 or newer, we are
 603                 # supposed to be able to interoperate with older versions of
 604                 # Tahoe which are allowed to use older versions of foolscap,
 605                 # including foolscap-0.2.5 . In addition, I've seen other
 606                 # foolscap problems triggered by 'reference' tokens (see #541
 607                 # for details). So we must keep this workaround in place.
 608
 609                 #testv = (0, 1, 'eq', "")
 610                 testv = tuple([0, 1, 'eq', ""])
 611
 612             testvs = [testv]
 613             # the write vector is simply the share
 614             writev = [(0, self.shares[shnum])]
 615
 616             if peerid not in all_tw_vectors:
 617                 all_tw_vectors[peerid] = {}
 618                 # maps shnum to (testvs, writevs, new_length)
 619             assert shnum not in all_tw_vectors[peerid]
 620
 621             all_tw_vectors[peerid][shnum] = (testvs, writev, None)
 622
 623         # we read the checkstring back from each share, however we only use
 624         # it to detect whether there was a new share that we didn't know
 625         # about. The success or failure of the write will tell us whether
 626         # there was a collision or not. If there is a collision, the first
 627         # thing we'll do is update the servermap, which will find out what
 628         # happened. We could conceivably reduce a roundtrip by using the
 629         # readv checkstring to populate the servermap, but really we'd have
 630         # to read enough data to validate the signatures too, so it wouldn't
 631         # be an overall win.
 632         read_vector = [(0, struct.calcsize(SIGNED_PREFIX))]
 633
 634         # ok, send the messages!
 635         self.log("sending %d shares" % len(all_tw_vectors), level=log.NOISY)
 636         started = time.time()
 637         for (peerid, tw_vectors) in all_tw_vectors.items():
 638
 639             write_enabler = self._node.get_write_enabler(peerid)
 640             renew_secret = self._node.get_renewal_secret(peerid)
 641             cancel_secret = self._node.get_cancel_secret(peerid)
 642             secrets = (write_enabler, renew_secret, cancel_secret)
 643             shnums = tw_vectors.keys()
 644
 645             for shnum in shnums:
 646                 self.outstanding.add( (peerid, shnum) )
 647
 648             d = self._do_testreadwrite(peerid, secrets,
 649                                        tw_vectors, read_vector)
 650             d.addCallbacks(self._got_write_answer, self._got_write_error,
 651                            callbackArgs=(peerid, shnums, started),
 652                            errbackArgs=(peerid, shnums, started))
 653             d.addCallback(self.loop)
 654             d.addErrback(self._fatal_error)
 655
 656         self._update_status()
 657         self.log("%d shares sent" % len(all_tw_vectors), level=log.NOISY)
 658
 659     def _do_testreadwrite(self, peerid, secrets,
 660                           tw_vectors, read_vector):
 661         storage_index = self._storage_index
 662         ss = self.connections[peerid]
 663
 664         #print "SS[%s] is %s" % (idlib.shortnodeid_b2a(peerid), ss), ss.tracker.interfaceName
 665         d = ss.callRemote("slot_testv_and_readv_and_writev",
 666                           storage_index,
 667                           secrets,
 668                           tw_vectors,
 669                           read_vector)
 670         return d
 671
 672     def _got_write_answer(self, answer, peerid, shnums, started):
 673         lp = self.log("_got_write_answer from %s" %
 674                       idlib.shortnodeid_b2a(peerid))
 675         for shnum in shnums:
 676             self.outstanding.discard( (peerid, shnum) )
 677
 678         now = time.time()
 679         elapsed = now - started
 680         self._status.add_per_server_time(peerid, elapsed)
 681
 682         wrote, read_data = answer
 683
 684         surprise_shares = set(read_data.keys()) - set(shnums)
 685
 686         surprised = False
 687         for shnum in surprise_shares:
 688             # read_data is a dict mapping shnum to checkstring (SIGNED_PREFIX)
 689             checkstring = read_data[shnum][0]
 690             their_version_info = unpack_checkstring(checkstring)
 691             if their_version_info == self._new_version_info:
 692                 # they have the right share, somehow
 693
 694                 if (peerid,shnum) in self.goal:
 695                     # and we want them to have it, so we probably sent them a
 696                     # copy in an earlier write. This is ok, and avoids the
 697                     # #546 problem.
 698                     continue
 699
 700                 # They aren't in our goal, but they are still for the right
 701                 # version. Somebody else wrote them, and it's a convergent
 702                 # uncoordinated write. Pretend this is ok (don't be
 703                 # surprised), since I suspect there's a decent chance that
 704                 # we'll hit this in normal operation.
 705                 continue
 706
 707             else:
 708                 # the new shares are of a different version
 709                 if peerid in self._servermap.reachable_peers:
 710                     # we asked them about their shares, so we had knowledge
 711                     # of what they used to have. Any surprising shares must
 712                     # have come from someone else, so UCW.
 713                     surprised = True
 714                 else:
 715                     # we didn't ask them, and now we've discovered that they
 716                     # have a share we didn't know about. This indicates that
 717                     # mapupdate should have wokred harder and asked more
 718                     # servers before concluding that it knew about them all.
 719
 720                     # signal UCW, but make sure to ask this peer next time,
 721                     # so we'll remember to update it if/when we retry.
 722                     surprised = True
 723                     # TODO: ask this peer next time. I don't yet have a good
 724                     # way to do this. Two insufficient possibilities are:
 725                     #
 726                     # self._servermap.add_new_share(peerid, shnum, verinfo, now)
 727                     #  but that requires fetching/validating/parsing the whole
 728                     #  version string, and all we have is the checkstring
 729                     # self._servermap.mark_bad_share(peerid, shnum, checkstring)
 730                     #  that will make publish overwrite the share next time,
 731                     #  but it won't re-query the server, and it won't make
 732                     #  mapupdate search further
 733
 734                     # TODO later: when publish starts, do
 735                     # servermap.get_best_version(), extract the seqnum,
 736                     # subtract one, and store as highest-replaceable-seqnum.
 737                     # Then, if this surprise-because-we-didn't-ask share is
 738                     # of highest-replaceable-seqnum or lower, we're allowed
 739                     # to replace it: send out a new writev (or rather add it
 740                     # to self.goal and loop).
 741                     pass
 742
 743                 surprised = True
 744
 745         if surprised:
 746             self.log("they had shares %s that we didn't know about" %
 747                      (list(surprise_shares),),
 748                      parent=lp, level=log.WEIRD, umid="un9CSQ")
 749             self.surprised = True
 750
 751         if not wrote:
 752             # TODO: there are two possibilities. The first is that the server
 753             # is full (or just doesn't want to give us any room), which means
 754             # we shouldn't ask them again, but is *not* an indication of an
 755             # uncoordinated write. The second is that our testv failed, which
 756             # *does* indicate an uncoordinated write. We currently don't have
 757             # a way to tell these two apart (in fact, the storage server code
 758             # doesn't have the option of refusing our share).
 759             #
 760             # If the server is full, mark the peer as bad (so we don't ask
 761             # them again), but don't set self.surprised. The loop() will find
 762             # a new server.
 763             #
 764             # If the testv failed, log it, set self.surprised, but don't
 765             # bother adding to self.bad_peers .
 766
 767             self.log("our testv failed, so the write did not happen",
 768                      parent=lp, level=log.WEIRD, umid="8sc26g")
 769             self.surprised = True
 770             self.bad_peers.add(peerid) # don't ask them again
 771             # use the checkstring to add information to the log message
 772             for (shnum,readv) in read_data.items():
 773                 checkstring = readv[0]
 774                 (other_seqnum,
 775                  other_roothash,
 776                  other_salt) = unpack_checkstring(checkstring)
 777                 expected_version = self._servermap.version_on_peer(peerid,
 778                                                                    shnum)
 779                 if expected_version:
 780                     (seqnum, root_hash, IV, segsize, datalength, k, N, prefix,
 781                      offsets_tuple) = expected_version
 782                     self.log("somebody modified the share on us:"
 783                              " shnum=%d: I thought they had #%d:R=%s,"
 784                              " but testv reported #%d:R=%s" %
 785                              (shnum,
 786                               seqnum, base32.b2a(root_hash)[:4],
 787                               other_seqnum, base32.b2a(other_roothash)[:4]),
 788                              parent=lp, level=log.NOISY)
 789                 # if expected_version==None, then we didn't expect to see a
 790                 # share on that peer, and the 'surprise_shares' clause above
 791                 # will have logged it.
 792             # self.loop() will take care of finding new homes
 793             return
 794
 795         for shnum in shnums:
 796             self.placed.add( (peerid, shnum) )
 797             # and update the servermap
 798             self._servermap.add_new_share(peerid, shnum,
 799                                           self.versioninfo, started)
 800
 801         # self.loop() will take care of checking to see if we're done
 802         return
 803
 804     def _got_write_error(self, f, peerid, shnums, started):
 805         for shnum in shnums:
 806             self.outstanding.discard( (peerid, shnum) )
 807         self.bad_peers.add(peerid)
 808         if self._first_write_error is None:
 809             self._first_write_error = f
 810         self.log(format="error while writing shares %(shnums)s to peerid %(peerid)s",
 811                  shnums=list(shnums), peerid=idlib.shortnodeid_b2a(peerid),
 812                  failure=f,
 813                  level=log.UNUSUAL)
 814         # self.loop() will take care of checking to see if we're done
 815         return
 816
 817
 818     def _done(self, res):
 819         if not self._running:
 820             return
 821         self._running = False
 822         now = time.time()
 823         self._status.timings["total"] = now - self._started
 824         self._status.set_active(False)
 825         if isinstance(res, failure.Failure):
 826             self.log("Publish done, with failure", failure=res,
 827                      level=log.WEIRD, umid="nRsR9Q")
 828             self._status.set_status("Failed")
 829         elif self.surprised:
 830             self.log("Publish done, UncoordinatedWriteError", level=log.UNUSUAL)
 831             self._status.set_status("UncoordinatedWriteError")
 832             # deliver a failure
 833             res = failure.Failure(UncoordinatedWriteError())
 834             # TODO: recovery
 835         else:
 836             self.log("Publish done, success")
 837             self._status.set_status("Done")
 838             self._status.set_progress(1.0)
 839         eventually(self.done_deferred.callback, res)
 840
 841