machoNetTransport.py (39391B)
1 #Embedded file name: C:\BuildAgent\work\20d73f1b24eed06e\eve\release\V24.01\carbon\common\script\net\machoNetTransport.py 2 from __future__ import absolute_import 3 import copy 4 import sys 5 import weakref 6 import zlib 7 import blue 8 import datetime 9 import uthread 10 import stackless 11 import carbon.common.script.sys.basesession as base 12 import log 13 import carbon.common.script.util.logUtil as logUtil 14 import carbon.common.script.net.machobase as macho 15 from carbon.common.lib import const 16 from carbon.common.script.net import packetthreshold 17 from carbon.common.script.net.GPSExceptions import GPSTransportClosed 18 from carbon.common.script.net.machoNetAddress import MachoAddress 19 from carbon.common.script.net.machoNetExceptions import SessionUnavailable, UnMachoDestination 20 from carbon.common.script.net.machoNetPacket import SessionChangeNotification, SessionInitialStateNotification, TransportClosed, MachoPacket 21 import cluster 22 import bluepy 23 from eveexceptions import ServiceNotFound 24 from eveprefs import prefs 25 import monolithdisconnectlogger 26 from stringutil import strx 27 import iocp 28 from carbon.common.script.sys.service import ROLE_SERVICE, ROLE_REMOTESERVICE 29 import localstorage 30 from journey.tracker import get_journey_id 31 from machonet_tracing import propagate_trace_context 32 import monolithmetrics 33 if '/disablePacketCompression' in blue.pyos.GetArg(): 34 log.general.Log('Packet Compression: Disabled', log.LGINFO) 35 MACHONET_COMPRESSION_DISABLED = True 36 elif iocp.UsingCompression(): 37 log.general.Log('Packet Compression: IOCP', log.LGINFO) 38 if iocp.UsingIOCP(): 39 import carbonio 40 carbonio.setCompressionThreshold(prefs.GetValue('machoNet.minimumBytesToCompress', 200)) 41 carbonio.setCompressionMinRatio(prefs.GetValue('packetCompressionMinimumRatio', 75)) 42 carbonio.setCompressionLevel(prefs.GetValue('packetCompressionLevel', 6)) 43 MACHONET_COMPRESSION_DISABLED = True 44 else: 45 log.general.Log('Could not turn on IOCP packet compression as IOCP is not enabled! Reverting to MachoNet compression...', log.LGERR) 46 MACHONET_COMPRESSION_DISABLED = False 47 else: 48 log.general.Log('Packet Compression: MachoNet', log.LGINFO) 49 MACHONET_COMPRESSION_DISABLED = False 50 from . import metrics 51 52 class MachoTransport(logUtil.LogMixin): 53 __guid__ = 'macho.MachoTransport' 54 __sessioninitorchangenotification__ = {cluster.MACHONETMSG_TYPE_SESSIONINITIALSTATENOTIFICATION, cluster.MACHONETMSG_TYPE_SESSIONCHANGENOTIFICATION} 55 56 def __init__(self, transportID, transport, transportName, machoNet): 57 self.machoNet = weakref.proxy(machoNet) 58 logUtil.LogMixin.__init__(self, '%s transport' % self.machoNet.__guid__) 59 self.nodeID = None 60 self.transportID = transportID 61 self.transport = transport 62 self.transportName = transportName 63 if self.transportName == 'tcp:packet:client': 64 self.userID = None 65 self.clientID = 0 66 self.sessionID = None 67 self.dependants = {} 68 self.sessions = {} 69 self.contextSessions = {} 70 self.queuedClientCalls = {} 71 self.calls = {} 72 self.currentReaders = 0 73 if self.transportName == 'tcp:packet:client': 74 self.desiredReaders = 2 75 else: 76 self.desiredReaders = 20 77 self.lastPing = None 78 self.pinging = False 79 self.estimatedRTT = 100 * const.MSEC 80 self.timeDiff = 0 81 self.compressionThreshold = prefs.GetValue('machoNet.minimumBytesToCompress', 200) 82 self.compressionPercentageThreshold = prefs.GetValue('machoNet.maxPercentagePreCompressedToCompress', 75) 83 self.largePacketLogSpamThreshold = prefs.GetValue('machoNet.largePacketLogSpamThreshold', None) 84 self.dropPacketThreshold = prefs.GetValue('machoNet.dropPacketThreshold', 5000000) 85 self.packetThreshold = packetthreshold.PacketThreshold(self.transportName, self.dropPacketThreshold, macho.mode) 86 self.message_type_metric_tags = {} 87 self.fetch_metric_tags() 88 89 def __str__(self): 90 return repr(self) 91 92 def __repr__(self): 93 try: 94 return "MachoTransport(nodeID=%s,transportID=%s,transport='%s',transportName='%s',clientID=%s" % (self.nodeID, 95 self.transportID, 96 str(self.transport), 97 self.transportName, 98 self.clientID) 99 except Exception: 100 sys.exc_clear() 101 return 'MachoTransport containing crappy data' 102 103 def Finalize(self, nodeID): 104 self.nodeID = nodeID 105 self.fetch_metric_tags() 106 107 def LogInfo(self, *args): 108 if self.machoNet.isLogInfo: 109 logUtil.LogMixin.LogInfo(self, *args) 110 111 def GetMachoAddressOfOtherSide(self): 112 if macho.mode == 'client': 113 return MachoAddress(nodeID=self.machoNet.myProxyNodeID) 114 elif self.clientID: 115 return MachoAddress(clientID=self.clientID) 116 else: 117 return MachoAddress(nodeID=self.nodeID) 118 119 def IsClosed(self): 120 if self.transport is not None: 121 return self.transport.IsClosed() 122 return True 123 124 def HasLegacyClient(self): 125 return self.transport is not None 126 127 def Close(self, reason, reasonCode = None, reasonArgs = {}, exception = None, noSend = False): 128 if self.transportID in self.machoNet.transportsByID: 129 blue.net.PurgeTransport(self.transportID) 130 oldNodeIDByServiceMask = {} 131 if self.transportName == 'ip:packet:server' and 'ip:packet:server' in self.machoNet.namedtransports: 132 self.machoNet.ClearAddressCache() 133 self.machoNet.ResetAutoResolveCache() 134 del self.machoNet.namedtransports['ip:packet:server'] 135 if self.nodeID is not None and (self.nodeID in self.machoNet.transportIDbyProxyNodeID and self.machoNet.transportIDbyProxyNodeID[self.nodeID] == self.transportID or self.nodeID in self.machoNet.transportIDbySolNodeID and self.machoNet.transportIDbySolNodeID[self.nodeID] == self.transportID): 136 if not self.machoNet.IsClusterShuttingDown(): 137 self.machoNet.LogError('Removing transport from transportIDbyProxyNodeID or transportIDbySolNodeID', self.nodeID, self, reason) 138 log.LogTraceback() 139 try: 140 if self.nodeID in self.machoNet.transportIDbyProxyNodeID: 141 del self.machoNet.transportIDbyProxyNodeID[self.nodeID] 142 blue.net.DelProxyNode(self.nodeID) 143 except Exception: 144 sys.exc_clear() 145 self.LogInfo('Exception during MachoTransport::Close ignored.') 146 147 try: 148 if self.nodeID in self.machoNet.transportIDbySolNodeID: 149 del self.machoNet.transportIDbySolNodeID[self.nodeID] 150 except Exception: 151 sys.exc_clear() 152 self.LogInfo('Exception during MachoTransport::Close ignored.') 153 154 oldNodeIDByServiceMask = copy.deepcopy(self.machoNet.GetFullServiceMaskAddressCache()) 155 self.machoNet.RemoveAllForNode(self.nodeID) 156 if macho.mode == 'proxy' and not self.machoNet.IsClusterShuttingDown(): 157 self.machoNet.ServerBroadcast('OnNodeDeath', self.nodeID, 1, "The node's transport was detected disconnected by the proxy. reason =%s" % strx(reason)) 158 deletions = {clientID for clientID, transportID in self.machoNet.transportIDbyClientID.iteritems() if transportID == self.transportID} 159 for each in deletions: 160 try: 161 del self.machoNet.transportIDbyClientID[each] 162 except Exception: 163 sys.exc_clear() 164 self.LogInfo('Exception during MachoTransport::Close ignored.') 165 166 try: 167 self.machoNet.remove_session_transport_lookup(self.sessionID) 168 except Exception: 169 sys.exc_clear() 170 self.LogInfo('Exception during MachoTransport::Close ignored.') 171 172 try: 173 del self.machoNet.transportsByID[self.transportID] 174 except Exception: 175 sys.exc_clear() 176 self.LogInfo('Exception during MachoTransport::Close ignored.') 177 178 if self.transport is not None: 179 try: 180 self.transport.Close(reason, reasonCode, reasonArgs, exception, noSend) 181 except Exception: 182 sys.exc_clear() 183 self.LogInfo('Exception during MachoTransport::Close ignored, but unable to successfully transport.Close().') 184 185 while self.calls: 186 try: 187 k, v = self.calls.items()[0] 188 try: 189 del self.calls[k] 190 finally: 191 v.send_exception(GPSTransportClosed, reason, reasonCode, reasonArgs) 192 193 except Exception: 194 log.LogException('Exception during transport closed broadcast. (Semi)-silently ignoring this.') 195 sys.exc_clear() 196 197 if not hasattr(self, 'done_broadcasting_close'): 198 self.done_broadcasting_close = 1 199 if len(self.dependants) and self.transportName == 'tcp:packet:client': 200 msg = TransportClosed(clientID=self.clientID, isRemote=0) 201 msg2 = TransportClosed(clientID=self.clientID, isRemote=1) 202 203 def CloseTransport(caller, tid, msg): 204 if tid in caller.transportsByID: 205 try: 206 caller.transportsByID[tid].Write(msg) 207 except Exception: 208 log.LogTraceback() 209 caller.transportsByID[tid].Close('Write failed big time in CloseTransport') 210 sys.exc_clear() 211 except: 212 log.LogTraceback() 213 caller.transportsByID[tid].Close('Write failed big time in CloseTransport, non-standard error') 214 raise 215 216 for tid in self.dependants.iterkeys(): 217 uthread.worker('machoNet::CloseTransport', CloseTransport, self.machoNet, tid, msg) 218 msg = msg2 219 220 if self.transportName == 'tcp:packet:client': 221 ipaddress = self.transport.address.split(':')[0] 222 self.machoNet.LogOffSession(self.sessionID, ipaddress=ipaddress) 223 monolithdisconnectlogger.log_disconnect(ipaddress) 224 if self.transportName == 'tcp:packet:machoNet': 225 disco = set() 226 for tid in self.machoNet.transportIDbyClientID.itervalues(): 227 if self.transportID in self.machoNet.transportsByID[tid].dependants: 228 disco.add(tid) 229 230 for tid in self.dependants: 231 if self.transportID in self.machoNet.transportsByID[tid].dependants: 232 disco.add(tid) 233 234 if disco: 235 mappingsForDeadNode = [] 236 if prefs.GetValue('DontDisconnectOnClusterSingletonDeath', 0): 237 for serviceID, mapping in oldNodeIDByServiceMask.iteritems(): 238 if self.nodeID in mapping: 239 mappingsForDeadNode.append(serviceID) 240 241 if mappingsForDeadNode == [cluster.SERVICE_CLUSTERSINGLETON]: 242 self.machoNet.LogWarn('NOT Disconnecting %s dependant users even though node %s has gone offline. The node was just servicing clustersingleton services.' % (len(disco), self.nodeID)) 243 else: 244 self.machoNet.LogWarn('Disconnecting %s dependant users because node %s has gone offline. The node had the following serviceIDs: %s' % (len(disco), self.nodeID, ','.join([ str(m) for m in mappingsForDeadNode ]))) 245 for tid in disco: 246 self.machoNet.transportsByID[tid].Close('A server node you were using has gone offline.', 'NODEDEATH') 247 248 self.dependants = {} 249 if self.transportName == 'tcp:packet:machoNet': 250 reason = 'Session terminated due to local transport closure' 251 if macho.mode == 'client' and reasonCode: 252 reason = reasonCode 253 while len(self.sessions): 254 for nodeID in self.sessions.keys(): 255 self.SessionClosed(nodeID, reason) 256 257 while len(self.contextSessions): 258 for nodeID in self.contextSessions.keys(): 259 self.SessionClosed(nodeID, reason) 260 261 @bluepy.TimedFunction('machoNet::Transport::SessionClosed') 262 def SessionClosed(self, clientID, reason, isRemote = 0): 263 try: 264 del self.machoNet.transportIDbyClientID[clientID] 265 except KeyError: 266 pass 267 268 blue.net.PurgeClient(clientID) 269 sess = None 270 if clientID in self.sessions: 271 sess = self.sessions[clientID] 272 del self.sessions[clientID] 273 if not sess and clientID in self.contextSessions: 274 sess = self.contextSessions[clientID] 275 del self.contextSessions[clientID] 276 if sess: 277 self.queuedClientCalls[clientID] = uthread.Channel() 278 with MachoCallOrNotification(self, sess, None): 279 mask = sess.Masquerade() 280 try: 281 if self.transportName == 'ip:packet:server': 282 sm.ScatterEvent('OnDisconnect', getattr(self, 'disconnectsilently', 0), reason) 283 if macho.mode != 'client': 284 sess.LogSessionHistory(reason) 285 base.CloseSession(sess, isRemote) 286 else: 287 sess.ClearAttributes(dontSendMessage=True) 288 finally: 289 mask.UnMask() 290 queueChannel = self.queuedClientCalls[clientID] 291 del self.queuedClientCalls[clientID] 292 queueChannel.preference = 1 293 while queueChannel.balance < 0: 294 queueChannel.send(None) 295 296 def _JitSessionToOtherSide(self, clientID): 297 clientTransportID = self.machoNet.transportIDbyClientID[clientID] 298 clientTransport = self.machoNet.transportsByID[clientTransportID] 299 remoteSessionVersion = clientTransport.dependants.get(self.transportID, 0) 300 sess = clientTransport.sessions[clientID] 301 if remoteSessionVersion != sess.version: 302 sessData = {} 303 self.machoNet.LogInfo('JITing session ', sess.sid, ' (clientID ', clientID, '), localVer = ', sess.version, ' remoteVer = ', remoteSessionVersion) 304 if remoteSessionVersion == 0: 305 for v in sess.GetDistributedProps(0): 306 sessData[v] = getattr(sess, v) 307 308 sessionprops = SessionInitialStateNotification(source=MachoAddress(clientID=clientID), destination=MachoAddress(nodeID=self.nodeID), sid=sess.sid, sessionType=sess.sessionType, initialstate=sessData) 309 else: 310 for v in sess.GetDistributedProps(1): 311 sessData[v] = (None, getattr(sess, v)) 312 313 sessionprops = SessionChangeNotification(source=MachoAddress(clientID=clientID), destination=MachoAddress(nodeID=self.nodeID), sid=sess.sid, change=(1, sessData)) 314 clientTransport.dependants[self.transportID] = sess.version 315 self.Write(sessionprops, jitSession=0) 316 317 @bluepy.TimedFunction('machoNet::MachoTransport::Write') 318 def Write(self, message, jitSession = 1): 319 propagate_trace_context(message) 320 if macho.mode == 'client': 321 message.journeyID = get_journey_id() 322 else: 323 ls = localstorage.GetLocalStorage() 324 message.journeyID = ls.get('journeyID', None) 325 if macho.mode == 'proxy' and jitSession: 326 if message.source.addressType == const.ADDRESS_TYPE_CLIENT: 327 self._JitSessionToOtherSide(message.source.clientID) 328 elif base.IsInClientContext() and session and hasattr(session, 'clientID'): 329 clientID = session.clientID 330 self._JitSessionToOtherSide(clientID) 331 message.contextKey = clientID 332 if hasattr(self, 'userID'): 333 message.userID = self.userID 334 if message.source.addressType == const.ADDRESS_TYPE_ANY and not message.command % 2: 335 message.source.nodeID = self.machoNet.nodeID 336 message.source.addressType = const.ADDRESS_TYPE_NODE 337 message.Changed() 338 elif message.source.addressType == const.ADDRESS_TYPE_NODE and message.source.nodeID is None: 339 message.source.nodeID = self.machoNet.nodeID 340 message.Changed() 341 thePickle = message.GetPickle() 342 self.LogInfo('Write: ', message) 343 if self.transportName != 'tcp:packet:machoNet' and message.compressedPart * 100 / len(thePickle) < self.compressionPercentageThreshold and len(thePickle) - message.compressedPart > self.compressionThreshold and not MACHONET_COMPRESSION_DISABLED: 344 before = len(thePickle) 345 try: 346 with bluepy.Timer('machoNet::MachoTransport::Write::Compress'): 347 compressed = zlib.compress(thePickle, 1) 348 except zlib.error as e: 349 raise RuntimeError('Compression Failure: ' + strx(e)) 350 351 after = len(compressed) 352 if after > before: 353 self.LogInfo('Compress would have exploded data from ', before, ' to ', after, ' bytes. Sending uncompressed.') 354 elif (before - after) * 100 / before <= 5: 355 self.LogInfo("Compress didn't help one bit. Would have compressed data from ", before, ' to ', after, " bytes, which is insignificant. Sending uncompressed, rather than wasting the recipient's CPU power for nothing.") 356 else: 357 thePickle = compressed 358 self.machoNet.compressedBytes.Add(before - after) 359 if self.transportName == 'tcp:packet:client' and macho.mode == 'proxy': 360 for objectID, refID in message.oob.get('OID+', {}).iteritems(): 361 s = self.sessions.get(self.clientID, None) 362 if s is not None: 363 s.RegisterMachoObject(objectID, None, refID) 364 365 if self.largePacketLogSpamThreshold != None and len(thePickle) > self.largePacketLogSpamThreshold: 366 log.LogTraceback(extraText='Packet larger than the %d byte largePacketLogSpamTreshhold being written out to wire (%d > %d)' % (self.largePacketLogSpamThreshold, len(thePickle), self.largePacketLogSpamThreshold)) 367 if self.packetThreshold.ShouldDropPacket(thePickle, message): 368 self.machoNet.LogError('Attempted to send a deadly (len=', len(thePickle), ') packet to client(s), PACKET DROPPED') 369 self.machoNet.LogError('Packet =', repr(message)[:1024]) 370 self.machoNet.LogError('Pickle starts with =', repr(thePickle)[:1024]) 371 self._buffer_drop_message_metrics(message, thePickle) 372 return 373 self._buffer_write_message_metrics(message, thePickle) 374 self.transport.Write(thePickle) 375 self.machoNet.dataSent.Add(len(thePickle)) 376 377 @bluepy.TimedFunction('machoNet::MachoTransport::Read') 378 def Read(self): 379 self.currentReaders += 1 380 try: 381 thePickle = self.transport.Read() 382 read_time = datetime.datetime.utcnow() 383 finally: 384 self.currentReaders -= 1 385 386 if getattr(self, 'userID', None) and len(thePickle) > 100000: 387 self.machoNet.LogWarn('Read a ', len(thePickle), ' byte packet (before decompression) from userID=', getattr(self, 'userID', 'non-user'), ' on address ', self.transport.address) 388 elif len(thePickle) > 5000000: 389 self.machoNet.LogWarn('Read a ', len(thePickle), ' byte packet (before decompression) from userID=', getattr(self, 'userID', 'non-user'), ' on address ', self.transport.address) 390 if thePickle[0] not in '}~': 391 before = len(thePickle) 392 try: 393 with bluepy.Timer('machoNet::MachoTransport::Read::DeCompress'): 394 thePickle = zlib.decompress(thePickle) 395 except zlib.error as e: 396 raise RuntimeError('Decompression Failure: ' + strx(e)) 397 398 after = len(thePickle) 399 if after <= before: 400 self.machoNet.LogError('Decompress shrank data from ', before, ' to ', after, ' bytes') 401 else: 402 self.machoNet.decompressedBytes.Add(after - before) 403 if getattr(self, 'userID', None) and len(thePickle) > 100000: 404 self.machoNet.LogWarn('Read a ', len(thePickle), ' byte packet (after decompression, if appropriate) from userID=', getattr(self, 'userID', 'non-user'), ' on address ', self.transport.address) 405 elif len(thePickle) > 5000000: 406 self.machoNet.LogWarn('Read a ', len(thePickle), ' byte packet (after decompression, if appropriate) from userID=', getattr(self, 'userID', 'non-user'), ' on address ', self.transport.address) 407 if self.clientID: 408 self.machoNet.dataReceived.Add(len(thePickle)) 409 else: 410 self.machoNet.dataReceived.AddFrom(self.nodeID, len(thePickle)) 411 try: 412 message = macho.Loads(thePickle) 413 except GPSTransportClosed as e: 414 self.transport.Close(**e.GetCloseArgs()) 415 raise 416 except Exception: 417 if self.transportName == 'tcp:packet:client': 418 self._LogPotentialAttackAndClose(thePickle) 419 raise 420 421 if not isinstance(message, MachoPacket): 422 self._LogPotentialAttackAndClose(thePickle) 423 raise GPSTransportClosed('An improperly formed packet was received') 424 message.SetPickle(thePickle) 425 message.SetReadTime(read_time) 426 if macho.mode == 'client' and message.source.addressType == const.ADDRESS_TYPE_NODE and message.destination.addressType == const.ADDRESS_TYPE_BROADCAST: 427 message.oob['sn'] = self.machoNet.notifySequenceIDByNodeID[message.source.nodeID] 428 self.machoNet.notifySequenceIDByNodeID[message.source.nodeID] += 1 429 if self.transportName == 'tcp:packet:client': 430 message.source = MachoAddress(clientID=self.clientID, callID=message.source.callID) 431 if message.contextKey: 432 self._LogPotentialAttackAndClose(thePickle) 433 raise Exception('Packet containing contextKey received on a client transport. Hack?') 434 if hasattr(self, 'userID'): 435 message.userID = self.userID 436 self.LogInfo('Read: ', message) 437 if macho.mode == 'proxy': 438 for objectID, refID in message.oob.get('OID-', {}).iteritems(): 439 s = self.sessions.get(self.clientID, None) 440 if s is None: 441 s = self.sessions.get(self.nodeID, None) 442 if s is not None: 443 s.UnregisterMachoObject(objectID, refID) 444 445 ls = localstorage.GetLocalStorage() 446 ls['journeyID'] = message.journeyID 447 self._buffer_read_message_metrics(message, thePickle) 448 return message 449 450 def _LogPotentialAttackAndClose(self, thePickle): 451 log.LogTraceback() 452 address = self.transport.address 453 self.transport.Close('An improperly formed or damaged packet was received from your client') 454 db = self.machoNet.session.ConnectToAnyService('DB2') 455 db.CallProc('zcluster.Attacks_Insert', getattr(self, 'userID', None), address.split(':')[0], int(address.split(':')[1]), strx(thePickle[:2000].replace('\\', '\\\\').replace('\x00', '\\0'))) 456 457 def TagPacketSizes(self, req, rsp = None): 458 ctk = localstorage.GetLocalStorage().get('calltimer.key', None) 459 if ctk is not None: 460 ct = base.GetCallTimes() 461 try: 462 s = session._Obj() 463 except: 464 sys.exc_clear() 465 s = session 466 467 if s: 468 if not s.role & ROLE_SERVICE: 469 if macho.mode == 'client': 470 ct = (ct[2], s.calltimes) 471 else: 472 ct = (ct[0], s.calltimes) 473 else: 474 ct = (ct[1], s.calltimes) 475 else: 476 ct = (ct[1],) 477 for calltimes in ct: 478 if ctk in calltimes: 479 calltimes[ctk][4] += req.GetPickleSize(self.machoNet) 480 if rsp is not None: 481 with bluepy.Timer('machoNet::HandleMessage::SessionCall::TagPacketSizes::GetPickleSize::Rsp'): 482 calltimes[ctk][5] += rsp.GetPickleSize(self.machoNet) 483 484 def InstallSessionIfRequired(self, serviceName, methodName): 485 if macho.mode != 'server': 486 raise RuntimeError('InstallSessionIfRequired called on %s (should only be called on server)' % macho.mode) 487 if not session.contextOnly: 488 raise RuntimeError('InstallSessionIfRequired called on full session') 489 sess = session.GetActualSession() 490 if sess.role & ROLE_SERVICE != 0: 491 return False 492 service = sm.GetService(serviceName) 493 if methodName != 'MachoResolveObject' and (hasattr(service, 'DoSessionChanging') or hasattr(service, 'ProcessSessionChange') or hasattr(service, 'OnSessionChanged')): 494 self.machoNet.LogInfo('CTXSESS: Installing session ', sess.sid, ' to call ', serviceName, '::', methodName) 495 sess.contextOnly = False 496 self.sessions[sess.clientID] = sess 497 del self.contextSessions[sess.clientID] 498 sess.DelayedInitialStateChange() 499 return True 500 self.machoNet.LogInfo('CTXSESS: no need to install ', sess.sid, ' before calling ', serviceName, '::', methodName) 501 return False 502 503 def RemoveSessionFromServer(self, sess): 504 if not (macho.mode == 'proxy' and self.transportName == 'tcp:packet:machoNet'): 505 raise RuntimeError('RemoveSessionFromServer should only be called on a proxy for a proxy/server transport') 506 clientTransportID = self.machoNet.transportIDbySessionID.get(sess.sid, None) 507 if clientTransportID is not None: 508 clientTransport = self.machoNet.transportsByID[clientTransportID] 509 try: 510 del clientTransport.dependants[self.transportID] 511 self.machoNet.LogInfo('Removed transport', self.transportID, 'from the dependants of client transport', clientTransportID) 512 except KeyError: 513 pass 514 515 try: 516 self.machoNet.LogInfo('Sending TransportClosed to remove session ', sess.sid, ' (client ', sess.clientID, ') on transport', self.transportID) 517 self.Write(TransportClosed(clientID=sess.clientID, isRemote=0)) 518 except Exception: 519 errStr = 'Write failed in RemoveSessionFromServer(%s)' % (sess.sid,) 520 log.LogException(errStr) 521 522 @bluepy.TimedFunction('machoNetTransport::SessionCall') 523 def SessionCall(self, packet): 524 try: 525 while macho.mode == 'client' and self.machoNet.authenticating: 526 blue.pyos.synchro.SleepWallclock(250) 527 528 msgSession, channel, theID = self._SessionAndChannelAndIDFromPacket(packet) 529 with MachoCallOrNotification(self, msgSession, packet) as currentcall: 530 mask = msgSession.Masquerade({'base.currentcall': weakref.ref(currentcall)}) 531 packet.srcTransport = self 532 try: 533 ret = None 534 if getattr(packet.destination, 'service', None) is not None: 535 if packet.destination.service not in msgSession.connectedServices: 536 try: 537 msgSession.connectedServices[packet.destination.service] = packet.service = msgSession.ConnectToService(packet.destination.service, remote=1) 538 except (ServiceNotFound, RuntimeError) as e: 539 ret = packet.ErrorResponse(cluster.MACHONETERR_WRAPPEDEXCEPTION, (macho.DumpsSanitized(e),)) 540 sys.exc_clear() 541 542 else: 543 packet.service = msgSession.connectedServices[packet.destination.service] 544 if ret is None: 545 if channel in self.machoNet.channelHandlersUp: 546 ret = self.machoNet.channelHandlersUp[channel].CallUp(packet) 547 else: 548 ret = packet.ErrorResponse(cluster.MACHONETERR_UNMACHOCHANNEL, 'The specified channel is not present on this server') 549 self.TagPacketSizes(packet, ret) 550 return ret 551 finally: 552 if msgSession.clientID not in (theID, packet.contextKey): 553 if msgSession.clientID is not None: 554 self.machoNet.LogError('Cleaning session ', msgSession.clientID, " because it's ID doesn't match ", theID) 555 self._CleanupSession(theID) 556 mask.UnMask() 557 if hasattr(packet, 'service'): 558 delattr(packet, 'service') 559 560 except Exception as e: 561 log.LogException('Error in session call') 562 raise 563 564 def SessionNotification(self, packet): 565 while macho.mode == 'client' and self.machoNet.authenticating: 566 blue.pyos.synchro.SleepWallclock(250) 567 568 msgSession, channel, theID = self._SessionAndChannelAndIDFromPacket(packet) 569 with MachoCallOrNotification(self, msgSession, packet) as currentcall: 570 mask = msgSession.Masquerade({'base.currentcall': weakref.ref(currentcall)}) 571 packet.srcTransport = self 572 try: 573 if getattr(packet.destination, 'service', None) is not None: 574 if packet.destination.service not in msgSession.connectedServices: 575 msgSession.connectedServices[packet.destination.service] = msgSession.ConnectToService(packet.destination.service, remote=1) 576 packet.service = msgSession.connectedServices[packet.destination.service] 577 if channel in self.machoNet.channelHandlersUp: 578 self.machoNet.channelHandlersUp[channel].NotifyUp(packet) 579 else: 580 self.LogInfo('Notification received for channel ', channel, ', but no GPCS handler available for that particular channel of transport', self) 581 self.TagPacketSizes(packet) 582 finally: 583 if msgSession.clientID not in (theID, packet.contextKey): 584 if msgSession.clientID is not None: 585 self.machoNet.LogError('Cleaning session ', msgSession.clientID, " because it's ID doesn't match ", theID) 586 self._CleanupSession(theID) 587 mask.UnMask() 588 if hasattr(packet, 'service'): 589 delattr(packet, 'service') 590 591 def _SessionFromClientID(self, clientID): 592 s = self.sessions.get(clientID, None) 593 if s is not None: 594 return s[0] 595 else: 596 return 597 598 def _SessionAndChannelAndIDFromPacket(self, packet): 599 if macho.mode == 'server' and packet.contextKey: 600 clientID = packet.source.nodeID 601 rsess = self.sessions.get(packet.contextKey, None) 602 if rsess is None: 603 rsess = self.contextSessions[packet.contextKey] 604 else: 605 clientID, rsess = self._AssociateWithSession(packet) 606 channel = None 607 if packet is not None: 608 channel = macho.packetTypeChannelMap.get(packet.command, None) 609 return (rsess, channel, clientID) 610 611 def _CleanupSession(self, theID): 612 if macho.mode != 'client': 613 sess = None 614 if theID in self.sessions: 615 sess = self.sessions[theID] 616 del self.sessions[theID] 617 if not sess and theID in self.contextSessions: 618 sess = self.contextSessions[theID] 619 del self.contextSessions[theID] 620 if sess: 621 base.CloseSession(sess) 622 623 def LockSession(self, sess, packet): 624 if packet is not None and not packet.command % 2: 625 if hasattr(packet, 'sessionVersion'): 626 sleepNum = 1 627 while packet.sessionVersion > sess.version: 628 logargs1 = ('Sleep #', 629 sleepNum, 630 ' while waiting for session change to complete. The packet is destined for session version ', 631 packet.sessionVersion, 632 ' but the session(', 633 sess.sid, 634 ') is currently version ', 635 sess.version) 636 logargs2 = ('packet=', packet) 637 if sleepNum % 250 == 0: 638 self.LogError(*logargs1) 639 self.LogError(*logargs2) 640 else: 641 self.LogInfo(*logargs1) 642 self.LogInfo(*logargs2) 643 blue.pyos.synchro.SleepWallclock(500) 644 sleepNum += 1 645 646 self.machoNet.WaitForSequenceNumber(packet.source, packet.oob.get('sn', 0)) 647 if sess.rwlock: 648 if packet is None or packet.command in self.__sessioninitorchangenotification__: 649 sess.rwlock.WRLock() 650 if sess.role & ROLE_SERVICE: 651 sess.rwlock.Unlock() 652 sess.LogSessionError('SESSIONFUXUP', 'Trying to run a session init or change in a service session context') 653 log.LogTraceback() 654 raise RuntimeError('Session map failure, attempting to run a session init or change in a service session context') 655 else: 656 sess.rwlock.RDLock() 657 658 def UnLockSession(self, sess, tasklet = None): 659 if sess.rwlock: 660 sess.rwlock.Unlock(tasklet) 661 662 def _AssociateWithSession(self, packet = None, forceNodeID = None): 663 if forceNodeID is not None: 664 nodeID = forceNodeID 665 serviceSession = True 666 elif macho.mode == 'client': 667 nodeID = 0 668 serviceSession = False 669 elif not packet.command % 2: 670 if packet.source.addressType == const.ADDRESS_TYPE_CLIENT: 671 nodeID = packet.source.clientID 672 serviceSession = False 673 else: 674 nodeID = packet.source.nodeID 675 serviceSession = True 676 elif packet.destination.addressType == const.ADDRESS_TYPE_CLIENT: 677 nodeID = packet.destination.clientID 678 serviceSession = False 679 else: 680 nodeID = packet.destination.nodeID 681 serviceSession = True 682 sess = self.sessions.get(nodeID, None) 683 if not sess: 684 sess = self.contextSessions.get(nodeID, None) 685 if not sess: 686 installSession = True 687 if macho.mode == 'client': 688 sess = session 689 elif serviceSession: 690 sess = base.GetServiceSession('remote:%d' % nodeID, True) 691 elif not packet.command % 2: 692 if nodeID in self.queuedClientCalls: 693 self.queuedClientCalls[nodeID].receive() 694 if isinstance(packet, SessionInitialStateNotification): 695 sess = base.CreateSession(packet.sid, packet.sessionType) 696 else: 697 installSession = False 698 sess = self.sessions.get(nodeID, self.contextSessions.get(nodeID, None)) 699 if not sess: 700 raise SessionUnavailable('Unable to load session for request') 701 else: 702 raise UnMachoDestination('Failed session association: cmd = %s, mode = %s' % (packet.command, macho.mode)) 703 if installSession: 704 sess.__dict__['clientID'] = nodeID 705 if sess.contextOnly: 706 self.contextSessions[nodeID] = sess 707 else: 708 self.sessions[nodeID] = sess 709 if serviceSession: 710 role = sess.__dict__['role'] | ROLE_SERVICE | ROLE_REMOTESERVICE 711 sess.SetAttributes({'role': role}) 712 sess.LogSessionHistory('machoNet associated remote service session with clientID/nodeID %s' % nodeID) 713 else: 714 sess.__dict__['rwlock'] = uthread.RWLock(('sessions', sess.sid)) 715 packetUserID = None if packet is None else packet.userID 716 if packetUserID is not None: 717 sess.SetAttributes({'userid': packetUserID}) 718 sess.LogSessionHistory('machoNet associated session with clientID %s and userID %s' % (nodeID, packetUserID)) 719 sess.lastRemoteCall = blue.os.GetWallclockTime() 720 return (nodeID, sess) 721 722 def _buffer_write_message_metrics(self, message, pickle): 723 if not self._machonet_metrics_enabled(): 724 return 725 tags = self._get_transport_metrics_tags(message) 726 metrics.buffer_write_message_metrics(message, pickle, tags) 727 728 def _buffer_drop_message_metrics(self, message, pickle): 729 if not self._machonet_metrics_enabled(): 730 return 731 tags = self._get_transport_metrics_tags(message) 732 metrics.buffer_drop_message_metrics(message, pickle, tags) 733 734 def _buffer_read_message_metrics(self, message, pickle): 735 if not self._machonet_metrics_enabled(): 736 return 737 tags = self._get_transport_metrics_tags(message) 738 metrics.buffer_read_message_metrics(message, pickle, tags) 739 740 def _machonet_metrics_enabled(self): 741 return macho.mode != 'client' and self.machoNet.metricsEnabled() 742 743 def fetch_metric_tags(self): 744 self.message_type_metric_tags = metrics.get_destination_type_labels(self.nodeID, self.transportName) 745 746 def _get_transport_metrics_tags(self, message): 747 return self.message_type_metric_tags[message.__class__] 748 749 750 class MachoCallOrNotification(object): 751 752 def __init__(self, transport, sess, packet): 753 self.transport, self.sess, self.packet = transport, sess, packet 754 self.tasklet = stackless.getcurrent() 755 756 def __enter__(self): 757 self.transport.LockSession(self.sess, self.packet) 758 return self 759 760 def __exit__(self, e, v, tb): 761 self.Unlock(None) 762 763 def UnLockSession(self): 764 self.Unlock(self.tasklet) 765 766 def Unlock(self, tasklet): 767 if self.sess: 768 s = self.sess 769 t = self.transport 770 self.transport = self.sess = self.tasklet = None 771 t.UnLockSession(s, tasklet) 772 773 def __repr__(self): 774 return '<MachoCallOrNotification, packet=%r>' % (self.packet,)