NovaBryan/EVE_Dump
Log | Files | 📦 下载最新源码 (.zip) | Refs | README

loadService.py (11257B)

      1 #Embedded file name: C:\BuildAgent\work\20d73f1b24eed06e\eve\release\V24.01\carbon\common\script\sys\loadService.py
      2 from __future__ import absolute_import
      3 import itertools
      4 import math
      5 import random
      6 from time import clock
      7 import blue
      8 import uthread
      9 from carbon.common.lib import const
     10 from carbon.common.script.sys.service import Service
     11 from carbon.common.script.sys.serviceConst import ROLE_PROGRAMMER, ROLE_SERVICE
     12 from test.pystone import pystones
     13 
     14 class LoadService(Service):
     15     __guid__ = 'svc.loadService'
     16     __displayname__ = 'Load Service'
     17     __exportedcalls__ = {'Ping': [ROLE_SERVICE | ROLE_PROGRAMMER],
     18      'GetTotalStats': [ROLE_SERVICE | ROLE_PROGRAMMER],
     19      'GatherStats': [ROLE_SERVICE | ROLE_PROGRAMMER],
     20      'StartLoad': [ROLE_PROGRAMMER],
     21      'StopLoad': [ROLE_SERVICE | ROLE_PROGRAMMER],
     22      'StartLoadOnAllNodes': [ROLE_SERVICE | ROLE_PROGRAMMER],
     23      'StopLoadOnAllNodes': [ROLE_SERVICE | ROLE_PROGRAMMER],
     24      'GetConfig': [ROLE_PROGRAMMER],
     25      'SetConfig': [ROLE_SERVICE | ROLE_PROGRAMMER],
     26      'IsRunning': [ROLE_SERVICE | ROLE_PROGRAMMER],
     27      'RemainingTime': [ROLE_SERVICE | ROLE_PROGRAMMER],
     28      'Calibrate': [ROLE_SERVICE | ROLE_PROGRAMMER]}
     29     __dependencies__ = ['machoNet']
     30     __notifyevents__ = []
     31 
     32     def __init__(self):
     33         Service.__init__(self)
     34         self.configVariables = ['pyStoneTasklets',
     35          'pyStonesPerSec',
     36          'pyStonesPerUnit',
     37          'netTasklets',
     38          'packetsP2S',
     39          'packetsS2P',
     40          'packetsS2S']
     41         self.pyStoneTasklets = 10
     42         self.pyStonesPerSec = 20000
     43         self.pyStonesPerUnit = 500
     44         self.netTasklets = 5
     45         self.packetsP2S = 150
     46         self.packetsS2P = 75
     47         self.packetsS2S = 50
     48         self.running = False
     49         self.ResetCounters()
     50 
     51     def GetConfig(self):
     52         return dict([ (v, getattr(self, v)) for v in self.configVariables ])
     53 
     54     def SetConfig(self, conf, updateOtherNodes = True):
     55         for v in self.configVariables:
     56             if v in conf:
     57                 setattr(self, v, conf[v])
     58 
     59         if updateOtherNodes:
     60             for node in self.machoNet.GetConnectedNodes():
     61                 self.session.ConnectToRemoteService('loadService', node).SetConfig(conf, False)
     62 
     63     def IsRunning(self):
     64         return self.running
     65 
     66     def RemainingTime(self):
     67         if self.duration is not None:
     68             return self.duration - (blue.os.GetWallclockTime() * 1e-07 - self.startTime)
     69 
     70     def StartLoadOnAllNodes(self, duration = None):
     71         self.session.ConnectToAllServices('loadService').StartLoad(duration)
     72 
     73     def StopLoadOnAllNodes(self):
     74         self.session.ConnectToAllServices('loadService').StopLoad()
     75 
     76     def TaskletWrap(self, func):
     77 
     78         def Wrapped(*args, **kwds):
     79             try:
     80                 func(*args, **kwds)
     81             finally:
     82                 self.nTasklets -= 1
     83                 blue.pyos.synchro.SleepWallclock(0)
     84 
     85         self.nTasklets += 1
     86         return Wrapped
     87 
     88     def StartLoad(self, duration = None):
     89         self.running = True
     90         self.startProcessTime = blue.pyos.taskletTimer.GetProcessTimes()
     91         self.startThreadTime = blue.pyos.taskletTimer.GetThreadTimes()
     92         self.ResetCounters()
     93         self.nTasklets = 0
     94         for t in range(self.pyStoneTasklets):
     95             uthread.worker('loadServiceService::CpuLoad::%s' % t, self.TaskletWrap(self.CpuCycleWorker))
     96 
     97         for t in range(self.netTasklets):
     98             uthread.worker('loadServiceService::NetworkTraffic::%s' % t, self.TaskletWrap(self.Network))
     99 
    100         self.startTime = blue.os.GetWallclockTime() * 1e-07
    101         self.duration = duration
    102         if duration:
    103             uthread.worker('loadServiceService::ShutdownCounter', self.ShutdownCounter, duration)
    104 
    105     def ShutdownCounter(self, duration):
    106         blue.pyos.synchro.SleepWallclock(int(duration * 1000))
    107         self.StopLoad()
    108 
    109     def StopLoad(self):
    110         self.running = False
    111         while self.nTasklets:
    112             self.LogInfo('Waiting for %d tasklets to finish' % self.nTasklets)
    113             blue.pyos.synchro.Yield()
    114 
    115         self.LogInfo('All tasklets finished')
    116         endProcessTime = blue.pyos.taskletTimer.GetProcessTimes()
    117         endThreadTime = blue.pyos.taskletTimer.GetThreadTimes()
    118         for k, v in self.startProcessTime.iteritems():
    119             endProcessTime[k] -= v
    120 
    121         for k, v in self.startThreadTime.iteritems():
    122             endThreadTime[k] -= v
    123 
    124         self.cpuStats['process'] = endProcessTime
    125         self.cpuStats['thread'] = endThreadTime
    126 
    127     def IntervalDispatcher(self, interval, stats, func):
    128 
    129         def times(dt):
    130             start = clock()
    131             for i in itertools.count():
    132                 yield start + (i + random.random()) * dt
    133 
    134         loopCount = 0
    135         slowCount = 0
    136         if interval:
    137             i = times(interval)
    138             while self.running:
    139                 now = clock()
    140                 when = i.next()
    141                 ms = int((when - now) * 1000)
    142                 if ms > 0:
    143                     blue.pyos.synchro.SleepWallclock(ms)
    144                     now = clock()
    145                 if now > when + interval:
    146                     slowCount += 1
    147                     while now > when + interval:
    148                         when = i.next()
    149 
    150                 func()
    151                 loopCount += 1
    152 
    153         else:
    154             while self.running:
    155                 func()
    156                 loopCount += 1
    157 
    158         stats['loopCount'] += loopCount
    159         stats['slowCount'] += slowCount
    160         self.LogInfo('IntervalDispatcher finished. %s of %s %s calls were slow (%s%%)' % (slowCount,
    161          loopCount,
    162          func.__name__,
    163          int(slowCount / float(loopCount) * 100)))
    164 
    165     def Calibrate(self, duration = 2):
    166         startTime = blue.os.GetWallclockTimeNow()
    167         stepCount = 0
    168         while blue.os.GetWallclockTimeNow() - startTime < duration * 10000000:
    169             stepCount += 1
    170             pystones(self.pyStonesPerUnit)
    171 
    172         return int(stepCount * self.pyStonesPerUnit / duration)
    173 
    174     def CpuCycleWorker(self):
    175         taskletLoad = float(self.pyStonesPerSec) / self.pyStoneTasklets
    176         interval = 1.0 / (taskletLoad / self.pyStonesPerUnit)
    177         self.LogInfo('CPU Tasklet. Will run %s pyStones every %ss' % (self.pyStonesPerUnit, interval))
    178         blue.pyos.synchro.SleepWallclock(int(random.random() * interval * 1000))
    179 
    180         def CpuWork():
    181             pystones(self.pyStonesPerUnit)
    182 
    183         self.IntervalDispatcher(interval, self.cpuStats, CpuWork)
    184 
    185     def Network(self):
    186         if self.machoNet.GetNodeID() > const.maxNodeID:
    187             solPackets = float(self.packetsP2S) / self.netTasklets
    188             proxyPackets = 0
    189         else:
    190             proxyPackets = float(self.packetsS2P) / self.netTasklets
    191             solPackets = float(self.packetsS2S) / self.netTasklets
    192         if solPackets:
    193             nodes = self.machoNet.GetConnectedSolNodes()
    194             nodes = [ (node, self.session.ConnectToRemoteService('loadService', node)) for node in nodes ]
    195             randomNodes = self.NodeRandomizer(nodes)
    196 
    197             def func():
    198                 self.NetworkSendPacket(randomNodes)
    199 
    200             interval = 1.0 / solPackets if solPackets > 0.0 else 0.0
    201             uthread.worker('loadServiceService::NetworkTraffic::ToSolServers', self.TaskletWrap(self.IntervalDispatcher), interval, self.netStats, func)
    202         if proxyPackets:
    203             nodes = self.machoNet.GetConnectedProxyNodes()
    204             nodes = [ (node, self.session.ConnectToRemoteService('loadService', node)) for node in nodes ]
    205             randomNodes = self.NodeRandomizer(nodes)
    206 
    207             def func():
    208                 self.NetworkSendPacket(randomNodes)
    209 
    210             interval = 1.0 / proxyPackets if proxyPackets > 0.0 else 0.0
    211             uthread.worker('loadServiceService::NetworkTraffic::ToProxyServers', self.TaskletWrap(self.IntervalDispatcher), interval, self.netStats, func)
    212 
    213     def NetworkSendPacket(self, randomNodes):
    214         PACKET_SIZE = 20
    215         PACKET_VAR = 2
    216         packet = range(int(random.gauss(PACKET_SIZE, PACKET_VAR)))
    217         try:
    218             node, conn = randomNodes.next()
    219         except StopIteration:
    220             self.LogWarn('No nodes returned from randomNodes')
    221             return
    222 
    223         packetSendTime = blue.os.GetWallclockTimeNow()
    224         conn.Ping(packet)
    225         packetReceiveTime = blue.os.GetWallclockTimeNow()
    226         self.UpdateNetworkStats(node, packetReceiveTime - packetSendTime)
    227 
    228     def NodeRandomizer(self, nodes):
    229         if not nodes:
    230             return
    231         nodes = nodes[:]
    232         while True:
    233             random.shuffle(nodes)
    234             for node in nodes:
    235                 yield node
    236 
    237     def Ping(self, packet):
    238         return packet
    239 
    240     def ResetCounters(self):
    241         self.cpuStats = {'loopCount': 0,
    242          'slowCount': 0,
    243          'pyStonesPerSec': 0}
    244         self.netStats = {'loopCount': 0,
    245          'slowCount': 0}
    246         self.totalStats = {}
    247 
    248     def UpdateNetworkStats(self, node, latency):
    249         if node not in self.netStats:
    250             self.netStats[node] = {}
    251             self.netStats[node]['totalPackets'] = 0
    252             self.netStats[node]['totalLatency'] = 0
    253             self.netStats[node]['totalLatencySq'] = 0
    254             self.netStats[node]['maxLatency'] = None
    255             self.netStats[node]['minLatency'] = None
    256         stats = self.netStats[node]
    257         latency *= 1e-07
    258         stats['totalPackets'] += 1
    259         stats['totalLatency'] += latency
    260         stats['totalLatencySq'] += latency * latency
    261         if stats['maxLatency'] is None:
    262             stats['maxLatency'] = stats['minLatency'] = latency
    263         elif latency > stats['maxLatency']:
    264             stats['maxLatency'] = latency
    265         elif latency < stats['minLatency']:
    266             stats['minLatency'] = latency
    267 
    268     def GetNetworkStats(self):
    269         for node in self.netStats:
    270             if node in ('loopCount', 'slowCount'):
    271                 continue
    272             self.netStats[node]['average'] = self.netStats[node]['totalLatency'] / self.netStats[node]['totalPackets']
    273             self.netStats[node]['stdev'] = math.sqrt((self.netStats[node]['totalLatencySq'] - self.netStats[node]['totalLatency'] * self.netStats[node]['average']) / self.netStats[node]['totalPackets'])
    274 
    275         return self.netStats
    276 
    277     def GetCpuStats(self):
    278         for stats in self.cpuStats.values():
    279             if type(stats) is not dict:
    280                 continue
    281             if 'cpu' in stats and 'wallclock' in stats:
    282                 stats['cpuPercentage'] = float(stats['cpu']) / stats['wallclock'] * 100
    283 
    284         if 'loopCount' in self.cpuStats and 'process' in self.cpuStats:
    285             self.cpuStats['pyStonesPerSec'] = self.pyStonesPerUnit * self.cpuStats['loopCount'] / (self.cpuStats['process']['wallclock'] * 1e-07)
    286         return self.cpuStats
    287 
    288     def GetTotalStats(self):
    289         return {'cpu': dict(self.GetCpuStats()),
    290          'net': dict(self.GetNetworkStats())}
    291 
    292     def GatherStats(self):
    293         for node in self.machoNet.GetConnectedNodes():
    294             self.totalStats[node] = self.machoNet.ConnectToRemoteService('loadService', node).GetTotalStats()
    295 
    296         self.totalStats[self.machoNet.GetNodeID()] = self.GetTotalStats()
    297         return self.totalStats