ratemanager.py

ABC-win32-v3.1 一个P2P软源代码

import sys
import wx
import os

from threading import Event
from time import time, clock

#from traceback import print_exc
#from cStringIO import StringIO

from Utility.constants import * #IGNORE:W0611
from Utility.helpers import union, difference


wxEVT_INVOKE = wx.NewEventType()

def EVT_INVOKE(win, func):
    win.Connect(-1, -1, wxEVT_INVOKE, func)
    
def DELEVT_INVOKE(win):
    win.Disconnect(-1, -1, wxEVT_INVOKE)

class InvokeEvent(wx.PyEvent):
    def __init__(self, func, args, kwargs):
        wx.PyEvent.__init__(self)
        self.SetEventType(wxEVT_INVOKE)
        self.func = func
        self.args = args
        self.kwargs = kwargs


################################################################
#
# Class: RateManger
#
# Keep the upload and download rates for torrents within
# the defined local and global limits
#
################################################################
class RateManager(wx.EvtHandler):
    def __init__(self, queue):
        wx.EvtHandler.__init__(self)
               
        self.queue = queue
        self.utility = queue.utility

        # For Upload Rate Maximizer
        # Time when the upload rate is lower than the URM threshold for the first time
        self.urm_time = { 'under' : 0.0, 
                          'checking'  : 0.0 }
                
        # bandwidth between reserved rate and measured rate
        # above which a torrent will have its reserved rate lowered
        self.calcupth1 = 2.3
        
        # bandwidth between reserved rate and measured rate
        # under which a torrent will have its reserved rate raised
        self.calcupth2 = 0.7
        
        self.meancalcupth = (self.calcupth1 + self.calcupth2) / 2
        
        self.lastmaxrate = -1

        # Minimum rate setting
        # 3 KB/s for uploads, 0.01KB/s for downloads
        # (effectively, no real limit for downloads)
        self.rateminimum = { "up": 3.0,
                             "down": 0.01 }

        self.flag = Event()
        EVT_INVOKE(self, self.onInvoke)

    def onInvoke(self, event):
        if not self.flag.isSet():
            event.func(*event.args, **event.kwargs)

    def invokeLater(self, func, args = None, kwargs = None):
        if args is None:
            args = []
        if kwargs is None:
            kwargs = {}

        if not self.flag.isSet():
            wx.PostEvent(self, InvokeEvent(func, args, kwargs))
                
    def RunTasks(self):
        self.invokeLater(self.RateTasks)
            
    def RateTasks(self):
        self.flag.set()
        
        self.UploadRateMaximizer()
        
        self.CalculateBandwidth("down")
        self.CalculateBandwidth("up")
        
        self.flag.clear()

    def MaxRate(self, dir = "up"):
        Read = self.utility.config.Read
        
        if dir == "up":
            if self.utility.torrents["downloading"]:
                # Static overall maximum up rate when downloading
                return Read('maxuploadrate', "float")
            else:
                # Static overall maximum up rate when seeding
                return Read('maxseeduploadrate', "float")
        else:
            return Read('maxdownloadrate', "float")
           
    # See if any torrents are in the "checking existing data"
    # or "allocating space" stages
    def torrentsChecking(self):
        for torrent in self.utility.torrents["active"].keys():
            if torrent.status.isCheckingOrAllocating():
                return True
        return False

    def UploadRateMaximizer(self):
        Read = self.utility.config.Read
               
        # Don't do anything if URM isn't enabled
        if not Read('urm', "boolean"):
            return
        
        # Don't start:
        # - if no torrents are inactive
        # - if any torrents are still in the "checking data" phase
        # - if not enough time has passed after torrents finished checking
        if not self.utility.torrents["inactive"]:
            self.urm_time['under'] = 0
            return
        
        if not self.urm_time['checking'] or self.torrentsChecking():
            self.urm_time['checking'] = time()
            return
            
        # See how long to wait between checking torrents
        delay = Read('urmdelay', "int")
            
        # If a torrent was checking, allow it some time to start up
        # (wait a minimum of 45 seconds for a torrent to start up,
        #  longer if the delay between starting torrents is longer)
        if time() - self.urm_time['checking'] < max(delay, 45):
            return

        # Find the "low" value for upload rate that we're checking for
        lowupthreshold = self.MaxRate("up") - Read('urmupthreshold', "int")
        if lowupthreshold < 0:
            lowupthreshold = 0
        
        uploadrate = self.queue.totals_kb['up']
        
        # Upload rate is below the threshold
        if uploadrate < lowupthreshold or (uploadrate == 0.0 and lowupthreshold == 0.0):
            if self.urm_time['under'] == 0:
                self.urm_time['under'] = time()
            
            # Threshold exceeded for more than urmdelay s ?
            elif time() - self.urm_time['under'] > Read('urmdelay', "int"):
                # Get the next torrent to start
                inactivetorrents = self.utility.queue.getInactiveTorrents(1)
                if not inactivetorrents:
                    return
                
                self.urm_time['under'] = 0
                
                self.utility.actionhandler.procRESUME(inactivetorrents)
                
                self.queue.UpdateRunningTorrentCounters()
        
        # The upload rate is good for now
        else:
            self.urm_time['under'] = 0

##########################################################################################
            
    def CalculateBandwidth(self, dir = "up"):
        if dir == "up":
            workingset = union(self.utility.torrents["downloading"], self.utility.torrents["seeding"])
        else:
            workingset = self.utility.torrents["downloading"]

        # Limit working set to active torrents with connections:
        workingset = [torrent for torrent in workingset if torrent.status.isActive() and torrent.connection.engine.hasConnections]

        # No active file, not need to calculate
        if not workingset:
            return
        
        maxrate = self.MaxRate(dir)
        # See if global rate settings are set to unlimited
        if maxrate == 0:
            # Unlimited rate
            for torrent in workingset:
                torrent.connection.maxrate[dir] = torrent.connection.getLocalRate(dir)
                torrent.connection.setRate(torrent.connection.maxrate[dir], dir)
            return
        
        #print "====================== BEGINNING ALGO ======= th1=%.1f ; th2=%.1f =============================" % (self.calcupth1, self.calcupth2)

        #######################################################
        # - Find number of completed/incomplete torrent
        # - Number of torrent using local setting
        # - bandwidth already used by torrents
        # - Sorting of torrents in lists according to their will in matter of upload rate :
        #   (tobelowered, toberaisedinpriority, toberaised, nottobechanged)
        #######################################################

        # Option set in Preferences/Queue.
        # If not set, torrents with rate local settings are treated like other torrents, except they have their
        # own max rate they can't cross over. The consequence is a behaviour slightly different from the behavior of
        # ABC releases prior to 2.7.0.
        # If set, this gives the algorithm the behaviour of ABC releases prior to 2.7.0 : the torrents with an rate
        # local setting will be granted bandwidth in priority to fulfill their local setting, even is this one
        # is higher than the global max rate setting, and even if this bandwidth must be taken from other active
        # torrents wihout a local rate setting. These torrents will not take part in the rate exchange
        # between active torrents when all bandwidth has been distributed, since they will have been served in priority.
        prioritizelocal = self.utility.config.Read('prioritizelocal', "boolean")

        # torrents with local rate settings when prioritizelocal is set (see prioritizelocal)
        localprioactive = []       
    
        # torrents for which measured rate is lowering and so reserved rate can be lowered
        tobelowered = []
     
        # torrents for which measured rate is growing and so reserved rate can be raised,
        # (with reserved rate > 3 kB/s for uploading)
        toberaised  = []                           

        # torrents for which reserved rate can be raised, with reserved rate < 3 kB/s
        # These will always be raised even there's no available up bandwith, to fulfill the min 3 kB/s rate rule
        toberaisedinpriority = []                             

        # torrents for which reserved rate is not to be changed and is > rateminimum; (for these, the
        # measured rate is between (max upload reserved - calcupth1) and (max upload reserved - calcupth2)
        nottobechanged = []   

        meanrate = 0.0            # mean max rate for torrents to be raised or not to be changed ; it will be used to decide which torrents
                              # must be raised amongst those that want to get higher and that must share the available up bandwidth. The torrents
                              # that don't want to change their rate and that are below 3 kB/s are not taken into account on purpose, because
                              # these ones will never be lowered to give more rate to a torrent that wants to raise its rate, or to
                              # compensate for the rate given to torrents to be raised in priority.

        for torrent in workingset:
            # Active Torrent
            currentrate = torrent.connection.rate[dir]

            maxrate_float = torrent.connection.maxrate[dir]

            # Torrents dispatch
            if prioritizelocal and torrent.connection.getLocalRate(dir, True):
                localprioactive.append(torrent)
            elif currentrate < 0.05:
                # These are the torrents that don't want to have their rate changed and that have an allmost null rate
                # They will not go lower, we can reset their reserved rate until they want to get higher
                torrent.connection.maxrate[dir] = 0.0
            elif maxrate_float - currentrate > self.calcupth1:
                tobelowered.append(torrent)
            elif maxrate_float - currentrate <= self.calcupth2:
                if currentrate < self.rateminimum[dir]:
                    toberaisedinpriority.append(torrent)
                else:
                    toberaised.append(torrent)
                    meanrate += maxrate_float
            elif currentrate > self.rateminimum[dir]:
                nottobechanged.append(torrent)
                meanrate += maxrate_float