serviceManager.py (41973B)
1 #Embedded file name: C:\BuildAgent\work\20d73f1b24eed06e\eve\release\V24.01\carbon\common\script\sys\serviceManager.py 2 from __future__ import print_function 3 from future.utils import raise_ 4 import __builtin__ 5 import sys 6 import numbers 7 import threading 8 import time 9 import types 10 import weakref 11 from collections import defaultdict 12 import blue 13 import bluepy 14 import carbon.common.script.sys.basesession as base 15 import log 16 import monolithmetrics 17 import stackless 18 import telemetry 19 import uthread 20 from caching.memoize import Memoize 21 from carbon.common.script.net.machobase import GetLogName as GetMachoLogName 22 from carbon.common.script.sys.service import Service 23 from carbon.common.script.sys.serviceConst import SERVICE_FAILED, SERVICE_RUNNING, SERVICE_STARTING_DEPENDENCIES, SERVICE_START_PENDING, SERVICE_STOPPED, SERVICE_STOP_PENDING 24 from carbon.common.script.util import logUtil 25 from carbon.common.script.util.commonutils import IsFullLogging 26 from carbon.common.script.util.timerstuff import ClockThisWithoutTheStars 27 from eveexceptions import MethodNotCalledFromClient, ServiceNotFound 28 from eveprefs import boot, prefs 29 from utillib import DAG 30 31 def ObjectShouldReceiveMessages(obj): 32 return not isinstance(obj, Service) or obj.state == SERVICE_RUNNING 33 34 35 class ServiceManager(logUtil.LogMixin): 36 __guid__ = 'service.ServiceManager' 37 __instance = None 38 __init_lock = threading.Semaphore() 39 40 @staticmethod 41 def Instance(): 42 if ServiceManager.__instance is None: 43 raise RuntimeError('ServiceManager not initialized') 44 return ServiceManager.__instance 45 46 def __init__(self, startInline = []): 47 with ServiceManager.__init_lock: 48 if ServiceManager.__instance is not None: 49 raise RuntimeError('Multiple instances of ServiceManager are not allowed in a process') 50 logUtil.LogMixin.__init__(self, 'svc.ServiceManager') 51 self.state = SERVICE_START_PENDING 52 self.services = {} 53 self.dependants = {} 54 self.notify = defaultdict(list) 55 self.notifyObs = defaultdict(list) 56 self.startInline = startInline 57 self.blockedServices = [] 58 self.startupTimes = {} 59 blue.pyos.scatterEvent = self.ScatterEvent 60 blue.pyos.sendEvent = self.SendEvent 61 blue.pyos.chainEvent = self.ChainEvent 62 ServiceManager.__instance = __builtin__.sm = self 63 64 def _BuildClassMap(self): 65 import svc 66 self.classmap = {name:(name, getattr(svc, name)) for name in dir(svc)} 67 self.classmapWithReplacements = self.classmap.copy() 68 for name, svcclass in self.classmap.values(): 69 self.classmapWithReplacements[name] = (name, getattr(svc, self.GetServiceImplementation(name))) 70 71 def _CreateDependencyGraph(self): 72 dag = DAG() 73 for k, v in self.services.iteritems(): 74 depends = v.__dependencies__ + getattr(v, '__startupdependencies__', []) 75 for d in depends: 76 if type(d) is not str: 77 d = d[0] 78 if d in self.services: 79 dag.InsertEdge(k, d) 80 81 dag.Invert() 82 return dag 83 84 def SaveDependencyGraph(self, filename = None): 85 try: 86 dag = self._CreateDependencyGraph() 87 if filename is None: 88 import os 89 filename = '%s/service-dependencies-%s.dot' % (os.getcwd(), boot.role) 90 with open(filename, 'w') as fp: 91 fp.write(dag.ToDot()) 92 except Exception: 93 log.LogException('Failed saving service dependency graph.') 94 95 def Run(self, servicesToRun, servicesToBlock = []): 96 self.run = 1 97 if self.state not in (SERVICE_START_PENDING, SERVICE_RUNNING): 98 if self.state == SERVICE_STOPPED: 99 self.state = SERVICE_START_PENDING 100 else: 101 raise_(RuntimeError, "can't run a service when state is " + repr(self.state)) 102 self._BuildClassMap() 103 blue.pyos.AddExitProc(self.Stop) 104 for block in servicesToBlock: 105 if block not in self.blockedServices: 106 self.blockedServices.append(block) 107 108 if len(servicesToRun): 109 print('Starting services') 110 was = blue.pyos.taskletTimer.timesliceWarning 111 try: 112 blue.pyos.taskletTimer.timesliceWarning = 5000 113 for each in servicesToRun: 114 self.StartService(each, reason='autoexec') 115 116 for each in self.services.keys(): 117 counter = 0 118 sleepTimeMs = 1 119 while self.services[each].state == SERVICE_START_PENDING: 120 counter += 1 121 if counter % 100 == 0: 122 print("Service start still pending: '%s', sleeping..." % self.services[each].__guid__) 123 blue.pyos.synchro.SleepWallclock(sleepTimeMs) 124 sleepTimeMs *= 2 125 if sleepTimeMs > 100: 126 sleepTimeMs = 100 127 128 for each in self.services.keys(): 129 if hasattr(self.services[each], 'PostRun'): 130 self.services[each].PostRun() 131 132 finally: 133 blue.pyos.taskletTimer.timesliceWarning = was 134 135 print('Starting services - Done') 136 self.state = SERVICE_RUNNING 137 138 def Stop(self): 139 self.logChannel.Log('ServiceManager.Stop(), stopping services') 140 dag = self._CreateDependencyGraph() 141 self.logChannel.Log('== BEGIN SERVICE LIST ==') 142 order = [] 143 while dag.graph: 144 leaves = dag.Leaves() 145 if not leaves: 146 break 147 order.extend(leaves) 148 for l in leaves: 149 dag.RemoveNode(l) 150 151 self.logChannel.Log('==leaves==') 152 self.logChannel.Log(','.join(leaves)) 153 154 if dag.graph: 155 leaves = dag.graph.keys() 156 order.extend(leaves) 157 self.logChannel.Log('==nonleaves==') 158 self.logChannel.Log(','.join(leaves)) 159 self.logChannel.Log('== ENDOF SERVICE LIST ==') 160 self.run = 0 161 import blue 162 old_block_trap = stackless.getcurrent().block_trap 163 stackless.getcurrent().block_trap = 1 164 self.state = SERVICE_STOP_PENDING 165 try: 166 for k in order: 167 try: 168 v = self.services[k] 169 except KeyError: 170 self.LogWarn('Warning, the service', k, 'has been stopped already but something that depends on it might still be running!') 171 sys.exc_clear() 172 continue 173 174 if not hasattr(v, 'state'): 175 self.logChannel.Log("ServiceManager.Stop(), service '" + str(k) + " doesn't have state therefore has already stopped") 176 elif v.state not in (SERVICE_STOPPED, SERVICE_STOP_PENDING): 177 self.logChannel.Log("ServiceManager.Stop(), stopping service '" + str(k) + "'") 178 try: 179 v.state = SERVICE_STOP_PENDING 180 for notify in v.__notifyevents__: 181 self.notify[notify].remove(v) 182 183 v.Stop(blue.MemStream()) 184 except Exception: 185 log.LogException() 186 sys.exc_clear() 187 finally: 188 v.state = SERVICE_STOPPED 189 190 else: 191 self.logChannel.Log("ServiceManager.Stop(), service '" + str(k) + "' is already stopping") 192 193 finally: 194 stackless.getcurrent().block_trap = old_block_trap 195 self.state = SERVICE_STOPPED 196 197 for v in self.services.itervalues(): 198 for a in v.__dict__.keys(): 199 if a not in ('logChannel', 'logContexts', 'state'): 200 delattr(v, a) 201 202 self.logChannel.Log('ServiceManager.Stop(), services stopped.') 203 204 def ParseServiceClass(self, serviceName): 205 return 'svc.' + serviceName 206 207 def Reload(self, serviceNames): 208 self._BuildClassMap() 209 if not serviceNames: 210 return 211 self.LogWarn('Reloading services: %s' % ', '.join(serviceNames)) 212 for each in serviceNames: 213 if each not in self.services: 214 continue 215 if getattr(self.services[each], '__update_on_reload__', True): 216 continue 217 ms = blue.MemStream() 218 try: 219 try: 220 self.StopService(each, 0, ms) 221 finally: 222 if each in self.services: 223 del self.services[each] 224 225 self.GetService(each, ms) 226 except Exception: 227 log.LogException('Trying to reload service: %s' % each) 228 sys.exc_clear() 229 230 self.LogWarn('Reloading services - Done') 231 232 def StartServiceAndWaitForRunningState(self, serviceName, ms = None, reason = None): 233 srv = self.StartService(serviceName, ms=None, reason=reason) 234 desiredStates = (SERVICE_RUNNING,) 235 errorStates = (SERVICE_FAILED, SERVICE_STOPPED) 236 self.WaitForServiceObjectState(srv, desiredStates, errorStates) 237 return srv 238 239 @telemetry.ZONE_METHOD 240 def WaitForServiceObjectState(self, svc, desiredStates, errorStates = (SERVICE_FAILED,)): 241 i = 0 242 sleepTimeMs = 1 243 while svc.state not in desiredStates: 244 if svc.state in errorStates: 245 svc.LogError('Service ', svc.__logname__, ' got in an unexpected state', 'raising error') 246 if svc.__error__: 247 raise_(svc.__error__[1], None, svc.__error__[2]) 248 else: 249 raise_(RuntimeError, 'Service %s made unexpected state transition' % svc.__logname__) 250 blue.pyos.synchro.SleepWallclock(sleepTimeMs) 251 sleepTimeMs *= 2 252 if sleepTimeMs > 100: 253 sleepTimeMs = 100 254 if i % 600 == 0 and i > 0: 255 svc.LogWarn('WaitForServiceObjectState has been sleeping for a long time waiting for ', svc.__logname__, ' to either get to state ', desiredStates, 'current state is', svc.state) 256 i += 1 257 258 def GetServiceIfStarted(self, serviceName): 259 if serviceName in self.services: 260 srv = self.services[serviceName] 261 self.WaitForServiceObjectState(srv, (SERVICE_RUNNING,)) 262 return srv 263 264 def GetServiceIfRunning(self, serviceName): 265 if self.IsServiceRunning(serviceName): 266 srv = self.services[serviceName] 267 return srv 268 269 def GetService(self, serviceName, ms = None): 270 srv = self.services.get(serviceName, None) 271 if srv and srv.state == SERVICE_RUNNING: 272 return srv 273 if serviceName in self.services: 274 if self.services[serviceName].state in (SERVICE_START_PENDING, SERVICE_STARTING_DEPENDENCIES): 275 log.LogTraceback('Possible service deadlock detected!', toAlertSvc=False, severity=log.LGWARN) 276 return self.StartServiceAndWaitForRunningState(serviceName, ms, reason='GetService') 277 278 @telemetry.ZONE_METHOD 279 def StartService(self, serviceName, ms = None, reason = 'StartService'): 280 telemetry.APPEND_TO_ZONE(serviceName) 281 srv = self.services.get(serviceName, None) 282 if srv and srv.state == SERVICE_RUNNING: 283 return srv 284 if serviceName in self.services: 285 srv = self.services[serviceName] 286 else: 287 if serviceName in self.blockedServices: 288 raise RuntimeError('%s has been blocked from running on this system' % serviceName) 289 srv = self.CreateServiceInstance(serviceName) 290 self.services[serviceName] = srv 291 if srv.state in (SERVICE_START_PENDING,): 292 return srv 293 if srv.state == SERVICE_STARTING_DEPENDENCIES: 294 desiredStates = (SERVICE_START_PENDING, SERVICE_RUNNING) 295 errorStates = (SERVICE_FAILED, SERVICE_STOPPED) 296 self.WaitForServiceObjectState(srv, desiredStates, errorStates) 297 return srv 298 if self.state in (SERVICE_STOP_PENDING, SERVICE_STOPPED): 299 raise_(RuntimeError, "Can't start service " + serviceName + ' when service manager is shutting down') 300 if 'machoNet' in self.services and self.services['machoNet'].IsClusterShuttingDown(): 301 log.LogTraceback("Attempt to start service '%s' during cluster shutdown" % serviceName) 302 raise RuntimeError("Cannot start service '%s' during cluster shutdown" % serviceName) 303 if srv.state == SERVICE_FAILED: 304 return srv 305 try: 306 r = reason 307 if reason in ('GetService', 'StartService'): 308 up = 4 309 if reason == 'StartService': 310 up = 2 311 r = '%s - called from %s' % (reason, log.WhoCalledMe(up)) 312 self.LogInfo('Starting', serviceName, '. Reason:', r) 313 except: 314 pass 315 316 srv.state = SERVICE_STARTING_DEPENDENCIES 317 srv.__error__ = None 318 try: 319 self.dependants[serviceName] = [] 320 if srv.__startupdependencies__: 321 self.LogInfo('starting startup dependencies for %s, which are: %s' % (serviceName, str(srv.__startupdependencies__))) 322 for each in srv.__startupdependencies__: 323 if each == srv.__guid__.split('.')[1]: 324 self.LogError('Found a service with a dependancy on it self:', each, '. The service reference will not be assigned, things will probaly blow up') 325 continue 326 if type(each) is str: 327 each = (each, each) 328 depname, asname = each 329 if not self.IsServiceRunning(depname): 330 self.LogInfo(serviceName, 'is waiting while', depname, 'is started') 331 depService = self.StartServiceAndWaitForRunningState(depname, reason='startup dependency for %s' % serviceName) 332 self.dependants[depname].append(serviceName) 333 if getattr(boot, 'replaceDependencyServiceWrappers', 'false').lower() != 'true' or not depService.IsRunning(): 334 setattr(srv, asname, srv.session.ConnectToService(depname)) 335 else: 336 setattr(srv, asname, depService) 337 338 srv.state = SERVICE_START_PENDING 339 if srv.__dependencies__: 340 uthread.new(self._LoadServiceDependenciesAsych, srv, serviceName).context = serviceName + ' _LoadServiceDependenciesAsych' 341 for notify in srv.__notifyevents__: 342 if not hasattr(srv, notify): 343 raise RuntimeError('MissingSvcExportAttribute', serviceName, 'notify', notify) 344 self.notify[notify].append(srv) 345 346 except Exception as e: 347 srv.state = SERVICE_FAILED 348 srv.__error__ = sys.exc_info() 349 raise 350 351 if ms: 352 ms.Seek(0) 353 args = (ms,) 354 if serviceName in self.startInline: 355 self.StartServiceRun(srv, args, serviceName) 356 else: 357 uthread.pool(serviceName + ' StartServiceRun', self.StartServiceRun, srv, args, serviceName, stackless_tracing_enabled=False) 358 return srv 359 360 def _LoadServiceDependenciesAsych(self, srv, serviceName): 361 self.LogInfo('starting dependencies for %s, which are: %s' % (serviceName, str(srv.__dependencies__))) 362 for each in srv.__dependencies__: 363 if type(each) is str: 364 each = (each, each) 365 depname, asname = each 366 depService = self.StartService(depname, reason='dependency for %s' % serviceName) 367 self.dependants[depname].append(serviceName) 368 if getattr(boot, 'replaceDependencyServiceWrappers', 'false').lower() != 'true' or not depService.IsRunning(): 369 setattr(srv, asname, srv.session.ConnectToService(depname)) 370 else: 371 setattr(srv, asname, depService) 372 373 @Memoize 374 def GetServiceImplementation(self, target): 375 found = target 376 foundPriority = -1 377 for name, svcclass in self.classmap.itervalues(): 378 replaces = getattr(svcclass, '__replaceservice__', None) 379 if replaces == target: 380 priority = getattr(svcclass, '__replacepriority__', 0) 381 if priority > foundPriority: 382 found = name 383 foundPriority = priority 384 385 return found 386 387 @telemetry.ZONE_METHOD 388 def CreateServiceInstance(self, serviceName): 389 old_block_trap = stackless.getcurrent().block_trap 390 stackless.getcurrent().block_trap = 1 391 try: 392 try: 393 createName, createClass = self.classmapWithReplacements[serviceName] 394 except KeyError: 395 self._BuildClassMap() 396 try: 397 createName, createClass = self.classmapWithReplacements[serviceName] 398 except KeyError: 399 raise ServiceNotFound(serviceName) 400 401 if createName != serviceName: 402 print('Replacing service %r with %r' % (serviceName, createName)) 403 replaceService = getattr(createClass, '__replaceservice__', None) 404 if replaceService is not None and replaceService != serviceName: 405 raise RuntimeError('Must not start %s directly as it replaces %s' % (serviceName, replaceService)) 406 srv = createClass() 407 if not isinstance(srv, Service): 408 raise RuntimeError('Service name %r does not resolve to a service class (%r)' % (serviceName, createClass)) 409 srv.__servicename__ = serviceName 410 srv.session = base.GetServiceSession(serviceName) 411 self.VerifyServiceExports(srv, serviceName) 412 return srv 413 finally: 414 stackless.getcurrent().block_trap = old_block_trap 415 416 def VerifyServiceExports(self, srv, serviceName): 417 for funcName, paramList in srv.__exportedcalls__.iteritems(): 418 if not hasattr(srv, funcName): 419 raise RuntimeError('MissingSvcExportAttribute', serviceName, 'exported call', funcName) 420 if type(paramList) == types.ListType: 421 tmp = {} 422 if len(paramList): 423 tmp['role'] = paramList[0] 424 tmp['preargs'] = paramList[1:] 425 paramList = tmp 426 if type(paramList) == types.DictType: 427 for k, v in paramList.iteritems(): 428 if k not in ('role', 'caching', 'preargs', 'fastcall', 'precall', 'postcall', 'callhandlerargs', 'input', 'resolve') or k == 'role' and not isinstance(v, numbers.Integral): 429 self.LogError('Service %s has illegal function declaration for %s: %s is not a valid metadata key' % (serviceName, funcName, k)) 430 elif k == 'fastcall': 431 if int(v) not in (0, 1): 432 self.LogError('Service %s has illegal function declaration for %s: %s is not a valid setting for fastcall' % (serviceName, funcName, v)) 433 elif k == 'preargs': 434 for eachArg in v: 435 if type(eachArg) != types.StringType or eachArg not in srv.session.__persistvars__ and eachArg not in srv.session.__nonpersistvars__: 436 self.LogError('Service %s has illegal function declaration for %s: %s is not a valid prearg' % (serviceName, funcName, eachArg)) 437 438 elif k == 'caching' and 'objectCaching' in self.services: 439 if not v: 440 self.LogError('Service %s has illegal function declaration for %s: caching must have subinfo' % (serviceName, funcName)) 441 for k2, v2 in v.iteritems(): 442 if k2 not in ('sessionInfo', 'versionCheck', 'client', 'server', 'proxy'): 443 self.LogError('Service %s has illegal function declaration for %s: %s is not a valid caching subinfo key' % (serviceName, funcName, k2)) 444 elif k2 == 'sessionInfo' and v2 not in srv.session.__persistvars__: 445 self.LogError('Service %s has illegal function declaration for %s: %s is not a valid session info prearg' % (serviceName, funcName, v2)) 446 447 else: 448 self.LogError('Service %s has illegal function declaration for %s: %s is not a valid metadata info type' % (serviceName, funcName, type(paramList))) 449 450 @telemetry.ZONE_METHOD 451 def StartServiceRun(self, svc, args, namen): 452 try: 453 t0 = time.clock() 454 try: 455 if boot.role == 'client' and getattr(prefs, 'clientServicesWait', 'true').lower() != 'true': 456 svc.state = SERVICE_RUNNING 457 svc.Run(*args) 458 else: 459 svc.state = SERVICE_START_PENDING 460 with bluepy.Timer('StartService::ServiceStartRun::' + namen): 461 svc.Run(*args) 462 svc.state = SERVICE_RUNNING 463 if getattr(boot, 'replaceDependencyServiceWrappers', 'false').lower() == 'true': 464 for depName in self.dependants[namen]: 465 depSvc = self.StartService(depName) 466 setattr(depSvc, namen, svc) 467 468 finally: 469 t = time.clock() - t0 470 471 except Exception as e: 472 svc.state = SERVICE_FAILED 473 svc.__error__ = sys.exc_info() 474 self.LogError('Failed to start service %s' % namen) 475 raise 476 477 self.startupTimes[namen] = t 478 if t < 60: 479 svc.LogInfo('Service', namen, 'required %-3.3f' % t, ' seconds to startup') 480 else: 481 svc.LogWarn('Service', namen, 'startup took %-3.3f' % t, ' seconds') 482 serviceStartupTimeTxt = 'Service %-25s Startup time: %-3.3fs' % (namen, t) 483 print(serviceStartupTimeTxt) 484 self.LogNotice(serviceStartupTimeTxt) 485 486 def StopService(self, serviceName, halt = 1, ms = None): 487 self.LogInfo('Stopping %s' % serviceName) 488 if serviceName not in self.services: 489 return 490 srv = self.services[serviceName] 491 if srv.state in (SERVICE_STOP_PENDING, SERVICE_STOPPED): 492 return 493 oldstate = srv.state 494 srv.state = SERVICE_STOP_PENDING 495 for notify in srv.__notifyevents__: 496 self.notify[notify].remove(srv) 497 498 if halt: 499 for each in self.dependants[serviceName]: 500 self.StopService(each) 501 502 try: 503 srv.Stop(ms) 504 except Exception: 505 srv.state = oldstate 506 log.LogException('Trying to stop service %s ' % serviceName) 507 sys.exc_clear() 508 509 if srv.state != SERVICE_STOPPED: 510 srv.state = SERVICE_STOPPED 511 del self.services[serviceName] 512 513 def GetDependecyGraph(self, startupDependencies = False): 514 depGraph = DAG() 515 import svc 516 depAttr = '__dependencies__' 517 if startupDependencies: 518 depAttr = '__startupdependencies__' 519 for k, v in svc.__dict__.items(): 520 if hasattr(v, '__guid__'): 521 depGraph.AddNode(k) 522 for dep in getattr(v, depAttr, []): 523 depGraph.InsertEdge(k, dep) 524 525 return depGraph 526 527 def IsServiceRunning(self, serviceName): 528 return serviceName in self.services and self.services[serviceName].IsRunning() 529 530 def RemoteSvc(self, serviceName): 531 if boot.role == 'client': 532 return session.ConnectToRemoteService(serviceName) 533 self.LogError('The method sm.RemoteSvc can be called from the client only.') 534 raise MethodNotCalledFromClient('sm.RemoteSvc') 535 536 def ProxySvc(self, serviceName): 537 if boot.role == 'client': 538 return session.ConnectToRemoteService(serviceName, self.services['machoNet'].myProxyNodeID) 539 self.LogError('The method sm.ProxySvc can be called from the client only.') 540 raise MethodNotCalledFromClient('sm.ProxySvc') 541 542 def GetActiveServices(self): 543 ret = [] 544 for k, srv in self.services.iteritems(): 545 if srv.state == SERVICE_RUNNING: 546 ret.append(k) 547 548 return ret 549 550 def GetServicesState(self): 551 ret = {} 552 for k, srv in self.services.iteritems(): 553 ret[k] = srv.state 554 555 return ret 556 557 def GetLogName(self, thingie): 558 if isinstance(thingie, weakref.ref): 559 thingie = thingie() 560 return getattr(thingie, '__logname__', GetMachoLogName(thingie)) 561 562 def SendEvent(self, eventid, *args, **keywords): 563 return self.SendEventWithoutTheStars(eventid, args, keywords) 564 565 def SendEventWithoutTheStars(self, eventid, args, keywords = None): 566 if keywords is None: 567 keywords = {} 568 if not eventid.startswith('Do'): 569 self.LogError('SendEvent called with event ', eventid, ". All events sent via SendEvent should start with 'Do'") 570 self.LogError("Not only is the programmer responsible for this a 10z3r, but he wears his mother's underwear as well") 571 log.LogTraceback() 572 if eventid not in self.notify and eventid not in self.notifyObs: 573 self.LogInfo("Orphan'd event. ", eventid, "doesn't have any listeners") 574 shouldLogMethodCalls = self.ShouldLogMethodCalls() 575 if shouldLogMethodCalls: 576 if IsFullLogging(): 577 self.LogMethodCall('SendEvent(', eventid, ',*args=', args, ',**kw=', keywords, ')') 578 else: 579 self.LogMethodCall('SendEvent(', eventid, ')') 580 prefix = blue.pyos.taskletTimer.GetCurrent() + '::SendEvent_' + eventid + '::' 581 old_block_trap = stackless.getcurrent().block_trap 582 stackless.getcurrent().block_trap = 1 583 ret = [] 584 try: 585 for srv in self.notify.get(eventid, []): 586 try: 587 logname = prefix + self.GetLogName(srv) 588 if ObjectShouldReceiveMessages(srv): 589 if shouldLogMethodCalls: 590 self.LogMethodCall('Calling ', logname) 591 ret.append(ClockThisWithoutTheStars(logname, getattr(srv, eventid), args, keywords)) 592 elif shouldLogMethodCalls: 593 self.LogMethodCall('Skipping ', logname, ' (service not running)') 594 except Exception: 595 self.LogError('In %s.%s' % (getattr(srv, '__guid__', logname), eventid)) 596 log.LogException() 597 sys.exc_clear() 598 599 notifiedToRemove = [] 600 for weakObj in self.notifyObs[eventid]: 601 obj = weakObj() 602 if obj is None: 603 notifiedToRemove.append(weakObj) 604 else: 605 try: 606 if shouldLogMethodCalls: 607 logname = prefix + str(obj) 608 self.LogMethodCall('Calling ', logname) 609 getattr(obj, eventid)(*args, **keywords) 610 except Exception: 611 self.LogError('In %s.%s' % (getattr(weakObj, '__guid__', self.GetLogName(weakObj)), eventid)) 612 log.LogException() 613 sys.exc_clear() 614 615 for toRemove in notifiedToRemove: 616 if toRemove in self.notifyObs[eventid]: 617 self.notifyObs[eventid].remove(toRemove) 618 619 finally: 620 bt = 0 621 if old_block_trap: 622 bt = 1 623 stackless.getcurrent().block_trap = bt 624 return tuple(ret) 625 626 def ChainEvent(self, eventid, *args, **keywords): 627 return self.ChainEventWithoutTheStars(eventid, args, keywords) 628 629 def ChainEventWithoutTheStars(self, eventid, args, keywords = None): 630 if keywords is None: 631 keywords = {} 632 if not eventid.startswith('Process'): 633 self.LogError('ChainEvent called with event ', eventid, ". All events sent via ChainEvent should start with 'Process'") 634 self.LogError("Not only is the programmer responsible for this a 10z3r, but he wears his mother's underwear as well") 635 log.LogTraceback() 636 if stackless.getcurrent().block_trap or stackless.getcurrent().is_main: 637 raise RuntimeError("ChainEvent is blocking by design, but you're block trapped. You have'll have to find some alternative means to do Your Thing, dude.") 638 if eventid not in self.notify and eventid not in self.notifyObs: 639 self.LogInfo("Orphan'd event. ", eventid, "doesn't have any listeners") 640 shouldLogMethodCalls = self.ShouldLogMethodCalls() 641 if shouldLogMethodCalls: 642 self.LogMethodCall('ChainEvent(', eventid, ',*args=', args, ',**kw=', keywords, ')') 643 prefix = blue.pyos.taskletTimer.GetCurrent() + '::ChainEvent_' + eventid + '::' 644 ret = [] 645 for srv in self.notify[eventid]: 646 contextName = self.GetLogName(srv) 647 try: 648 logname = prefix + contextName 649 if ObjectShouldReceiveMessages(srv): 650 if shouldLogMethodCalls: 651 self.LogMethodCall('Calling ', logname) 652 retval = ClockThisWithoutTheStars(logname, getattr(srv, eventid), args, keywords) 653 ret.append(retval) 654 elif shouldLogMethodCalls: 655 self.LogMethodCall('Skipping ', logname, ' (service not running)') 656 except Exception: 657 self.LogError('In %s.%s' % (contextName, eventid)) 658 log.LogException() 659 sys.exc_clear() 660 661 notifiedToRemove = [] 662 for weakObj in self.notifyObs[eventid]: 663 obj = weakObj() 664 if obj is None: 665 notifiedToRemove.append(weakObj) 666 else: 667 contextName = self.GetLogName(weakObj) 668 try: 669 logname = prefix + contextName 670 if shouldLogMethodCalls: 671 self.LogMethodCall('Calling ', prefix + str(obj)) 672 ClockThisWithoutTheStars(logname, getattr(obj, eventid), args, keywords) 673 except Exception: 674 self.LogError('In %s.%s:' % (contextName, eventid)) 675 log.LogException() 676 sys.exc_clear() 677 678 for toRemove in notifiedToRemove: 679 if toRemove in self.notifyObs[eventid]: 680 self.notifyObs[eventid].remove(toRemove) 681 682 return tuple(ret) 683 684 def ScatterEvent(self, eventid, *args, **keywords): 685 return self.ScatterEventWithoutTheStars(eventid, args, keywords) 686 687 def ScatterEventWithoutTheStars(self, eventid, args, keywords = None): 688 keywords = dict(keywords) if keywords else {} 689 tracingEnabled = keywords.pop(bluepy.STACKLESS_TRACING_ENABLED_KEY, False) 690 if not eventid.startswith('On'): 691 self.LogError('ScatterEvent called with event ', eventid, ". All events sent via ScatterEvent should start with 'On'.") 692 self.LogError("Not only is the programmer responsible for this a 10z3r, but he wears his mother's underwear as well") 693 log.LogTraceback() 694 shouldLogMethodCalls = self.ShouldLogMethodCalls() 695 if shouldLogMethodCalls: 696 if IsFullLogging(): 697 self.LogMethodCall('ScatterEvent(', eventid, ',*args=', args, ',**kw=', keywords, ')') 698 else: 699 self.LogMethodCall('ScatterEvent(', eventid, ')') 700 prefix = blue.pyos.taskletTimer.GetCurrent() + '::ScatterEvent_' + eventid + '::' 701 for srv in self.notify[eventid]: 702 try: 703 logname = prefix + self.GetLogName(srv) 704 if ObjectShouldReceiveMessages(srv): 705 if shouldLogMethodCalls: 706 self.LogMethodCall('Calling ', logname) 707 srvGuid = getattr(srv, '__guid__', logname) 708 uthread.worker(prefix + srvGuid, self.MollycoddledUthread, srvGuid, eventid, getattr(srv, eventid), args, keywords, stackless_tracing_enabled=tracingEnabled) 709 elif shouldLogMethodCalls: 710 self.LogMethodCall('Skipping ', logname, ' (service not running)') 711 except Exception: 712 log.LogException() 713 714 observerKeywords = dict(keywords) 715 observerKeywords[bluepy.STACKLESS_TRACING_ENABLED_KEY] = tracingEnabled 716 notifiedToRemove = [] 717 for weakObj in self.notifyObs[eventid]: 718 try: 719 obj = weakObj() 720 func = getattr(obj, eventid, None) 721 if obj is None or func is None: 722 notifiedToRemove.append(weakObj) 723 else: 724 if shouldLogMethodCalls: 725 logname = prefix + str(obj) 726 self.LogMethodCall('Calling ', logname) 727 uthread.workerWithoutTheStars('', func, args, observerKeywords) 728 except Exception: 729 log.LogException() 730 731 for toRemove in notifiedToRemove: 732 if toRemove in self.notifyObs[eventid]: 733 self.notifyObs[eventid].remove(toRemove) 734 735 def NotifySessionChange(self, eventid, *args, **keywords): 736 return self.NotifySessionChangeWithoutTheStars(eventid, args, keywords) 737 738 def NotifySessionChangeWithoutTheStars(self, eventid, args, keywords = None): 739 if keywords is None: 740 keywords = {} 741 if eventid not in ('DoSessionChanging', 'ProcessSessionChange', 'OnSessionChanged'): 742 raise RuntimeError('NotifySessionChange called with eventid ', eventid, '. Must be one of DoSessionChanging, ProcessSessionChange or OnSessionChanged') 743 if eventid == 'ProcessSessionChange' and stackless.getcurrent().block_trap or stackless.getcurrent().is_main: 744 raise RuntimeError("ChainEvent (NotifySessionChange) is blocking by design, but you're block trapped.") 745 ret = [] 746 if 'machoNet' in self.services and self.services['machoNet'].IsClusterShuttingDown(): 747 self.LogNotice('Not notifying anyone about %s as the cluster is shutting down' % eventid) 748 return ret 749 if eventid == 'DoSessionChanging': 750 old_block_trap = stackless.getcurrent().block_trap 751 stackless.getcurrent().block_trap = 1 752 shouldLogMethodCalls = self.ShouldLogMethodCalls() 753 if shouldLogMethodCalls: 754 self.LogMethodCall('NotifySessionChange(', eventid, ',*args=', args, ',**kw=', keywords, ')') 755 prefix = blue.pyos.taskletTimer.GetCurrent() + '::NotifySessionChange_' + eventid + '::' 756 myNodeID = self.services['machoNet'].GetNodeID() 757 isSessionRelevant = boot.role != 'server' or session.contextOnly 758 try: 759 for srv in self.notify[eventid]: 760 try: 761 logname = prefix + self.GetLogName(srv) 762 if ObjectShouldReceiveMessages(srv): 763 if shouldLogMethodCalls: 764 self.LogMethodCall('Calling ' + logname) 765 if eventid == 'OnSessionChanged': 766 if not isSessionRelevant and srv.MachoResolve(session.GetActualSession()) == myNodeID: 767 isSessionRelevant = True 768 srvGuid = getattr(srv, '__guid__', logname) 769 uthread.worker(prefix + srvGuid, self.MollycoddledUthread, srvGuid, eventid, getattr(srv, eventid), args, keywords) 770 else: 771 retval = ClockThisWithoutTheStars(logname, getattr(srv, eventid), args, keywords) 772 ret.append(retval) 773 elif shouldLogMethodCalls: 774 self.LogMethodCall('Skipping ', logname, ' (service not running)') 775 except Exception: 776 log.LogException('In %s.%s' % (getattr(srv, '__guid__', self.GetLogName(srv)), eventid)) 777 sys.exc_clear() 778 779 notifiedToRemove = [] 780 for weakObj in self.notifyObs[eventid]: 781 obj = weakObj() 782 if obj is None or not getattr(obj, eventid, None): 783 notifiedToRemove.append(weakObj) 784 else: 785 try: 786 if shouldLogMethodCalls: 787 logname = prefix + str(obj) 788 self.LogMethodCall('Calling ', logname) 789 if eventid == 'OnSessionChanged': 790 uthread.workerWithoutTheStars('', getattr(obj, eventid), args, keywords) 791 else: 792 ClockThisWithoutTheStars(prefix + self.GetLogName(weakObj), getattr(obj, eventid), args, keywords) 793 except Exception: 794 self.LogError('In %s.%s' % (getattr(weakObj, '__guid__', self.GetLogName(srv)), eventid)) 795 log.LogException() 796 sys.exc_clear() 797 798 for toRemove in notifiedToRemove: 799 if toRemove in self.notifyObs[eventid]: 800 self.notifyObs[eventid].remove(toRemove) 801 802 if not session.contextOnly: 803 if not isSessionRelevant: 804 if session.irrelevanceTime is None: 805 session.irrelevanceTime = blue.os.GetWallclockTime() 806 log.LogInfo('CTXSESS: session ', session.sid, ' = ', session, ' is no longer relevant to any service') 807 elif session.irrelevanceTime is not None: 808 log.LogInfo('CTXSESS: session ', session.sid, ' became relevant again') 809 session.irrelevanceTime = None 810 finally: 811 if eventid == 'DoSessionChanging': 812 bt = 0 813 if old_block_trap: 814 bt = 1 815 stackless.getcurrent().block_trap = bt 816 817 return tuple(ret) 818 819 def MollycoddledUthread(self, guid, eventid, func, args, keywords): 820 try: 821 func(*args, **keywords) 822 except: 823 self.LogError('In %s.%s' % (guid, eventid)) 824 log.LogException() 825 sys.exc_clear() 826 827 def FavourMe(self, fn): 828 self.notify[fn.__name__].remove(fn.__self__) 829 self.notify[fn.__name__] = [fn.__self__] + self.notify[fn.__name__] 830 831 def UnfavourMe(self, fn): 832 if fn.__self__ in self.notify[fn.__name__]: 833 self.notify[fn.__name__].remove(fn.__self__) 834 self.notify[fn.__name__] = self.notify[fn.__name__] + [fn.__self__] 835 else: 836 self.LogWarn('Cannot unfavour ', fn.__self__, ' from ', fn.__name__, ", since it's not a notification listener") 837 838 def RegisterForNotifyEvent(self, ob, notify): 839 if ob not in self.notify[notify]: 840 self.LogInfo('Adding event', notify, 'for', ob) 841 self.notify[notify].append(ob) 842 843 def UnregisterForNotifyEvent(self, ob, notify): 844 try: 845 self.notify[notify].remove(ob) 846 except ValueError: 847 pass 848 except KeyError: 849 pass 850 851 def RegisterForNotifyEvent_WeakRef(self, ob, notify): 852 if weakref.ref(ob) not in self.notifyObs[notify]: 853 self.notifyObs[notify].append(weakref.ref(ob)) 854 self._RemoveDeadObjects(notify) 855 856 def UnregisterForNotifyEvent_WeakRef(self, ob, notify): 857 try: 858 self.notifyObs[notify] = filter(lambda x: x != weakref.ref(ob) and x() is not None, self.notifyObs[notify]) 859 except ValueError: 860 pass 861 except KeyError: 862 pass 863 864 def _RemoveDeadObjects(self, eventid): 865 try: 866 notifiedToRemove = [] 867 for weakObj in self.notifyObs[eventid]: 868 obj = weakObj() 869 if obj is None: 870 notifiedToRemove.append(weakObj) 871 872 for toRemove in notifiedToRemove: 873 if toRemove in self.notifyObs[eventid]: 874 self.notifyObs[eventid].remove(toRemove) 875 876 except Exception: 877 log.LogException('_RemoveDeadObjects: Failed to remove dead objects') 878 879 @telemetry.ZONE_METHOD 880 def RegisterNotify(self, ob): 881 if hasattr(ob, '__notifyevents__'): 882 if isinstance(ob, Service): 883 for notify in ob.__notifyevents__: 884 if ob not in self.notify[notify]: 885 self.notify[notify].append(ob) 886 887 else: 888 for notify in ob.__notifyevents__: 889 if weakref.ref(ob) not in self.notifyObs[notify]: 890 self.notifyObs[notify].append(weakref.ref(ob)) 891 892 else: 893 self.LogError('An object is calling registernotify without there being any notifyevents, the object is ', ob) 894 log.LogTraceback() 895 896 def UnregisterNotify(self, ob): 897 if hasattr(ob, '__notifyevents__'): 898 if isinstance(ob, Service): 899 for notify in ob.__notifyevents__: 900 if ob in self.notify[notify]: 901 self.notify[notify].remove(ob) 902 903 else: 904 for notify in ob.__notifyevents__: 905 self.notifyObs[notify] = filter(lambda x: x != weakref.ref(ob) and x() is not None, self.notifyObs[notify]) 906 907 else: 908 self.LogError('An object is calling unregisternotify without there being any notifyevents, the object is ', ob) 909 log.LogTraceback()