xmlrpcClient.py (5634B)
1 #Embedded file name: C:\BuildAgent\work\20d73f1b24eed06e\eve\release\V24.01\carbon\common\script\net\xmlrpcClient.py 2 from future.utils import raise_ 3 import collections 4 import contextlib 5 import socket 6 import urlparse 7 import xmlrpclib 8 import locks 9 from carbon.common.script.sys.service import Service 10 from carbon.common.script.sys.serviceConst import ROLE_SERVICE 11 12 class CachedMultiMap(object): 13 14 def __init__(self, highWater = 10, lowWater = None): 15 self.map = collections.defaultdict(list) 16 self.times = {} 17 self.time = 0 18 self.highWater = highWater 19 if lowWater is None: 20 lowWater = self.highWater / 2 21 self.lowWater = min(lowWater, self.highWater) 22 23 def __len__(self): 24 return len(self.times) 25 26 def Insert(self, key, val): 27 time = self.time 28 self.time += 1 29 e = (time, val) 30 self.map[key].append(e) 31 self.times[time] = key 32 if len(self.times) > self.highWater: 33 self.Trim(self.lowWater) 34 35 def Get(self, key): 36 l = self.map[key] 37 t0, v = l[-1] 38 t = self.time 39 self.time += 1 40 del self.times[t0] 41 self.times[t] = key 42 l[-1] = (t, v) 43 return v 44 45 def Pop(self, key, first = False): 46 l = self.map[key] 47 try: 48 t0, v = l.pop(0 if first else -1) 49 except IndexError: 50 raise_(KeyError, key) 51 52 del self.times[t0] 53 if not l: 54 del self.map[key] 55 return v 56 57 def Trim(self, max): 58 times = self.times 59 if len(times) > max: 60 k = self.times.keys() 61 k.sort() 62 k = k[:-max] 63 Pop = self.Pop 64 for t in k: 65 Pop(times[t], True) 66 67 def Clear(self): 68 self.map = collections.defaultdict(list) 69 self.times = {} 70 self.time = self.n = 0 71 72 73 class RetryProxyMethod: 74 75 def __init__(self, send, name, retries = 3): 76 self.__send = send 77 self.__name = name 78 self.__retries = retries 79 80 def __getattr__(self, name): 81 return RetryProxyMethod(self.__send, '.'.join((self.__name, name)), self.__retries) 82 83 def __call__(self, *args): 84 for i in xrange(self.__retries): 85 try: 86 return self.__send(self.__name, args) 87 except socket.error: 88 if i == self.__retries - 1: 89 raise 90 91 92 class RetryProxy(xmlrpclib.ServerProxy): 93 94 def __init__(self, uri, transport = None, encoding = None, verbose = 0, allow_none = 0, use_datetime = 0, retries = 3): 95 xmlrpclib.ServerProxy.__init__(self, uri, transport, encoding, verbose, allow_none, use_datetime) 96 self.__retries = retries 97 98 def __getattr__(self, name): 99 return RetryProxyMethod(self._ServerProxy__request, name, self.__retries) 100 101 102 class XmlrpcClient(Service): 103 __exportedcalls__ = {'GetProxy': [ROLE_SERVICE], 104 'ReturnProxy': [ROLE_SERVICE], 105 'Proxy': [ROLE_SERVICE]} 106 __guid__ = 'svc.xmlrpcClient' 107 __dependencies__ = [] 108 109 def Run(self, memStream = None): 110 self.proxiesByKey = CachedMultiMap(10) 111 self.dnsMap = {} 112 113 def Stop(self, memStream): 114 self.proxiesByKey.Clear() 115 116 def GetProxy(self, uri, proxyClass = xmlrpclib.ServerProxy, proxyArgs = ()): 117 key = (uri, proxyClass, proxyArgs) 118 try: 119 return (key, self.proxiesByKey.Pop(key)) 120 except KeyError: 121 uri = self.FixUri(uri) 122 return (key, proxyClass(uri, *proxyArgs)) 123 124 def ReturnProxy(self, key, proxy, discard = False): 125 if not discard: 126 self.proxiesByKey.Insert(key, proxy) 127 128 def FixUri(self, uri): 129 parts = urlparse.urlsplit(uri) 130 key = parts[1] 131 try: 132 while True: 133 netloc = self.dnsMap[key] 134 if isinstance(netloc, locks.Event): 135 netloc.wait() 136 else: 137 break 138 139 except KeyError: 140 event = locks.Event() 141 self.dnsMap[key] = event 142 try: 143 netloc = key 144 hostport = netloc.split(':') 145 if len(hostport) == 2: 146 host, port = hostport 147 else: 148 host, port = hostport[0], None 149 try: 150 info = socket.getaddrinfo(host, port) 151 info = [ e for e in info if e[0] == socket.AF_INET ] 152 if info: 153 if port is not None: 154 host, port = info[0][4] 155 netloc = host + ':' + str(port) 156 else: 157 netloc = host 158 except socket.gaierror: 159 pass 160 161 except: 162 del self.dnsMap[key] 163 else: 164 self.dnsMap[key] = netloc 165 finally: 166 event.set() 167 168 if netloc != parts[1]: 169 uri = urlparse.urlunsplit((parts[0],) + (netloc,) + parts[2:]) 170 return uri 171 172 def Call(self, uri, method, *args, **kw): 173 with self.ProxyContext(uri, kw.get('proxyClass', xmlrpclib.ServerProxy), kw.get('proxyArgs', ())) as p: 174 return getattr(p, method)(*args) 175 176 @contextlib.contextmanager 177 def Proxy(self, uri, proxyClass = xmlrpclib.ServerProxy, proxyArgs = ()): 178 k, p = self.GetProxy(uri, proxyClass, proxyArgs) 179 discard = False 180 try: 181 yield p 182 except socket.error: 183 discard = True 184 raise 185 finally: 186 self.ReturnProxy(k, p, discard)