Lib/idlelib/rpc.py

   1 """RPC Implemention, originally written for the Python Idle IDE
   2
   3 For security reasons, GvR requested that Idle's Python execution server process
   4 connect to the Idle process, which listens for the connection.  Since Idle has
   5 has only one client per server, this was not a limitation.
   6
   7    +---------------------------------+ +-------------+
   8    | SocketServer.BaseRequestHandler | | SocketIO    |
   9    +---------------------------------+ +-------------+
  10                    ^                   | register()  |
  11                    |                   | unregister()|
  12                    |                   +-------------+
  13                    |                      ^  ^
  14                    |                      |  |
  15                    | + -------------------+  |
  16                    | |                       |
  17    +-------------------------+        +-----------------+
  18    | RPCHandler              |        | RPCClient       |
  19    | [attribute of RPCServer]|        |                 |
  20    +-------------------------+        +-----------------+
  21
  22 The RPCServer handler class is expected to provide register/unregister methods.
  23 RPCHandler inherits the mix-in class SocketIO, which provides these methods.
  24
  25 See the Idle run.main() docstring for further information on how this was
  26 accomplished in Idle.
  27
  28 """
  29
  30 import sys
  31 import os
  32 import socket
  33 import select
  34 import SocketServer
  35 import struct
  36 import cPickle as pickle
  37 import threading
  38 import Queue
  39 import traceback
  40 import copy_reg
  41 import types
  42 import marshal
  43
  44
  45 def unpickle_code(ms):
  46     co = marshal.loads(ms)
  47     assert isinstance(co, types.CodeType)
  48     return co
  49
  50 def pickle_code(co):
  51     assert isinstance(co, types.CodeType)
  52     ms = marshal.dumps(co)
  53     return unpickle_code, (ms,)
  54
  55 # XXX KBK 24Aug02 function pickling capability not used in Idle
  56 #  def unpickle_function(ms):
  57 #      return ms
  58
  59 #  def pickle_function(fn):
  60 #      assert isinstance(fn, type.FunctionType)
  61 #      return repr(fn)
  62
  63 copy_reg.pickle(types.CodeType, pickle_code, unpickle_code)
  64 # copy_reg.pickle(types.FunctionType, pickle_function, unpickle_function)
  65
  66 BUFSIZE = 8*1024
  67 LOCALHOST = '127.0.0.1'
  68
  69 class RPCServer(SocketServer.TCPServer):
  70
  71     def __init__(self, addr, handlerclass=None):
  72         if handlerclass is None:
  73             handlerclass = RPCHandler
  74         SocketServer.TCPServer.__init__(self, addr, handlerclass)
  75
  76     def server_bind(self):
  77         "Override TCPServer method, no bind() phase for connecting entity"
  78         pass
  79
  80     def server_activate(self):
  81         """Override TCPServer method, connect() instead of listen()
  82
  83         Due to the reversed connection, self.server_address is actually the
  84         address of the Idle Client to which we are connecting.
  85
  86         """
  87         self.socket.connect(self.server_address)
  88
  89     def get_request(self):
  90         "Override TCPServer method, return already connected socket"
  91         return self.socket, self.server_address
  92
  93     def handle_error(self, request, client_address):
  94         """Override TCPServer method
  95
  96         Error message goes to __stderr__.  No error message if exiting
  97         normally or socket raised EOF.  Other exceptions not handled in
  98         server code will cause os._exit.
  99
 100         """
 101         try:
 102             raise
 103         except SystemExit:
 104             raise
 105         except:
 106             erf = sys.__stderr__
 107             print>>erf, '\n' + '-'*40
 108             print>>erf, 'Unhandled server exception!'
 109             print>>erf, 'Thread: %s' % threading.currentThread().getName()
 110             print>>erf, 'Client Address: ', client_address
 111             print>>erf, 'Request: ', repr(request)
 112             traceback.print_exc(file=erf)
 113             print>>erf, '\n*** Unrecoverable, server exiting!'
 114             print>>erf, '-'*40
 115             os._exit(0)
 116
 117 #----------------- end class RPCServer --------------------
 118
 119 objecttable = {}
 120 request_queue = Queue.Queue(0)
 121 response_queue = Queue.Queue(0)
 122
 123
 124 class SocketIO(object):
 125
 126     nextseq = 0
 127
 128     def __init__(self, sock, objtable=None, debugging=None):
 129         self.sockthread = threading.currentThread()
 130         if debugging is not None:
 131             self.debugging = debugging
 132         self.sock = sock
 133         if objtable is None:
 134             objtable = objecttable
 135         self.objtable = objtable
 136         self.responses = {}
 137         self.cvars = {}
 138
 139     def close(self):
 140         sock = self.sock
 141         self.sock = None
 142         if sock is not None:
 143             sock.close()
 144
 145     def exithook(self):
 146         "override for specific exit action"
 147         os._exit()
 148
 149     def debug(self, *args):
 150         if not self.debugging:
 151             return
 152         s = self.location + " " + str(threading.currentThread().getName())
 153         for a in args:
 154             s = s + " " + str(a)
 155         print>>sys.__stderr__, s
 156
 157     def register(self, oid, object):
 158         self.objtable[oid] = object
 159
 160     def unregister(self, oid):
 161         try:
 162             del self.objtable[oid]
 163         except KeyError:
 164             pass
 165
 166     def localcall(self, seq, request):
 167         self.debug("localcall:", request)
 168         try:
 169             how, (oid, methodname, args, kwargs) = request
 170         except TypeError:
 171             return ("ERROR", "Bad request format")
 172         if not self.objtable.has_key(oid):
 173             return ("ERROR", "Unknown object id: %r" % (oid,))
 174         obj = self.objtable[oid]
 175         if methodname == "__methods__":
 176             methods = {}
 177             _getmethods(obj, methods)
 178             return ("OK", methods)
 179         if methodname == "__attributes__":
 180             attributes = {}
 181             _getattributes(obj, attributes)
 182             return ("OK", attributes)
 183         if not hasattr(obj, methodname):
 184             return ("ERROR", "Unsupported method name: %r" % (methodname,))
 185         method = getattr(obj, methodname)
 186         try:
 187             if how == 'CALL':
 188                 ret = method(*args, **kwargs)
 189                 if isinstance(ret, RemoteObject):
 190                     ret = remoteref(ret)
 191                 return ("OK", ret)
 192             elif how == 'QUEUE':
 193                 request_queue.put((seq, (method, args, kwargs)))
 194                 return("QUEUED", None)
 195             else:
 196                 return ("ERROR", "Unsupported message type: %s" % how)
 197         except SystemExit:
 198             raise
 199         except socket.error:
 200             raise
 201         except:
 202             msg = "*** Internal Error: rpc.py:SocketIO.localcall()\n\n"\
 203                   " Object: %s \n Method: %s \n Args: %s\n"
 204             print>>sys.__stderr__, msg % (oid, method, args)
 205             traceback.print_exc(file=sys.__stderr__)
 206             return ("EXCEPTION", None)
 207
 208     def remotecall(self, oid, methodname, args, kwargs):
 209         self.debug("remotecall:asynccall: ", oid, methodname)
 210         seq = self.asynccall(oid, methodname, args, kwargs)
 211         return self.asyncreturn(seq)
 212
 213     def remotequeue(self, oid, methodname, args, kwargs):
 214         self.debug("remotequeue:asyncqueue: ", oid, methodname)
 215         seq = self.asyncqueue(oid, methodname, args, kwargs)
 216         return self.asyncreturn(seq)
 217
 218     def asynccall(self, oid, methodname, args, kwargs):
 219         request = ("CALL", (oid, methodname, args, kwargs))
 220         seq = self.newseq()
 221         if threading.currentThread() != self.sockthread:
 222             cvar = threading.Condition()
 223             self.cvars[seq] = cvar
 224         self.debug(("asynccall:%d:" % seq), oid, methodname, args, kwargs)
 225         self.putmessage((seq, request))
 226         return seq
 227
 228     def asyncqueue(self, oid, methodname, args, kwargs):
 229         request = ("QUEUE", (oid, methodname, args, kwargs))
 230         seq = self.newseq()
 231         if threading.currentThread() != self.sockthread:
 232             cvar = threading.Condition()
 233             self.cvars[seq] = cvar
 234         self.debug(("asyncqueue:%d:" % seq), oid, methodname, args, kwargs)
 235         self.putmessage((seq, request))
 236         return seq
 237
 238     def asyncreturn(self, seq):
 239         self.debug("asyncreturn:%d:call getresponse(): " % seq)
 240         response = self.getresponse(seq, wait=0.05)
 241         self.debug(("asyncreturn:%d:response: " % seq), response)
 242         return self.decoderesponse(response)
 243
 244     def decoderesponse(self, response):
 245         how, what = response
 246         if how == "OK":
 247             return what
 248         if how == "QUEUED":
 249             return None
 250         if how == "EXCEPTION":
 251             self.debug("decoderesponse: EXCEPTION")
 252             return None
 253         if how == "EOF":
 254             self.debug("decoderesponse: EOF")
 255             self.decode_interrupthook()
 256             return None
 257         if how == "ERROR":
 258             self.debug("decoderesponse: Internal ERROR:", what)
 259             raise RuntimeError, what
 260         raise SystemError, (how, what)
 261
 262     def decode_interrupthook(self):
 263         ""
 264         raise EOFError
 265
 266     def mainloop(self):
 267         """Listen on socket until I/O not ready or EOF
 268
 269         pollresponse() will loop looking for seq number None, which
 270         never comes, and exit on EOFError.
 271
 272         """
 273         try:
 274             self.getresponse(myseq=None, wait=0.05)
 275         except EOFError:
 276             self.debug("mainloop:return")
 277             return
 278
 279     def getresponse(self, myseq, wait):
 280         response = self._getresponse(myseq, wait)
 281         if response is not None:
 282             how, what = response
 283             if how == "OK":
 284                 response = how, self._proxify(what)
 285         return response
 286
 287     def _proxify(self, obj):
 288         if isinstance(obj, RemoteProxy):
 289             return RPCProxy(self, obj.oid)
 290         if isinstance(obj, types.ListType):
 291             return map(self._proxify, obj)
 292         # XXX Check for other types -- not currently needed
 293         return obj
 294
 295     def _getresponse(self, myseq, wait):
 296         self.debug("_getresponse:myseq:", myseq)
 297         if threading.currentThread() is self.sockthread:
 298             # this thread does all reading of requests or responses
 299             while 1:
 300                 response = self.pollresponse(myseq, wait)
 301                 if response is not None:
 302                     return response
 303         else:
 304             # wait for notification from socket handling thread
 305             cvar = self.cvars[myseq]
 306             cvar.acquire()
 307             while not self.responses.has_key(myseq):
 308                 cvar.wait()
 309             response = self.responses[myseq]
 310             self.debug("_getresponse:%s: thread woke up: response: %s" %
 311                        (myseq, response))
 312             del self.responses[myseq]
 313             del self.cvars[myseq]
 314             cvar.release()
 315             return response
 316
 317     def newseq(self):
 318         self.nextseq = seq = self.nextseq + 2
 319         return seq
 320
 321     def putmessage(self, message):
 322         self.debug("putmessage:%d:" % message[0])
 323         try:
 324             s = pickle.dumps(message)
 325         except pickle.PicklingError:
 326             print >>sys.__stderr__, "Cannot pickle:", repr(message)
 327             raise
 328         s = struct.pack("<i", len(s)) + s
 329         while len(s) > 0:
 330             try:
 331                 r, w, x = select.select([], [self.sock], [])
 332                 n = self.sock.send(s[:BUFSIZE])
 333             except (AttributeError, TypeError):
 334                 raise IOError, "socket no longer exists"
 335             except socket.error:
 336                 raise
 337             else:
 338                 s = s[n:]
 339
 340     buffer = ""
 341     bufneed = 4
 342     bufstate = 0 # meaning: 0 => reading count; 1 => reading data
 343
 344     def pollpacket(self, wait):
 345         self._stage0()
 346         if len(self.buffer) < self.bufneed:
 347             r, w, x = select.select([self.sock.fileno()], [], [], wait)
 348             if len(r) == 0:
 349                 return None
 350             try:
 351                 s = self.sock.recv(BUFSIZE)
 352             except socket.error:
 353                 raise EOFError
 354             if len(s) == 0:
 355                 raise EOFError
 356             self.buffer += s
 357             self._stage0()
 358         return self._stage1()
 359
 360     def _stage0(self):
 361         if self.bufstate == 0 and len(self.buffer) >= 4:
 362             s = self.buffer[:4]
 363             self.buffer = self.buffer[4:]
 364             self.bufneed = struct.unpack("<i", s)[0]
 365             self.bufstate = 1
 366
 367     def _stage1(self):
 368         if self.bufstate == 1 and len(self.buffer) >= self.bufneed:
 369             packet = self.buffer[:self.bufneed]
 370             self.buffer = self.buffer[self.bufneed:]
 371             self.bufneed = 4
 372             self.bufstate = 0
 373             return packet
 374
 375     def pollmessage(self, wait):
 376         packet = self.pollpacket(wait)
 377         if packet is None:
 378             return None
 379         try:
 380             message = pickle.loads(packet)
 381         except pickle.UnpicklingError:
 382             print >>sys.__stderr__, "-----------------------"
 383             print >>sys.__stderr__, "cannot unpickle packet:", repr(packet)
 384             traceback.print_stack(file=sys.__stderr__)
 385             print >>sys.__stderr__, "-----------------------"
 386             raise
 387         return message
 388
 389     def pollresponse(self, myseq, wait):
 390         """Handle messages received on the socket.
 391
 392         Some messages received may be asynchronous 'call' or 'queue' requests,
 393         and some may be responses for other threads.
 394
 395         'call' requests are passed to self.localcall() with the expectation of
 396         immediate execution, during which time the socket is not serviced.
 397
 398         'queue' requests are used for tasks (which may block or hang) to be
 399         processed in a different thread.  These requests are fed into
 400         request_queue by self.localcall().  Responses to queued requests are
 401         taken from response_queue and sent across the link with the associated
 402         sequence numbers.  Messages in the queues are (sequence_number,
 403         request/response) tuples and code using this module removing messages
 404         from the request_queue is responsible for returning the correct
 405         sequence number in the response_queue.
 406
 407         pollresponse() will loop until a response message with the myseq
 408         sequence number is received, and will save other responses in
 409         self.responses and notify the owning thread.
 410
 411         """
 412         while 1:
 413             # send queued response if there is one available
 414             try:
 415                 qmsg = response_queue.get(0)
 416             except Queue.Empty:
 417                 pass
 418             else:
 419                 seq, response = qmsg
 420                 message = (seq, ('OK', response))
 421                 self.putmessage(message)
 422             # poll for message on link
 423             try:
 424                 message = self.pollmessage(wait)
 425                 if message is None:  # socket not ready
 426                     return None
 427             except EOFError:
 428                 self.handle_EOF()
 429                 return None
 430             except AttributeError:
 431                 return None
 432             seq, resq = message
 433             how = resq[0]
 434             self.debug("pollresponse:%d:myseq:%s" % (seq, myseq))
 435             # process or queue a request
 436             if how in ("CALL", "QUEUE"):
 437                 self.debug("pollresponse:%d:localcall:call:" % seq)
 438                 response = self.localcall(seq, resq)
 439                 self.debug("pollresponse:%d:localcall:response:%s"
 440                            % (seq, response))
 441                 if how == "CALL":
 442                     self.putmessage((seq, response))
 443                 elif how == "QUEUE":
 444                     # don't acknowledge the 'queue' request!
 445                     pass
 446                 continue
 447             # return if completed message transaction
 448             elif seq == myseq:
 449                 return resq
 450             # must be a response for a different thread:
 451             else:
 452                 cv = self.cvars.get(seq, None)
 453                 # response involving unknown sequence number is discarded,
 454                 # probably intended for prior incarnation of server
 455                 if cv is not None:
 456                     cv.acquire()
 457                     self.responses[seq] = resq
 458                     cv.notify()
 459                     cv.release()
 460                 continue
 461
 462     def handle_EOF(self):
 463         "action taken upon link being closed by peer"
 464         self.EOFhook()
 465         self.debug("handle_EOF")
 466         for key in self.cvars:
 467             cv = self.cvars[key]
 468             cv.acquire()
 469             self.responses[key] = ('EOF', None)
 470             cv.notify()
 471             cv.release()
 472         # call our (possibly overridden) exit function
 473         self.exithook()
 474
 475     def EOFhook(self):
 476         "Classes using rpc client/server can override to augment EOF action"
 477         pass
 478
 479 #----------------- end class SocketIO --------------------
 480
 481 class RemoteObject(object):
 482     # Token mix-in class
 483     pass
 484
 485 def remoteref(obj):
 486     oid = id(obj)
 487     objecttable[oid] = obj
 488     return RemoteProxy(oid)
 489
 490 class RemoteProxy(object):
 491
 492     def __init__(self, oid):
 493         self.oid = oid
 494
 495 class RPCHandler(SocketServer.BaseRequestHandler, SocketIO):
 496
 497     debugging = False
 498     location = "#S"  # Server
 499
 500     def __init__(self, sock, addr, svr):
 501         svr.current_handler = self ## cgt xxx
 502         SocketIO.__init__(self, sock)
 503         SocketServer.BaseRequestHandler.__init__(self, sock, addr, svr)
 504
 505     def handle(self):
 506         "handle() method required by SocketServer"
 507         self.mainloop()
 508
 509     def get_remote_proxy(self, oid):
 510         return RPCProxy(self, oid)
 511
 512 class RPCClient(SocketIO):
 513
 514     debugging = False
 515     location = "#C"  # Client
 516
 517     nextseq = 1 # Requests coming from the client are odd numbered
 518
 519     def __init__(self, address, family=socket.AF_INET, type=socket.SOCK_STREAM):
 520         self.listening_sock = socket.socket(family, type)
 521         self.listening_sock.setsockopt(socket.SOL_SOCKET,
 522                                        socket.SO_REUSEADDR, 1)
 523         self.listening_sock.bind(address)
 524         self.listening_sock.listen(1)
 525
 526     def accept(self):
 527         working_sock, address = self.listening_sock.accept()
 528         if self.debugging:
 529             print>>sys.__stderr__, "****** Connection request from ", address
 530         if address[0] == LOCALHOST:
 531             SocketIO.__init__(self, working_sock)
 532         else:
 533             print>>sys.__stderr__, "** Invalid host: ", address
 534             raise socket.error
 535
 536     def get_remote_proxy(self, oid):
 537         return RPCProxy(self, oid)
 538
 539 class RPCProxy(object):
 540
 541     __methods = None
 542     __attributes = None
 543
 544     def __init__(self, sockio, oid):
 545         self.sockio = sockio
 546         self.oid = oid
 547
 548     def __getattr__(self, name):
 549         if self.__methods is None:
 550             self.__getmethods()
 551         if self.__methods.get(name):
 552             return MethodProxy(self.sockio, self.oid, name)
 553         if self.__attributes is None:
 554             self.__getattributes()
 555         if self.__attributes.has_key(name):
 556             value = self.sockio.remotecall(self.oid, '__getattribute__',
 557                                            (name,), {})
 558             return value
 559         else:
 560             raise AttributeError, name
 561
 562     def __getattributes(self):
 563         self.__attributes = self.sockio.remotecall(self.oid,
 564                                                 "__attributes__", (), {})
 565
 566     def __getmethods(self):
 567         self.__methods = self.sockio.remotecall(self.oid,
 568                                                 "__methods__", (), {})
 569
 570 def _getmethods(obj, methods):
 571     # Helper to get a list of methods from an object
 572     # Adds names to dictionary argument 'methods'
 573     for name in dir(obj):
 574         attr = getattr(obj, name)
 575         if callable(attr):
 576             methods[name] = 1
 577     if type(obj) == types.InstanceType:
 578         _getmethods(obj.__class__, methods)
 579     if type(obj) == types.ClassType:
 580         for super in obj.__bases__:
 581             _getmethods(super, methods)
 582
 583 def _getattributes(obj, attributes):
 584     for name in dir(obj):
 585         attr = getattr(obj, name)
 586         if not callable(attr):
 587             attributes[name] = 1
 588
 589 class MethodProxy(object):
 590
 591     def __init__(self, sockio, oid, name):
 592         self.sockio = sockio
 593         self.oid = oid
 594         self.name = name
 595
 596     def __call__(self, *args, **kwargs):
 597         value = self.sockio.remotecall(self.oid, self.name, args, kwargs)
 598         return value
 599
 600
 601 # XXX KBK 09Sep03  We need a proper unit test for this module.  Previously
 602 #                  existing test code was removed at Rev 1.27.