src/allmydata/immutable/filenode.py

   1 import copy, os.path, stat
   2 from cStringIO import StringIO
   3 from zope.interface import implements
   4 from twisted.internet import defer
   5 from twisted.internet.interfaces import IPushProducer, IConsumer
   6 from twisted.protocols import basic
   7 from foolscap.eventual import eventually
   8 from allmydata.interfaces import IFileNode, IFileURI, ICheckable, \
   9      IDownloadTarget, IUploadResults
  10 from allmydata.util import dictutil, log, base32
  11 from allmydata.util.assertutil import precondition
  12 from allmydata import uri as urimodule
  13 from allmydata.immutable.checker import Checker
  14 from allmydata.check_results import CheckResults, CheckAndRepairResults
  15 from allmydata.immutable.repairer import Repairer
  16 from allmydata.immutable import download
  17
  18 class _ImmutableFileNodeBase(object):
  19     implements(IFileNode, ICheckable)
  20
  21     def __init__(self, uri, client):
  22         precondition(urimodule.IImmutableFileURI.providedBy(uri), uri)
  23         self.u = IFileURI(uri)
  24         self._client = client
  25
  26     def get_readonly_uri(self):
  27         return self.get_uri()
  28
  29     def is_mutable(self):
  30         return False
  31
  32     def is_readonly(self):
  33         return True
  34
  35     def __hash__(self):
  36         return self.u.__hash__()
  37     def __eq__(self, other):
  38         if IFileNode.providedBy(other):
  39             return self.u.__eq__(other.u)
  40         else:
  41             return False
  42     def __ne__(self, other):
  43         if IFileNode.providedBy(other):
  44             return self.u.__eq__(other.u)
  45         else:
  46             return True
  47
  48 class PortionOfFile:
  49     # like a list slice (things[2:14]), but for a file on disk
  50     def __init__(self, fn, offset=0, size=None):
  51         self.f = open(fn, "rb")
  52         self.f.seek(offset)
  53         self.bytes_left = size
  54
  55     def read(self, size=None):
  56         # bytes_to_read = min(size, self.bytes_left), but None>anything
  57         if size is None:
  58             bytes_to_read = self.bytes_left
  59         elif self.bytes_left is None:
  60             bytes_to_read = size
  61         else:
  62             bytes_to_read = min(size, self.bytes_left)
  63         data = self.f.read(bytes_to_read)
  64         if self.bytes_left is not None:
  65             self.bytes_left -= len(data)
  66         return data
  67
  68 class DownloadCache:
  69     implements(IDownloadTarget)
  70
  71     def __init__(self, node, cachefile):
  72         self._downloader = node._client.getServiceNamed("downloader")
  73         self._uri = node.get_uri()
  74         self._storage_index = node.get_storage_index()
  75         self.milestones = set() # of (offset,size,Deferred)
  76         self.cachefile = cachefile
  77         self.download_in_progress = False
  78         # five states:
  79         #  new FileNode, no downloads ever performed
  80         #  new FileNode, leftover file (partial)
  81         #  new FileNode, leftover file (whole)
  82         #  download in progress, not yet complete
  83         #  download complete
  84
  85     def when_range_available(self, offset, size):
  86         assert isinstance(offset, (int,long))
  87         assert isinstance(size, (int,long))
  88
  89         d = defer.Deferred()
  90         self.milestones.add( (offset,size,d) )
  91         self._check_milestones()
  92         if self.milestones and not self.download_in_progress:
  93             self.download_in_progress = True
  94             log.msg(format=("immutable filenode read [%(si)s]: " +
  95                             "starting download"),
  96                     si=base32.b2a(self._storage_index),
  97                     umid="h26Heg", level=log.OPERATIONAL)
  98             d2 = self._downloader.download(self._uri, self)
  99             d2.addBoth(self._download_done)
 100             d2.addErrback(self._download_failed)
 101             d2.addErrback(log.err, umid="cQaM9g")
 102         return d
 103
 104     def read(self, consumer, offset, size):
 105         assert offset+size <= self.get_filesize()
 106         f = PortionOfFile(self.cachefile.get_filename(), offset, size)
 107         d = basic.FileSender().beginFileTransfer(f, consumer)
 108         d.addCallback(lambda lastSent: consumer)
 109         return d
 110
 111     def _download_done(self, res):
 112         # clear download_in_progress, so failed downloads can be re-tried
 113         self.download_in_progress = False
 114         return res
 115
 116     def _download_failed(self, f):
 117         # tell anyone who's waiting that we failed
 118         for m in self.milestones:
 119             (offset,size,d) = m
 120             eventually(d.errback, f)
 121         self.milestones.clear()
 122
 123     def _check_milestones(self):
 124         current_size = self.get_filesize()
 125         for m in list(self.milestones):
 126             (offset,size,d) = m
 127             if offset+size <= current_size:
 128                 log.msg(format=("immutable filenode read [%(si)s] " +
 129                                 "%(offset)d+%(size)d vs %(filesize)d: " +
 130                                 "done"),
 131                         si=base32.b2a(self._storage_index),
 132                         offset=offset, size=size, filesize=current_size,
 133                         umid="nuedUg", level=log.NOISY)
 134                 self.milestones.discard(m)
 135                 eventually(d.callback, None)
 136             else:
 137                 log.msg(format=("immutable filenode read [%(si)s] " +
 138                                 "%(offset)d+%(size)d vs %(filesize)d: " +
 139                                 "still waiting"),
 140                         si=base32.b2a(self._storage_index),
 141                         offset=offset, size=size, filesize=current_size,
 142                         umid="8PKOhg", level=log.NOISY)
 143
 144     def get_filesize(self):
 145         try:
 146             filesize = os.stat(self.cachefile.get_filename())[stat.ST_SIZE]
 147         except OSError:
 148             filesize = 0
 149         return filesize
 150
 151
 152     def open(self, size):
 153         self.f = open(self.cachefile.get_filename(), "wb")
 154
 155     def write(self, data):
 156         self.f.write(data)
 157         self._check_milestones()
 158
 159     def close(self):
 160         self.f.close()
 161         self._check_milestones()
 162
 163     def fail(self, why):
 164         pass
 165     def register_canceller(self, cb):
 166         pass
 167     def finish(self):
 168         return None
 169     # The following methods are just because the target might be a repairer.DownUpConnector,
 170     # and just because the current CHKUpload object expects to find the storage index and
 171     # encoding parameters in its Uploadable.
 172     def set_storageindex(self, storageindex):
 173         pass
 174     def set_encodingparams(self, encodingparams):
 175         pass
 176
 177
 178 class FileNode(_ImmutableFileNodeBase, log.PrefixingLogMixin):
 179     def __init__(self, uri, client, cachefile):
 180         _ImmutableFileNodeBase.__init__(self, uri, client)
 181         self.download_cache = DownloadCache(self, cachefile)
 182         prefix = uri.get_verify_cap().to_string()
 183         log.PrefixingLogMixin.__init__(self, "allmydata.immutable.filenode", prefix=prefix)
 184         self.log("starting", level=log.OPERATIONAL)
 185
 186     def get_uri(self):
 187         return self.u.to_string()
 188
 189     def get_size(self):
 190         return self.u.get_size()
 191
 192     def get_verify_cap(self):
 193         return self.u.get_verify_cap()
 194
 195     def get_repair_cap(self):
 196         # CHK files can be repaired with just the verifycap
 197         return self.u.get_verify_cap()
 198
 199     def get_storage_index(self):
 200         return self.u.storage_index
 201
 202     def check_and_repair(self, monitor, verify=False):
 203         verifycap = self.get_verify_cap()
 204         servers = self._client.get_servers("storage")
 205
 206         c = Checker(client=self._client, verifycap=verifycap, servers=servers, verify=verify, monitor=monitor)
 207         d = c.start()
 208         def _maybe_repair(cr):
 209             crr = CheckAndRepairResults(self.u.storage_index)
 210             crr.pre_repair_results = cr
 211             if cr.is_healthy():
 212                 crr.post_repair_results = cr
 213                 return defer.succeed(crr)
 214             else:
 215                 crr.repair_attempted = True
 216                 crr.repair_successful = False # until proven successful
 217                 def _gather_repair_results(ur):
 218                     assert IUploadResults.providedBy(ur), ur
 219                     # clone the cr -- check results to form the basic of the prr -- post-repair results
 220                     prr = CheckResults(cr.uri, cr.storage_index)
 221                     prr.data = copy.deepcopy(cr.data)
 222
 223                     sm = prr.data['sharemap']
 224                     assert isinstance(sm, dictutil.DictOfSets), sm
 225                     sm.update(ur.sharemap)
 226                     servers_responding = set(prr.data['servers-responding'])
 227                     servers_responding.union(ur.sharemap.iterkeys())
 228                     prr.data['servers-responding'] = list(servers_responding)
 229                     prr.data['count-shares-good'] = len(sm)
 230                     prr.data['count-good-share-hosts'] = len(sm)
 231                     is_healthy = bool(len(sm) >= self.u.total_shares)
 232                     is_recoverable = bool(len(sm) >= self.u.needed_shares)
 233                     prr.set_healthy(is_healthy)
 234                     prr.set_recoverable(is_recoverable)
 235                     crr.repair_successful = is_healthy
 236                     prr.set_needs_rebalancing(len(sm) >= self.u.total_shares)
 237
 238                     crr.post_repair_results = prr
 239                     return crr
 240                 def _repair_error(f):
 241                     # as with mutable repair, I'm not sure if I want to pass
 242                     # through a failure or not. TODO
 243                     crr.repair_successful = False
 244                     crr.repair_failure = f
 245                     return f
 246                 r = Repairer(client=self._client, verifycap=verifycap, monitor=monitor)
 247                 d = r.start()
 248                 d.addCallbacks(_gather_repair_results, _repair_error)
 249                 return d
 250
 251         d.addCallback(_maybe_repair)
 252         return d
 253
 254     def check(self, monitor, verify=False):
 255         v = Checker(client=self._client, verifycap=self.get_verify_cap(), servers=self._client.get_servers("storage"), verify=verify, monitor=monitor)
 256         return v.start()
 257
 258     def read(self, consumer, offset=0, size=None):
 259         if size is None:
 260             size = self.get_size() - offset
 261         size = min(size, self.get_size() - offset)
 262
 263         if offset == 0 and size == self.get_size():
 264             # don't use the cache, just do a normal streaming download
 265             self.log("doing normal full download", umid="VRSBwg", level=log.OPERATIONAL)
 266             return self.download(download.ConsumerAdapter(consumer))
 267
 268         d = self.download_cache.when_range_available(offset, size)
 269         d.addCallback(lambda res:
 270                       self.download_cache.read(consumer, offset, size))
 271         return d
 272
 273     def download(self, target):
 274         downloader = self._client.getServiceNamed("downloader")
 275         history = self._client.get_history()
 276         return downloader.download(self.get_uri(), target, self._parentmsgid,
 277                                    history=history)
 278
 279     def download_to_data(self):
 280         downloader = self._client.getServiceNamed("downloader")
 281         history = self._client.get_history()
 282         return downloader.download_to_data(self.get_uri(), history=history)
 283
 284 class LiteralProducer:
 285     implements(IPushProducer)
 286     def resumeProducing(self):
 287         pass
 288     def stopProducing(self):
 289         pass
 290
 291
 292 class LiteralFileNode(_ImmutableFileNodeBase):
 293
 294     def __init__(self, uri, client):
 295         precondition(urimodule.IImmutableFileURI.providedBy(uri), uri)
 296         _ImmutableFileNodeBase.__init__(self, uri, client)
 297
 298     def get_uri(self):
 299         return self.u.to_string()
 300
 301     def get_size(self):
 302         return len(self.u.data)
 303
 304     def get_verify_cap(self):
 305         return None
 306
 307     def get_repair_cap(self):
 308         return None
 309
 310     def get_storage_index(self):
 311         return None
 312
 313     def check(self, monitor, verify=False):
 314         return defer.succeed(None)
 315
 316     def check_and_repair(self, monitor, verify=False):
 317         return defer.succeed(None)
 318
 319     def read(self, consumer, offset=0, size=None):
 320         if size is None:
 321             data = self.u.data[offset:]
 322         else:
 323             data = self.u.data[offset:offset+size]
 324
 325         # We use twisted.protocols.basic.FileSender, which only does
 326         # non-streaming, i.e. PullProducer, where the receiver/consumer must
 327         # ask explicitly for each chunk of data. There are only two places in
 328         # the Twisted codebase that can't handle streaming=False, both of
 329         # which are in the upload path for an FTP/SFTP server
 330         # (protocols.ftp.FileConsumer and
 331         # vfs.adapters.ftp._FileToConsumerAdapter), neither of which is
 332         # likely to be used as the target for a Tahoe download.
 333
 334         d = basic.FileSender().beginFileTransfer(StringIO(data), consumer)
 335         d.addCallback(lambda lastSent: consumer)
 336         return d
 337
 338     def download(self, target):
 339         # note that this does not update the stats_provider
 340         data = self.u.data
 341         if IConsumer.providedBy(target):
 342             target.registerProducer(LiteralProducer(), True)
 343         target.open(len(data))
 344         target.write(data)
 345         if IConsumer.providedBy(target):
 346             target.unregisterProducer()
 347         target.close()
 348         return defer.maybeDeferred(target.finish)
 349
 350     def download_to_data(self):
 351         data = self.u.data
 352         return defer.succeed(data)