instrumentation/next-share/BaseLib/Core/Subtitles/PeerHaveManager.py

   1 # Written by Andrea Reale
   2 # see LICENSE.txt for license information
   3
   4 from __future__ import with_statement
   5 import time
   6 from BaseLib.Core.Subtitles.MetadataDomainObjects import Languages
   7 import threading
   8
   9
  10
  11 PEERS_RESULT_LIMIT = 5
  12 HAVE_VALIDITY_TIME = 7*86400 # one week (too big? to small?)
  13
  14 # how often (in seconds) old have messages will be removed from the database
  15 # -1 means that they will be cleaned up only at Tribler's startup
  16 CLEANUP_PERIOD = -1
  17
  18 class PeersHaveManager(object):
  19     '''
  20     Manages the insertion, retrieval and manipulation of
  21     subtitle have messages from other peers.
  22
  23     The public interface consists only of the two methods:
  24
  25     + getPeersHaving(channel, infohash, bitmask)
  26     + newHaveReceived(channel, infohash, peer_id, havemask)
  27
  28     See method descriptions for further details
  29     '''
  30
  31     __single = None
  32     _singletonLock = threading.RLock()
  33     def __init__(self):
  34
  35         with PeersHaveManager._singletonLock:
  36             #Singleton pattern not enforced: this makes testing easier
  37             PeersHaveManager.__single = self
  38
  39         self._haveDb = None
  40         self._olBridge = None
  41         self._cleanupPeriod = CLEANUP_PERIOD
  42         self._haveValidityTime = HAVE_VALIDITY_TIME
  43         self._langsUtility = Languages.LanguagesProvider.getLanguagesInstance()
  44         self._firstCleanedUp = False
  45
  46         self._registered = False
  47
  48     @staticmethod
  49     def getInstance():
  50         with PeersHaveManager._singletonLock:
  51             if PeersHaveManager.__single == None:
  52                 PeersHaveManager()
  53
  54         return PeersHaveManager.__single
  55
  56     def register(self, haveDb, olBridge):
  57         '''
  58         Inject dependencies
  59
  60         @type haveDb: BaseLib.Core.CacheDB.MetadataDBHandler
  61         @type olBridge: OverlayBridge
  62
  63         '''
  64         assert haveDb is not None
  65         assert olBridge is not None
  66
  67         self._haveDb = haveDb
  68         self._olBridge = olBridge
  69
  70         self._registered = True
  71
  72     def isRegistered(self):
  73         return self._registered
  74
  75
  76     def getPeersHaving(self, channel, infohash, bitmask, limit=PEERS_RESULT_LIMIT):
  77         '''
  78         Returns a list of permids of peers having all the subtitles for
  79         (channel, infohash) specified in the bitmask
  80
  81         Notice that if there exist a peer that has only some of the subtitles
  82         specified in the bitmask, that peer will not be included
  83         in the returned list.
  84
  85         This implementation returns the peers sorted by most recently received
  86         have message first.
  87
  88         @type channel: str
  89         @param channel: binary channel_id
  90
  91         @type infohash: str
  92         @param infohash: binary infohash
  93
  94         @type bitmask: int
  95         @param bitmask: a 32 bit bitmask specifieng the desired subtitles languages
  96                         for returned peers to have.
  97
  98         @type limit: int
  99         @param limit: an upper bound on the size of the returned list. Notice
 100                       that anyway the returned list may be smaller then limit
 101                       (Default 5)
 102
 103         @rtype: list
 104         @return: a list of binary permids of peers that have all the subitles
 105                  specified by the bitmask. If there is no suitable entry the returned
 106                  list will be empty
 107         '''
 108
 109         # results are already ordered by timestamp due the current
 110         # MetadataDBHandler implementation
 111         peersTuples = self._haveDb.getHaveEntries(channel, infohash)
 112         peers_length = len(peersTuples)
 113         length = peers_length if peers_length < limit else limit
 114
 115         results = list()
 116
 117         for i in range(length):
 118             peer_id, havemask, timestamp = peersTuples[i]
 119             if havemask & bitmask == bitmask:
 120                 results.append(peer_id)
 121
 122         if len(results) == 0:
 123             #if no results, and if the channel owner was not in the initial
 124             #list, consider him always as a valid source
 125             results.append(channel)
 126
 127         return results
 128
 129
 130     def newHaveReceived(self, channel, infohash, peer_id, havemask):
 131         '''
 132         Notify the PeerHaveManager that a new SUBTITLE HAVE announcement has been
 133         received.
 134
 135         @type channel: str
 136         @param channel: binary channel_id
 137
 138         @type infohash: str
 139         @param infohash: binary infohash
 140
 141         @type peer_id: str
 142         @param channel: binary permid of the peer that sent
 143                         this havemask
 144
 145         @type havemask: int
 146         @param havemask: integer bitmask representing which combination of subtitles
 147                          peer_id has for the given (channel, infohash) pair
 148         '''
 149
 150
 151         timestamp = int(time.time())
 152         self._haveDb.insertOrUpdateHave(channel, infohash, peer_id, havemask, timestamp)
 153
 154
 155     def retrieveMyHaveMask(self, channel, infohash):
 156         '''
 157         Creates the havemask for locally available subtitles for channel,infohash
 158
 159         @type channel: str
 160         @param channel: a channelid to retrieve the local availability mask for (binary)
 161         @type infohash: str
 162         @param infohash: the infohash of the torrent to retrieve to local availability mask
 163                         for (binary)
 164
 165         @rtype: int
 166         @return: a bitmask reprsenting wich subtitles languages are locally available
 167                  for the given (channel, infohash) pair. If no one is available, or even
 168                  if no rich metadata has been ever received for that pair, a zero bitmask
 169                  will be returned. (i.e. this method should never thorow an exception if the
 170                  passed parametrers are formally valid)
 171         '''
 172
 173         localSubtitlesDict = self._haveDb.getLocalSubtitles(channel, infohash)
 174
 175         havemask = self._langsUtility.langCodesToMask(localSubtitlesDict.keys())
 176
 177         return havemask
 178
 179     def startupCleanup(self):
 180         '''
 181         Cleanup old entries in the have database.
 182
 183         This method is meant to be called only one time in PeersManager instance lifetime,
 184         i.e. at Tribler's startup. Successive calls will have no effect.
 185
 186         If CLEANUP_PERIOD is set to a positive value, period cleanups actions will be
 187         scheduled.
 188         '''
 189         if not self._firstCleanedUp:
 190             self._firstCleanedUp = True
 191             self._schedulePeriodicCleanup()
 192
 193     def _schedulePeriodicCleanup(self):
 194
 195         minimumAllowedTS = int(time.time()) - self._haveValidityTime
 196         self._haveDb.cleanupOldHave(minimumAllowedTS)
 197
 198         if self._cleanupPeriod > 0:
 199             self._olBridge.add_task(self._schedulePeriodicCleanup, self._cleanupPeriod)
 200
 201
 202
 203
 204