apt_dht/apt_dht.py

   1
   2 from binascii import b2a_hex
   3 from urlparse import urlunparse
   4 import os, re
   5
   6 from twisted.internet import defer
   7 from twisted.web2 import server, http, http_headers
   8 from twisted.python import log
   9 from twisted.python.filepath import FilePath
  10
  11 from apt_dht_conf import config
  12 from PeerManager import PeerManager
  13 from HTTPServer import TopLevel
  14 from MirrorManager import MirrorManager
  15 from CacheManager import CacheManager
  16 from Hash import HashObject
  17 from db import DB
  18 from util import findMyIPAddr
  19
  20 download_dir = 'cache'
  21
  22 class AptDHT:
  23     def __init__(self, dht):
  24         log.msg('Initializing the main apt_dht application')
  25         self.cache_dir = FilePath(config.get('DEFAULT', 'cache_dir'))
  26         if not self.cache_dir.child(download_dir).exists():
  27             self.cache_dir.child(download_dir).makedirs()
  28         self.db = DB(self.cache_dir.child('apt-dht.db'))
  29         self.dht = dht
  30         self.dht.loadConfig(config, config.get('DEFAULT', 'DHT'))
  31         self.dht.join().addCallbacks(self.joinComplete, self.joinError)
  32         self.http_server = TopLevel(self.cache_dir.child(download_dir), self)
  33         self.setDirectories = self.http_server.setDirectories
  34         self.http_site = server.Site(self.http_server)
  35         self.peers = PeerManager()
  36         self.mirrors = MirrorManager(self.cache_dir)
  37         self.cache = CacheManager(self.cache_dir.child(download_dir), self.db, self)
  38         self.my_addr = None
  39
  40     def getSite(self):
  41         return self.http_site
  42
  43     def joinComplete(self, result):
  44         self.my_addr = findMyIPAddr(result, config.getint(config.get('DEFAULT', 'DHT'), 'PORT'))
  45         if not self.my_addr:
  46             raise RuntimeError, "IP address for this machine could not be found"
  47         self.cache.scanDirectories()
  48
  49     def joinError(self, failure):
  50         log.msg("joining DHT failed miserably")
  51         log.err(failure)
  52         raise RuntimeError, "IP address for this machine could not be found"
  53
  54     def check_freshness(self, path, modtime, resp):
  55         log.msg('Checking if %s is still fresh' % path)
  56         d = self.peers.get([path], "HEAD", modtime)
  57         d.addCallback(self.check_freshness_done, path, resp)
  58         return d
  59
  60     def check_freshness_done(self, resp, path, orig_resp):
  61         if resp.code == 304:
  62             log.msg('Still fresh, returning: %s' % path)
  63             return orig_resp
  64         else:
  65             log.msg('Stale, need to redownload: %s' % path)
  66             return self.get_resp(path)
  67
  68     def get_resp(self, path):
  69         d = defer.Deferred()
  70
  71         log.msg('Trying to find hash for %s' % path)
  72         findDefer = self.mirrors.findHash(path)
  73
  74         findDefer.addCallbacks(self.findHash_done, self.findHash_error,
  75                                callbackArgs=(path, d), errbackArgs=(path, d))
  76         findDefer.addErrback(log.err)
  77         return d
  78
  79     def findHash_error(self, failure, path, d):
  80         log.err(failure)
  81         self.findHash_done(HashObject(), path, d)
  82
  83     def findHash_done(self, hash, path, d):
  84         if hash.expected() is None:
  85             log.msg('Hash for %s was not found' % path)
  86             self.lookupHash_done([], hash, path, d)
  87         else:
  88             log.msg('Found hash %s for %s' % (hash.hexexpected(), path))
  89             # Lookup hash from DHT
  90             key = hash.normexpected(bits = config.getint(config.get('DEFAULT', 'DHT'), 'HASH_LENGTH'))
  91             lookupDefer = self.dht.getValue(key)
  92             lookupDefer.addCallback(self.lookupHash_done, hash, path, d)
  93
  94     def lookupHash_done(self, locations, hash, path, d):
  95         if not locations:
  96             log.msg('Peers for %s were not found' % path)
  97             getDefer = self.peers.get([path])
  98             getDefer.addCallback(self.cache.save_file, hash, path)
  99             getDefer.addErrback(self.cache.save_error, path)
 100             getDefer.addCallbacks(d.callback, d.errback)
 101         else:
 102             log.msg('Found peers for %s: %r' % (path, locations))
 103             # Download from the found peers
 104             getDefer = self.peers.get(locations)
 105             getDefer.addCallback(self.check_response, hash, path)
 106             getDefer.addCallback(self.cache.save_file, hash, path)
 107             getDefer.addErrback(self.cache.save_error, path)
 108             getDefer.addCallbacks(d.callback, d.errback)
 109
 110     def check_response(self, response, hash, path):
 111         if response.code < 200 or response.code >= 300:
 112             log.msg('Download from peers failed, going to direct download: %s' % path)
 113             getDefer = self.peers.get([path])
 114             return getDefer
 115         return response
 116
 117     def new_cached_file(self, file_path, hash, urlpath, url = None):
 118         """Add a newly cached file to the DHT.
 119
 120         If the file was downloaded, set url to the path it was downloaded for.
 121         """
 122         if url:
 123             self.mirrors.updatedFile(url, file_path)
 124
 125         if self.my_addr and hash:
 126             site = self.my_addr + ':' + str(config.getint('DEFAULT', 'PORT'))
 127             full_path = urlunparse(('http', site, urlpath, None, None, None))
 128             key = hash.norm(bits = config.getint(config.get('DEFAULT', 'DHT'), 'HASH_LENGTH'))
 129             storeDefer = self.dht.storeValue(key, full_path)
 130             storeDefer.addCallback(self.store_done, full_path)
 131             storeDefer.addErrback(log.err)
 132
 133     def store_done(self, result, path):
 134         log.msg('Added %s to the DHT: %r' % (path, result))
 135