+++ /dev/null
-import sys\r
-from Boxing import Box, dump_exception, load_exception\r
-from ModuleNetProxy import RootImporter\r
-from Lib import raise_exception, AttrFrontend\r
-\r
-\r
-class Connection(object):\r
- """\r
- the rpyc connection layer (protocol and APIs). generally speaking, the only \r
- things you'll need to access directly from this object are:\r
- * modules - represents the remote python interprerer's modules namespace\r
- * execute - executes the given code on the other side of the connection\r
- * namespace - the namespace in which the code you `execute` resides\r
-\r
- the rest of the attributes should be of no intresent to you, except maybe for \r
- `remote_conn`, which represents the other side of the connection. it is unlikely,\r
- however, you'll need to use it (it is used interally).\r
- \r
- when you are done using a connection, and wish to release the resources it\r
- uses, you should call close(). you don't have to, but if you don't, the gc\r
- can't release the memory because of cyclic references.\r
- """\r
- \r
- def __init__(self, channel):\r
- self._closed = False\r
- self._local_namespace = {}\r
- self.channel = channel\r
- self.box = Box(self)\r
- self.async_replies = {}\r
- self.sync_replies = {}\r
- self.request_seq = 0\r
- self.module_cache = {}\r
- # user APIs:\r
- self.modules = RootImporter(self)\r
- self.remote_conn = self.sync_request("handle_getconn")\r
- self.namespace = AttrFrontend(self.remote_conn._local_namespace)\r
- \r
- def __repr__(self):\r
- if self._closed:\r
- return "<%s - closed>" % (self.__class__.__name__,)\r
- else:\r
- return "<%s(%r)>" % (self.__class__.__name__, self.channel)\r
-\r
- # \r
- # file api layer\r
- #\r
- def close(self):\r
- if self._closed:\r
- return\r
- self._closed = True\r
- self.box.close()\r
- self.channel.close()\r
- # untangle circular references\r
- del self._local_namespace\r
- del self.channel\r
- del self.box\r
- del self.async_replies\r
- del self.sync_replies\r
- del self.request_seq\r
- del self.module_cache\r
- del self.modules\r
- del self.remote_conn\r
- del self.namespace\r
- \r
- def fileno(self):\r
- return self.channel.fileno()\r
-\r
- #\r
- # protocol layer\r
- #\r
- def _recv(self):\r
- return self.box.unpack(self.channel.recv())\r
-\r
- def _send(self, *args):\r
- return self.channel.send(self.box.pack(args))\r
- \r
- def send_request(self, handlername, *args):\r
- try:\r
- self.channel.lock.acquire()\r
- # this must be atomic {\r
- self.request_seq += 1 \r
- self._send(handlername, self.request_seq, args)\r
- return self.request_seq\r
- # }\r
- finally:\r
- self.channel.lock.release()\r
-\r
- def send_exception(self, seq, exc_info):\r
- self._send("exception", seq, dump_exception(*exc_info))\r
-\r
- def send_result(self, seq, obj):\r
- self._send("result", seq, obj)\r
-\r
- def dispatch_result(self, seq, obj):\r
- if seq in self.async_replies:\r
- self.async_replies.pop(seq)("result", obj)\r
- else: \r
- self.sync_replies[seq] = obj\r
- \r
- def dispatch_exception(self, seq, obj):\r
- excobj = load_exception(obj)\r
- if seq in self.async_replies:\r
- self.async_replies.pop(seq)("exception", excobj)\r
- else:\r
- raise_exception(*excobj)\r
-\r
- def dispatch_request(self, handlername, seq, args):\r
- try:\r
- res = getattr(self, handlername)(*args)\r
- except SystemExit:\r
- raise\r
- except:\r
- self.send_exception(seq, sys.exc_info())\r
- else:\r
- self.send_result(seq, res)\r
-\r
- def sync_request(self, *args):\r
- """performs a synchronous (blocking) request"""\r
- seq = self.send_request(*args)\r
- while seq not in self.sync_replies:\r
- self.serve()\r
- return self.sync_replies.pop(seq)\r
- \r
- def async_request(self, callback, *args):\r
- """performs an asynchronous (non-blocking) request"""\r
- seq = self.send_request(*args)\r
- self.async_replies[seq] = callback\r
- \r
- #\r
- # servers api\r
- #\r
- def poll(self):\r
- """if available, serves a single request, otherwise returns (non-blocking serve)"""\r
- if self.channel.is_available():\r
- self.serve()\r
- return True\r
- else:\r
- return False\r
- \r
- def serve(self):\r
- """serves a single request or reply (may block)"""\r
- self.channel.wait()\r
- handler, seq, obj = self._recv()\r
- if handler == "result":\r
- self.dispatch_result(seq, obj)\r
- elif handler == "exception":\r
- self.dispatch_exception(seq, obj)\r
- else:\r
- self.dispatch_request(handler, seq, obj) \r
-\r
- #\r
- # root requests\r
- #\r
- def rimport(self, modulename):\r
- """imports a module by name (as a string)"""\r
- if modulename not in self.module_cache:\r
- module = self.sync_request("handle_import", modulename)\r
- self.module_cache[modulename] = module\r
- return self.module_cache[modulename] \r
-\r
- def execute(self, expr, mode = "exec"):\r
- """execute the given code at the remote side of the connection"""\r
- return self.sync_request("handle_execute", expr, mode)\r
-\r
- #\r
- # handlers layer\r
- #\r
- def handle_incref(self, oid):\r
- self.box.incref(oid)\r
- \r
- def handle_decref(self, oid):\r
- self.box.decref(oid)\r
- \r
- def handle_delattr(self, oid, name):\r
- delattr(self.box[oid], name)\r
-\r
- def handle_getattr(self, oid, name):\r
- return getattr(self.box[oid], name)\r
-\r
- def handle_setattr(self, oid, name, value):\r
- setattr(self.box[oid], name, value)\r
-\r
- def handle_delitem(self, oid, index):\r
- del self.box[oid][index]\r
-\r
- def handle_getitem(self, oid, index):\r
- return self.box[oid][index]\r
-\r
- def handle_setitem(self, oid, index, value):\r
- self.box[oid][index] = value\r
-\r
- def handle_call(self, oid, args, kwargs):\r
- return self.box[oid](*args, **kwargs)\r
-\r
- def handle_repr(self, oid):\r
- return repr(self.box[oid])\r
-\r
- def handle_str(self, oid):\r
- return str(self.box[oid])\r
-\r
- def handle_bool(self, oid):\r
- return bool(self.box[oid])\r
-\r
- def handle_import(self, modulename):\r
- return __import__(modulename, None, None, modulename.split(".")[-1])\r
-\r
- def handle_getconn(self):\r
- return self\r
-\r
- def handle_execute(self, expr, mode):\r
- codeobj = compile(expr, "<from %s>" % (self,), mode)\r
- return eval(codeobj, self._local_namespace)\r
-\r
-\r
-\r
git://git.onelab.eu / monitor.git / blobdiff