objectCaching.py (32170B)
1 #Embedded file name: C:\BuildAgent\work\20d73f1b24eed06e\eve\release\V24.01\carbon\common\script\net\objectCaching.py 2 import binascii 3 import cPickle 4 import numbers 5 import os 6 import random 7 import sys 8 import types 9 import zlib 10 import blue 11 import carbon.common.lib.const as const 12 import carbon.common.script.net.machobase as macho 13 import log 14 import telemetry 15 import uthread 16 from carbon.common.script.net.cachedObject import CachedObject as utilCachedObject 17 from carbon.common.script.net.machoNetAddress import MachoAddress 18 from carbon.common.script.sys import basesession, service 19 from carbon.common.script.sys.serviceConst import ROLE_ANY, ROLE_SERVICE 20 from carbon.common.script.sys.sessions import CachedMethodCalled 21 from carbon.common.script.util.format import FmtDateEng 22 from carbon.common.script.util.timerstuff import ClockThis 23 from stringutil import strx 24 cacheSessionVariables = None 25 26 class CoreObjectCachingSvc(service.Service): 27 __guid__ = 'svc.objectCaching' 28 __displayname__ = 'Object Caching Service' 29 __dependencies__ = ['machoNet'] 30 __exportedcalls__ = {'GetCachableObject': [ROLE_ANY], 31 'GetCachedObject': [ROLE_ANY], 32 'GetCachedObjectVersion': [ROLE_ANY], 33 'InvalidateCachedObjects': [ROLE_SERVICE], 34 'InvalidateCachedMethodCall': [ROLE_SERVICE], 35 'CacheObject': [ROLE_SERVICE], 36 'RefreshCache': [ROLE_SERVICE], 37 'GetCachedMethodCallVersion': [ROLE_SERVICE]} 38 __counters__ = {'cachedBytes': 'normal'} 39 __versionchecktimes__ = {'year': const.YEAR360, 40 '6 months': const.MONTH30 * 6, 41 '3 months': const.MONTH30 * 3, 42 'month': const.MONTH30, 43 'week': const.DAY * 7, 44 'day': const.DAY, 45 '12 hours': const.HOUR * 12, 46 '6 hours': const.HOUR * 6, 47 '3 hours': const.HOUR * 3, 48 '2 hours': const.HOUR * 2, 49 '1 hour': const.HOUR * 1, 50 '30 minutes': const.MIN * 30, 51 '15 minutes': const.MIN * 15, 52 '5 minutes': const.MIN * 5, 53 '1 minute': const.MIN, 54 '30 seconds': const.SEC * 30, 55 '15 seconds': const.SEC * 15, 56 '5 seconds': const.SEC * 5, 57 '1 second': const.SEC} 58 __cachedsessionvariables__ = ['languageID'] 59 60 def __init__(self): 61 if macho.mode == 'server': 62 self.__dependencies__.append('DB2') 63 service.Service.__init__(self) 64 self.runid = blue.os.GetWallclockTime() 65 self.lastChange = {} 66 self.cachedObjects = {} 67 self.cachedMethodCalls = {} 68 self.methodCallCachingDetails = {} 69 self.cachedObjectReturnStatistics = {} 70 self.lastFileWriteTime = {} 71 self.downloadedCachedObjects = {} 72 73 def GetHtmlStateDetails(self, k, v, detailed): 74 from carbon.common.script.util import htmlwriter 75 if k == 'cachedObjects': 76 if detailed: 77 hd = ['ObjectID', 78 'Time Stamp', 79 'Version', 80 'Shared', 81 'NodeID', 82 'Size'] 83 li = [] 84 for ik, iv in v.iteritems(): 85 l = '???' 86 if iv.pickle: 87 l = '%d' % len(iv.pickle) 88 if iv.compressed: 89 l += ' (compressed)' 90 li.append([iv.objectID, 91 FmtDateEng(iv.version[0]), 92 iv.version[1], 93 iv.shared, 94 iv.nodeID, 95 l]) 96 97 return ('Object Cache', htmlwriter.GetTable(hd, li, useFilter=True)) 98 else: 99 return ('Object Cache', '%d entries' % len(v)) 100 elif k == 'cachedObjectReturnStatistics': 101 if detailed: 102 hd = ['ObjectID', 'Count', 'Bytes'] 103 li = [] 104 for ik, iv in v.iteritems(): 105 if type(ik) == types.TupleType and len(ik) == 2 and ik[0] in sm.services: 106 ik = '%s::%s' % ik 107 li.append([htmlwriter.Swing(str(ik)), iv[0], iv[1]]) 108 109 li.sort() 110 return ('Object Cache Statistics', htmlwriter.GetTable(hd, li, useFilter=True)) 111 else: 112 totCalls = 0 113 totBytes = 0 114 for each in v.itervalues(): 115 totCalls += each[0] 116 totBytes += each[1] 117 118 return ('Object Cache Statistics', '%d distinct objects returned via %d calls, totalling %d bytes' % (len(v), totCalls, totBytes)) 119 120 def Run(self, memStream = None): 121 service.Service.Run(self, memStream) 122 self.methodCallCachingDetails = {} 123 self.cachedMethodCalls = {} 124 self.cachedObjects = {} 125 126 @telemetry.ZONE_METHOD 127 def LoadCache(self, ipaddress): 128 ipaddress = ipaddress.lower().split(':')[0] 129 self.cachePath = blue.paths.ResolvePathForWriting(u'cache:/MachoNet/%s/%d/' % (ipaddress, macho.version)) 130 paths = [blue.paths.ResolvePath(u'cache:/'), 131 blue.paths.ResolvePathForWriting(u'cache:/MachoNet/'), 132 blue.paths.ResolvePathForWriting(u'cache:/MachoNet/%s' % ipaddress), 133 self.cachePath, 134 self.cachePath + 'CachedMethodCalls', 135 self.cachePath + 'MethodCallCachingDetails', 136 self.cachePath + 'CachedObjects'] 137 self.methodCallCachingDetails = {} 138 self.cachedMethodCalls = {} 139 self.cachedObjects = {} 140 for path in paths: 141 try: 142 os.listdir(path) 143 except OSError: 144 os.mkdir(path) 145 sys.exc_clear() 146 147 def RefreshCache(self): 148 self.cachedMethodCalls = {} 149 150 def __CacheIsDirty(self, what, key): 151 if macho.mode == 'client': 152 ret = blue.os.GetWallclockTime() 153 shouldFlush = len(self.lastChange) == 0 154 self.lastChange[what, key] = ret 155 filename = self.cachePath + what + '/%s.cache' % self.KeyToFileName(key) 156 self.downloadedCachedObjects[key] = filename 157 if shouldFlush: 158 uthread.worker('objectCaching::FlushCache', self.__FlushCache, True) 159 return ret 160 161 def __FlushCache(self, yieldFirst): 162 if yieldFirst: 163 blue.pyos.synchro.Yield() 164 while self.lastChange: 165 flushWhat, lastChange = self.lastChange.items()[0] 166 del self.lastChange[flushWhat] 167 what, key = flushWhat 168 filename = self.cachePath + what + '/%s.cache' % self.KeyToFileName(key) 169 if os.access(filename, os.F_OK) and not os.access(filename, os.W_OK): 170 try: 171 os.chmod(filename, 666) 172 except Exception: 173 self.LogWarn('Failed to reset access priviledges on file from file system (', filename, ')') 174 sys.exc_clear() 175 176 try: 177 store = getattr(self, what) 178 if key not in store: 179 try: 180 os.unlink(filename) 181 self.LogInfo('Deleted ', key, ' from file system (', filename, ')') 182 except Exception as e: 183 self.LogWarn('Failed to delete file from file system (', filename, '):', e) 184 sys.exc_clear() 185 186 else: 187 ob = store[key] 188 if not isinstance(ob, uthread.Semaphore): 189 blue.AtomicFileWrite(filename, blue.marshal.Save((key, ob))) 190 self.LogInfo('Wrote ', key, ' to file (', filename, ')') 191 self.lastFileWriteTime[self.__KeyFromFileName(filename)] = os.path.getmtime(filename) 192 except Exception: 193 log.LogException('Error while writing cached data') 194 sys.exc_clear() 195 196 def KeyToFileName(self, key): 197 pickle = cPickle.dumps(key) 198 checksum = binascii.crc_hqx(pickle, 0) 199 return '%x' % checksum 200 201 def __KeyFromFileName(self, fileName): 202 fileName = fileName.replace('\\', '/') 203 i = fileName.rindex('/') 204 if i >= 0: 205 fileName = fileName[i:] 206 i = fileName.rindex('.') 207 if i >= 0: 208 fileName = fileName[0:i] 209 fileName = fileName.lower() 210 return fileName 211 212 def InvalidateCachedMethodCall(self, serviceOrObjectName, method, *args): 213 return self.InvalidateCachedMethodCalls([(serviceOrObjectName, method, args)]) 214 215 def InvalidateCachedMethodCalls(self, methodcalls): 216 invalidate = [] 217 for serviceOrObjectName, method, args in methodcalls: 218 si = [] 219 details = self.__GetMethodCallCachingDetails(None, serviceOrObjectName, method) 220 if details is not None: 221 if 'sessionInfo' in details: 222 si = [getattr(session, details['sessionInfo'], None)] 223 k = tuple([serviceOrObjectName, method] + si + list(args)) 224 invalidate.append(k) 225 226 self.InvalidateCachedObjects(invalidate) 227 228 def __ListPartialMatch(self, l, o, pm): 229 if len(pm) == len(l) and len(l) == len(o): 230 for i in range(len(l)): 231 if pm[i]: 232 if i < 2: 233 if type(o[i]) in (types.TupleType, types.ListType): 234 if l[i] not in o[i]: 235 return 0 236 if not l[i] == o[i]: 237 return 0 238 239 return 1 240 return 0 241 242 def InvalidateCachedObjects(self, objectIDs, forward = 1): 243 deleted = 0 244 for each in objectIDs: 245 if each in self.cachedMethodCalls: 246 del self.cachedMethodCalls[each] 247 self.__CacheIsDirty('cachedMethodCalls', each) 248 if type(each) in (types.TupleType, types.ListType): 249 objectID = each[0] 250 version = each[1] 251 else: 252 objectID = each 253 version = None 254 if objectID in self.cachedObjects and not isinstance(self.cachedObjects[objectID], uthread.Semaphore): 255 if version is None or self.__OlderVersion(self.cachedObjects[objectID].version, version): 256 del self.cachedObjects[objectID] 257 self.__CacheIsDirty('cachedObjects', objectID) 258 259 gpcs = sm.services['machoNet'] 260 gpcs.GetGPCS() 261 if macho.mode == 'client': 262 return 263 if macho.mode == 'proxy': 264 for eachNodeID in sm.services['machoNet'].GetConnectedSolNodes(): 265 gpcs.RemoteServiceNotify(self.session, MachoAddress(nodeID=eachNodeID, service='objectCaching'), 'objectCaching', 'InvalidateCachedObjects', objectIDs, 0) 266 267 elif forward and sm.services['machoNet'].GetConnectedProxyNodes(): 268 eachNodeID = random.choice(sm.services['machoNet'].GetConnectedProxyNodes()) 269 gpcs.RemoteServiceNotify(self.session, MachoAddress(nodeID=eachNodeID, service='objectCaching'), 'objectCaching', 'InvalidateCachedObjects', objectIDs, 0) 270 271 def GetCachedObject(self, objectID): 272 if objectID in self.cachedObjects and not isinstance(self.cachedObjects[objectID], uthread.Semaphore): 273 return self.cachedObjects[objectID].object 274 275 def GetCachedObjectVersion(self, objectID): 276 if objectID in self.cachedObjects and not isinstance(self.cachedObjects[objectID], uthread.Semaphore): 277 v = self.cachedObjects[objectID].version 278 if v[0] is None: 279 return v[1] 280 else: 281 return v 282 283 def GetCachableObject(self, shared, objectID, objectVersion, nodeID): 284 callTimer = basesession.CallTimer('objectCaching::GetCachableObject') 285 try: 286 if macho.mode == 'client': 287 gpcs = sm.services['machoNet'] 288 else: 289 gpcs = sm.services['machoNet'] 290 gpcs.GetGPCS() 291 if objectID in self.cachedObjects and isinstance(self.cachedObjects[objectID], utilCachedObject) and macho.mode == 'proxy': 292 if sm.services['machoNet'].GetTransportOfNode(self.cachedObjects[objectID].__nodeID__) is None: 293 del self.cachedObjects[objectID] 294 if objectID in self.cachedObjects and isinstance(self.cachedObjects[objectID], uthread.Semaphore) or objectID not in self.cachedObjects or not isinstance(self.cachedObjects[objectID], uthread.Semaphore) and self.__OlderVersion(self.cachedObjects[objectID].version, objectVersion): 295 if objectID not in self.cachedObjects or not isinstance(self.cachedObjects[objectID], uthread.Semaphore) and self.__OlderVersion(self.cachedObjects[objectID].version, objectVersion): 296 self.cachedObjects[objectID] = uthread.Semaphore(('objectCaching', objectID)) 297 while 1: 298 semaphore = self.cachedObjects[objectID] 299 semaphore.acquire() 300 try: 301 if not isinstance(self.cachedObjects[objectID], uthread.Semaphore): 302 self.__UpdateStatisticsGetCachableObject(objectID, objectVersion, self.cachedObjects[objectID]) 303 return self.cachedObjects[objectID] 304 if macho.mode == 'client': 305 if shared: 306 if not sm.services['machoNet'].myProxyNodeID: 307 proxyNodeID = nodeID 308 else: 309 proxyNodeID = sm.services['machoNet'].myProxyNodeID 310 remoteObject = gpcs.RemoteServiceCall(session, proxyNodeID, 'objectCaching', 'GetCachableObject', shared, objectID, objectVersion, nodeID) 311 else: 312 remoteObject = gpcs.RemoteServiceCall(session, nodeID, 'objectCaching', 'GetCachableObject', shared, objectID, objectVersion, nodeID) 313 self.cachedObjects[objectID] = remoteObject 314 self.__CacheIsDirty('cachedObjects', objectID) 315 elif macho.mode == 'proxy': 316 remoteObject = gpcs.RemoteServiceCall(self.session, nodeID, 'objectCaching', 'GetCachableObject', shared, objectID, objectVersion, nodeID) 317 if not remoteObject.compressed and len(remoteObject.pickle) > 200: 318 try: 319 t = ClockThis('objectCaching::compress', zlib.compress, remoteObject.pickle, 1) 320 except zlib.error as e: 321 raise RuntimeError('Compression Failure: ' + strx(e)) 322 323 if len(t) < len(remoteObject.pickle): 324 remoteObject.compressed = 1 325 remoteObject.pickle = t 326 if remoteObject.shared: 327 self.cachedObjects[objectID] = remoteObject 328 self.__CacheIsDirty('cachedObjects', objectID) 329 else: 330 del self.cachedObjects[objectID] 331 return remoteObject 332 elif macho.mode == 'server': 333 self.LogError("Some dude asked me for a cached object I just don't have, objectID=", objectID, ', objectVersion=', objectVersion, ', nodeID=', nodeID) 334 del self.cachedObjects[objectID] 335 raise RuntimeError("I don't have %s, which just don't make sense...." % repr(objectID)) 336 finally: 337 semaphore.release() 338 339 if not isinstance(self.cachedObjects[objectID], uthread.Semaphore): 340 self.__UpdateStatisticsGetCachableObject(objectID, objectVersion, self.cachedObjects[objectID]) 341 return self.cachedObjects[objectID] 342 343 else: 344 if macho.mode == 'server' and not self.cachedObjects[objectID].shared: 345 tmp = self.cachedObjects[objectID] 346 self.__UpdateStatisticsGetCachableObject(objectID, objectVersion, self.cachedObjects[objectID]) 347 del self.cachedObjects[objectID] 348 return tmp 349 self.__UpdateStatisticsGetCachableObject(objectID, objectVersion, self.cachedObjects[objectID]) 350 return self.cachedObjects[objectID] 351 finally: 352 callTimer.Done() 353 354 def CacheObject(self, object): 355 shared, objectID, objectVersion, nodeID, cachedObject, thePickle, compressed = object.MachoGetCachedObjectGuts() 356 if objectID not in self.cachedObjects or not isinstance(self.cachedObjects[objectID], uthread.Semaphore) and self.__NewerVersion(self.cachedObjects[objectID].version, objectVersion): 357 self.LogInfo('CacheObject: inserting or updating ', objectID, ' into cachedObjects') 358 if cachedObject is not None or thePickle is not None: 359 if objectID in self.cachedObjects and self.cachedObjects[objectID].pickle is not None: 360 self.cachedBytes.Dec(len(self.cachedObjects[objectID].pickle)) 361 self.cachedObjects[objectID] = CachedObject(shared, objectID, objectVersion, nodeID, cachedObject, thePickle, compressed) 362 if thePickle is not None: 363 self.cachedBytes.Add(len(thePickle)) 364 else: 365 log.LogTraceback('Unexpected CacheObject request') 366 self.__CacheIsDirty('cachedObjects', objectID) 367 368 def __NewerVersion(self, oldVersion, newVersion): 369 if oldVersion[1] != newVersion[1]: 370 return newVersion[0] > oldVersion[0] 371 return 0 372 373 def __OlderVersion(self, oldVersion, newVersion): 374 if oldVersion[1] != newVersion[1]: 375 return oldVersion[0] < newVersion[0] 376 return 0 377 378 def __GetMethodCallCachingDetails(self, serviceOrObject, serviceOrObjectName, method): 379 if serviceOrObject is not None: 380 exportedCalls = getattr(serviceOrObject, '__exportedcalls__', None) 381 if exportedCalls is not None: 382 methodInfo = exportedCalls.get(method, None) 383 if methodInfo is not None: 384 if type(methodInfo) == types.DictType and 'caching' in methodInfo: 385 return methodInfo['caching'] 386 return 387 else: 388 return self.methodCallCachingDetails.get((serviceOrObjectName, method), None) 389 390 def GetCachedMethodCallVersion(self, serviceOrObject, serviceOrObjectName, method, args): 391 details = self.__GetMethodCallCachingDetails(serviceOrObject, serviceOrObjectName, method) 392 if details is not None: 393 si = [] 394 if details.get('sessionInfo', None): 395 si = [getattr(session, details['sessionInfo'], None)] 396 k = tuple([serviceOrObjectName, method] + si + list(args)) 397 if k in self.cachedMethodCalls: 398 if isinstance(self.cachedMethodCalls[k]['rret'].result, utilCachedObject) and macho.mode == 'proxy': 399 if sm.services['machoNet'].GetTransportOfNode(self.cachedMethodCalls[k]['rret'].result.__nodeID__) is None: 400 del self.cachedMethodCalls[k] 401 self.LogInfo('ObjectCaching ', serviceOrObjectName, '::', method, '(', args, ') says cachable, but on dead server so go for it') 402 return 0 403 return self.cachedMethodCalls[k]['version'][1] 404 return 0 405 406 def __UpdateStatisticsGetCachableObject(self, objectID, objectVersion, object): 407 k = objectID 408 if len(objectID) == 3 and objectID[0] == 'Method Call': 409 if objectID[1] != macho.mode: 410 k = '::'.join(objectID[2][:2]) 411 else: 412 k = objectID[1:] 413 k = self.__SanitizeCachingKey(k) 414 if k not in self.cachedObjectReturnStatistics: 415 self.cachedObjectReturnStatistics[k] = [0, 0] 416 self.cachedObjectReturnStatistics[k][0] += 1 417 if macho.mode != 'client': 418 self.cachedObjectReturnStatistics[k][1] += object.GetSize() 419 420 def __SanitizeCachingKey(self, k): 421 if type(k) in (types.ListType, types.TupleType): 422 ret = [] 423 for each in k: 424 tmp = self.__SanitizeCachingKey(each) 425 if tmp: 426 ret.append(tmp) 427 428 if ret and ret[0] == macho.mode: 429 ret = ret[1:] 430 if len(ret) == 1: 431 return ret[0] 432 return tuple(ret) 433 elif isinstance(k, numbers.Real): 434 return 0 435 else: 436 return k 437 438 def ReturnCachedResults(self, cachable, versionCheck, throwOK, cachedResultRecord, cacheKey, cacheDetails): 439 CachedMethodCalled(cacheKey, cacheDetails) 440 return (cachable, 441 versionCheck, 442 throwOK, 443 cachedResultRecord, 444 cacheKey, 445 cacheDetails) 446 447 def PerformCachedMethodCall(self, serviceOrObject, serviceOrObjectName, method, args, remoteVersion = None): 448 details = self.__GetMethodCallCachingDetails(serviceOrObject, serviceOrObjectName, method) 449 if details is not None: 450 si = [] 451 if details.get('sessionInfo', None): 452 si = [getattr(session, details['sessionInfo'], None)] 453 k = tuple([serviceOrObjectName, method] + si + list(args)) 454 if k in self.cachedMethodCalls: 455 if isinstance(self.cachedMethodCalls[k]['rret'].result, utilCachedObject) and macho.mode == 'proxy': 456 if sm.services['machoNet'].GetTransportOfNode(self.cachedMethodCalls[k]['rret'].result.__nodeID__) is None: 457 del self.cachedMethodCalls[k] 458 self.LogInfo('ObjectCaching ', serviceOrObjectName, '::', method, '(', args, ') says cachable, but on dead server so go for it') 459 return self.ReturnCachedResults(1, 0, 0, None, k, details) 460 versionCheck = self.__ShouldVersionCheck(details, self.cachedMethodCalls[k]) 461 if remoteVersion is None or remoteVersion == 1 or self.__OlderVersion(remoteVersion, self.cachedMethodCalls[k]['version']): 462 throwOK = 0 463 else: 464 throwOK = 1 465 self.cachedMethodCalls[k]['used'] = blue.os.GetWallclockTime() 466 if versionCheck: 467 self.LogInfo('ObjectCaching ', serviceOrObjectName, '::', method, '(', args, ') returning a cached result that requires version checking.') 468 else: 469 self.LogInfo('ObjectCaching ', serviceOrObjectName, '::', method, '(', args, ') returning a cached result') 470 return self.ReturnCachedResults(1, versionCheck, throwOK, self.cachedMethodCalls[k], k, details) 471 elif remoteVersion is not None and remoteVersion != 1 and self.__GetVersionCheckType(details) == 'never': 472 self.LogInfo('ObjectCaching ', serviceOrObjectName, '::', method, '(', args, ') the caller seems to know the result, and we can live with that') 473 return self.ReturnCachedResult(1, 0, 1, None, k, details) 474 else: 475 if remoteVersion is not None and remoteVersion != 1: 476 self.LogInfo('ObjectCaching ', serviceOrObjectName, '::', method, '(', args, ') is cachable, but not cached. remoteVersion=', remoteVersion, ', details=', details, ', vc type=', self.__GetVersionCheckType(details)) 477 else: 478 self.LogInfo('ObjectCaching ', serviceOrObjectName, '::', method, '(', args, ') is cachable, but not cached.') 479 return self.ReturnCachedResults(1, 0, 0, None, k, details) 480 else: 481 return self.ReturnCachedResults(0, 0, 0, None, None, None) 482 483 if cacheSessionVariables is not None: 484 __cachedsessionvariables__ = tuple(list(cacheSessionVariables) + __cachedsessionvariables__) 485 else: 486 __cachedsessionvariables__ = tuple(__cachedsessionvariables__) 487 488 def CacheMethodCall(self, serviceOrObjectName, method, args, cmcr): 489 k = (serviceOrObjectName, method) 490 if k not in self.methodCallCachingDetails: 491 self.methodCallCachingDetails[k] = cmcr.details 492 self.__CacheIsDirty('methodCallCachingDetails', k) 493 details = cmcr.details 494 else: 495 details = self.methodCallCachingDetails[serviceOrObjectName, method] 496 CachedMethodCalled(k, cmcr.details) 497 if self.__GetVersionCheckType(details) is not None: 498 si = [] 499 if details.get('sessionInfo', None): 500 if macho.mode != 'client' and details['sessionInfo'] not in self.__cachedsessionvariables__: 501 self.LogInfo('ObjectCaching cannot cache by sessionInfo ', details['sessionInfo'], ' on ', macho.mode) 502 return 503 gsi = getattr(session, details['sessionInfo'], None) 504 si = [gsi] 505 ret, version = cmcr.GetResult(), cmcr.GetVersion() 506 k = tuple([serviceOrObjectName, method] + si + list(args)) 507 used = self.__CacheIsDirty('cachedMethodCalls', k) 508 self.cachedMethodCalls[k] = {'lret': ret, 509 'rret': cmcr, 510 'version': version, 511 'used': used, 512 'runid': self.runid} 513 self.LogInfo('ObjectCaching ', serviceOrObjectName, '::', method, '(', args, ') cached the method call for future reference.') 514 515 def __GetVersionCheckType(self, details): 516 vc = details.get('versionCheck', 'run') 517 if type(vc) == types.TupleType: 518 if macho.mode == 'server': 519 vc = vc[2] 520 elif macho.mode == 'proxy': 521 vc = vc[1] 522 else: 523 vc = vc[0] 524 return vc 525 526 def IsCachedOnProxy(self, details): 527 if details.get('sessionInfo', None) and details.get('sessionInfo', None) not in self.__cachedsessionvariables__: 528 return False 529 vc = details.get('versionCheck', 'run') 530 if type(vc) == types.TupleType: 531 return vc[1] is not None 532 return vc is not None 533 534 def UpdateVersionCheckPeriod(self, cacheKey): 535 try: 536 cacheEntry = self.cachedMethodCalls.get(cacheKey, None) 537 if cacheEntry is None: 538 self.LogWarn('ObjectCaching unable to update validity period for cache key', cacheKey, 'key not valid') 539 else: 540 self.LogInfo('ObjectCaching resetting validity period for cache key', cacheKey) 541 cacheEntry['version'][0] = blue.os.GetWallclockTime() 542 except Exception as e: 543 log.LogException('ObjectCaching unknown exception while resetting validity period') 544 545 def __ShouldVersionCheck(self, details, info): 546 vc = self.__GetVersionCheckType(details) 547 now = blue.os.GetWallclockTime() 548 if type(vc) in types.StringTypes: 549 if vc == 'never': 550 return 0 551 if vc == 'always': 552 return 1 553 if vc in ('run', 'utcmidnight', 'utcmidnight_or_3hours'): 554 if info.get('runid', 0) == self.runid: 555 if vc != 'run': 556 midnight = (info.get('runid', 0) / const.DAY + 1) * const.DAY 557 if vc == 'utcmidnight': 558 maxAge = midnight - now 559 else: 560 maxAge = min(midnight - now, 3 * const.HOUR) 561 else: 562 return 0 563 else: 564 info['runid'] = self.runid 565 return 1 566 else: 567 maxAge = self.__versionchecktimes__[vc] 568 else: 569 maxAge = int(vc) 570 age = now - info['version'][0] 571 if maxAge <= age: 572 return 1 573 else: 574 return 0 575 576 577 class CachedObject(): 578 __passbyvalue__ = 1 579 580 def __init__(self, s, oid, v, n, o, p, c): 581 self.version, self.object, self.nodeID, self.shared, self.pickle, self.compressed, self.objectID = (v, 582 o, 583 n, 584 s, 585 p, 586 c, 587 oid) 588 589 def __getstate__(self): 590 if self.pickle is not None: 591 o = None 592 else: 593 o = self.object 594 return (self.version, 595 o, 596 self.nodeID, 597 self.shared, 598 self.pickle, 599 self.compressed, 600 self.objectID) 601 602 def __str__(self): 603 try: 604 return 'objectCaching.CachedObject(objectID=%s, version=%s, nodeID=%s, shared=%s, compressed=%s)' % (self.objectID, 605 self.version, 606 self.nodeID, 607 self.shared, 608 self.compressed) 609 except: 610 sys.exc_clear() 611 return 'objectCaching.CachedObject(???)' 612 613 def __setstate__(self, state): 614 self.object = None 615 self.version, self.object, self.nodeID, self.shared, self.pickle, self.compressed, self.objectID = state 616 617 def GetObject(self): 618 if self.object is None: 619 if self.pickle is None: 620 raise RuntimeError('Getting cached object contents, but both the object and the pickle are none') 621 try: 622 if self.compressed: 623 try: 624 dasPickle = zlib.decompress(self.pickle) 625 except zlib.error as e: 626 raise RuntimeError('Decompression Failure: ' + strx(e)) 627 628 self.object = blue.marshal.Load(dasPickle) 629 else: 630 self.object = blue.marshal.Load(self.pickle) 631 except: 632 sm.GetService('objectCaching').LogError('Failed to acquire object from objectCaching.CachedObject, self=', self) 633 raise 634 635 return self.object 636 637 def CompressedPart(self): 638 if not self.compressed or self.pickle is None: 639 return 0 640 return len(self.pickle) 641 642 def GetSize(self): 643 if self.pickle is None: 644 self.pickle = blue.marshal.Save(self.object) 645 return len(self.pickle) 646 647 648 class CachedMethodCallResult(): 649 __passbyvalue__ = 1 650 651 def __init__(self, key, details, result): 652 self.details = details 653 if isinstance(result, utilCachedObject): 654 self.result = result 655 self.version = None 656 elif sm.services['objectCaching'].IsCachedOnProxy(details) and macho.mode != 'client': 657 self.result = utilCachedObject(1, ('Method Call', macho.mode, key), result) 658 sm.services['objectCaching'].CacheObject(self.result) 659 self.version = None 660 else: 661 self.result = blue.marshal.Save(result) 662 self.version = [blue.os.GetWallclockTime(), zlib.adler32(self.result)] 663 664 def __getstate__(self): 665 return (self.details, self.result, self.version) 666 667 def __setstate__(self, state): 668 self.details, self.result, self.version = state 669 670 def GetVersion(self): 671 if isinstance(self.result, utilCachedObject): 672 return list(self.result.__objectVersion__) 673 return self.version 674 675 def IsSameVersion(self, machoVersion): 676 if machoVersion != 1: 677 return self.GetVersion()[1] == machoVersion[1] 678 return 0 679 680 def GetResult(self): 681 if isinstance(self.result, utilCachedObject): 682 return self.result.GetCachedObject() 683 else: 684 return blue.marshal.Load(self.result) 685 686 def __str__(self): 687 try: 688 if isinstance(self.result, utilCachedObject): 689 return 'objectCaching.CachedMethodCallResult(version=%s, result=%s)' % (self.version, self.result) 690 return 'objectCaching.CachedMethodCallResult(version=%s)' % (self.version,) 691 except: 692 sys.exc_clear() 693 return 'objectCaching.CachedMethodCallResult(???)' 694 695 696 class CacheOK(Exception): 697 __passbyvalue__ = 1 698 699 def __init__(self, value = 'CacheOK', *args): 700 Exception.__init__(self, value, *args)