From 010e0e8d51ad16aab6aa87e31887ad9055a804bb Mon Sep 17 00:00:00 2001 From: Alexander Koshevoy Date: Wed, 16 May 2018 17:38:04 +0300 Subject: [PATCH] PY-18029 Python console Thrift communication implemented on Python side A fix-up required for Python console tests! --- .../_pydev_bundle/pydev_console_utils.py | 22 +- .../helpers/pydev/pydev_console/__init__.py | 0 .../pydev_console/thrift_communication.py | 6 + .../helpers/pydev/pydev_console/thrift_rpc.py | 35 +++ .../pydev/pydev_console/thrift_transport.py | 220 ++++++++++++++++++ python/helpers/pydev/pydev_run_in_console.py | 10 +- python/helpers/pydev/pydevconsole.py | 33 ++- .../tests_pydevd/test_pydev_ipython_011.py | 17 +- .../pydev/tests_pydevd/test_pydevconsole.py | 5 +- 9 files changed, 319 insertions(+), 29 deletions(-) create mode 100644 python/helpers/pydev/pydev_console/__init__.py create mode 100644 python/helpers/pydev/pydev_console/thrift_communication.py create mode 100644 python/helpers/pydev/pydev_console/thrift_rpc.py create mode 100644 python/helpers/pydev/pydev_console/thrift_transport.py diff --git a/python/helpers/pydev/_pydev_bundle/pydev_console_utils.py b/python/helpers/pydev/_pydev_bundle/pydev_console_utils.py index a90b760bb9d0..8b0cb7c0a5f4 100644 --- a/python/helpers/pydev/_pydev_bundle/pydev_console_utils.py +++ b/python/helpers/pydev/_pydev_bundle/pydev_console_utils.py @@ -3,7 +3,7 @@ import sys import traceback from _pydev_bundle._pydev_calltip_util import get_description -from _pydev_bundle.pydev_imports import xmlrpclib, _queue, Exec +from _pydev_bundle.pydev_imports import _queue, Exec from _pydev_imps._pydev_saved_modules import thread from _pydevd_bundle import pydevd_thrift from _pydevd_bundle import pydevd_vars @@ -112,20 +112,17 @@ class StdIn(BaseStdIn): Object to be added to stdin (to emulate it as non-blocking while the next line arrives) ''' - def __init__(self, interpreter, host, client_port, original_stdin=sys.stdin): + def __init__(self, interpreter, rpc_client, original_stdin=sys.stdin): BaseStdIn.__init__(self, original_stdin) self.interpreter = interpreter - self.client_port = client_port - self.host = host + self.rpc_client = rpc_client def readline(self, *args, **kwargs): # Ok, callback into the client to get the new input try: - # @alexander todo uncomment - temporary stub - # @alexander todo use backward direction of present connection # server = xmlrpclib.Server('http://%s:%s' % (self.host, self.client_port)) # requested_input = server.RequestInput() - requested_input = 'this is test input\n' + requested_input = self.rpc_client.RequestInput() if not requested_input: return '\n' # Yes, a readline must return something (otherwise we can get an EOFError on the input() call). return requested_input @@ -181,7 +178,7 @@ class CodeFragment: # BaseInterpreterInterface # ======================================================================================================================= class BaseInterpreterInterface: - def __init__(self, mainThread, connect_status_queue=None): + def __init__(self, mainThread, connect_status_queue=None, rpc_client=None): self.mainThread = mainThread self.interruptable = False self.exec_queue = _queue.Queue(0) @@ -191,6 +188,8 @@ class BaseInterpreterInterface: self.mpl_modules_for_patching = {} self.init_mpl_modules_for_patching() + self.rpc_client = rpc_client + def build_banner(self): return 'print({0})\n'.format(repr(self.get_greeting_msg())) @@ -238,7 +237,7 @@ class BaseInterpreterInterface: def create_std_in(self, debugger=None, original_std_in=None): if debugger is None: # @alexander todo fix `StdIn.readline()` - return StdIn(self, self.host, self.client_port, original_stdin=original_std_in) + return StdIn(self, self.rpc_client, original_stdin=original_std_in) else: return DebugConsoleStdIn(dbg=debugger, original_stdin=original_std_in) @@ -454,9 +453,8 @@ class BaseInterpreterInterface: self.interruptable = True def get_server(self): - if getattr(self, 'host', None) is not None: - # @alexander todo deprecate and use the other side of the connection - return xmlrpclib.Server('http://%s:%s' % (self.host, self.client_port)) + if getattr(self, 'rpc_client', None) is not None: + return self.rpc_client else: return None diff --git a/python/helpers/pydev/pydev_console/__init__.py b/python/helpers/pydev/pydev_console/__init__.py new file mode 100644 index 000000000000..e69de29bb2d1 diff --git a/python/helpers/pydev/pydev_console/thrift_communication.py b/python/helpers/pydev/pydev_console/thrift_communication.py new file mode 100644 index 000000000000..d9d90b5387b2 --- /dev/null +++ b/python/helpers/pydev/pydev_console/thrift_communication.py @@ -0,0 +1,6 @@ +import os + +import thriftpy + +# dynamically import console.thrift classes into `console_thrift` module +console_thrift = thriftpy.load(os.path.join(os.path.dirname(os.path.realpath(__file__)), "console.thrift"), module_name="console_thrift") diff --git a/python/helpers/pydev/pydev_console/thrift_rpc.py b/python/helpers/pydev/pydev_console/thrift_rpc.py new file mode 100644 index 000000000000..193955e3dea0 --- /dev/null +++ b/python/helpers/pydev/pydev_console/thrift_rpc.py @@ -0,0 +1,35 @@ +from pydev_console.thrift_transport import TBidirectionalClientTransport +from thriftpy.protocol import TBinaryProtocolFactory +from thriftpy.server import TThreadedServer +from thriftpy.thrift import TProcessor, TClient + + +def make_rpc_client(client_service, host, port, proto_factory=TBinaryProtocolFactory()): + """ + + :param client_service: + :param server_service: + :param server_handler: + :param host: connection host + :param port: connection port + :param proto_factory: protocol factory for client + :return: + """ + + # instantiate client + transport = TBidirectionalClientTransport(host, port) + protocol = proto_factory.get_protocol(transport) + transport.open() + + client = TClient(client_service, protocol) + + server_transport = transport.get_server_transport() + + return client, server_transport + + +def make_rpc_server(server_transport, server_service, server_handler, proto_factory=TBinaryProtocolFactory()): + # setup server + processor = TProcessor(server_service, server_handler) + + return TThreadedServer(processor, server_transport, iprot_factory=proto_factory) diff --git a/python/helpers/pydev/pydev_console/thrift_transport.py b/python/helpers/pydev/pydev_console/thrift_transport.py new file mode 100644 index 000000000000..6b17ecbe6d43 --- /dev/null +++ b/python/helpers/pydev/pydev_console/thrift_transport.py @@ -0,0 +1,220 @@ +import socket +import struct +import threading +from io import BytesIO + +from thriftpy.transport import TTransportBase, readall + +REQUEST = 0 +RESPONSE = 1 + + +def _readall_buffered(read_fn, try_fill_buffer_fn, sz): + buff = b'' + have = 0 + while have < sz: + chunk = read_fn(sz - have) + + have += len(chunk) + buff += chunk + + if len(chunk) == 0: + # `fill_buffer_fn` should raise an exception if it could not fill the buffer + try_fill_buffer_fn(sz - have) + + return buff + + +class MultiplexedSocketReader(object): + + def __init__(self, s): + self._socket = s + + self._request_buffer = BytesIO() + self._response_buffer = BytesIO() + + self._request_buffer_lock = threading.RLock() + self._response_buffer_lock = threading.RLock() + + self._read_socket_lock = threading.RLock() + + def read_request(self, sz): + """ + Invoked form server-side of the bidirectional transport. + """ + return _readall_buffered(self._read_request_buffer, self._try_fill_buffer, sz) + + def read_response(self, sz): + """ + Invoked form client-side of the bidirectional transport. + """ + return _readall_buffered(self._read_response_buffer, self._try_fill_buffer, sz) + + def _read_request_buffer(self, sz): + with self._request_buffer_lock: + self._request_buffer.read(sz) + + def _read_response_buffer(self, sz): + with self._response_buffer_lock: + self._response_buffer.read(sz) + + def _try_fill_buffer(self): + with self._read_socket_lock: + direction, frame = self._read_frame() + + if direction == REQUEST: + with self._request_buffer_lock: + self._request_buffer.write(frame) + elif direction == RESPONSE: + with self._response_buffer_lock: + self._response_buffer.write(frame) + + def _read_frame(self): + # todo use sz argument! + + buff = readall(self._socket.recv, 4) + sz, = struct.unpack('!i', buff) + if sz == 0: + # this is an empty message even without a direction byte + return None, None + else: + buff = readall(self._socket.recv, 1) + # todo is this actually the equivalent of `buff[0]`? + direction, = struct.unpack('!b', buff) + frame = readall(self._socket.recv, sz - 1) + return direction, frame + + +class SocketWriter(object): + + def __init__(self, sock): + self._socket = sock + + def write(self, buf): + self._socket.send(buf) + self._socket.flush() + + +class FramedWriter(object): + + def __init__(self): + self._buffer = BytesIO() + + def _get_writer(self): + raise NotImplementedError + + def _get_write_direction(self): + raise NotImplementedError + + def write(self, buf): + self._buffer.write(buf) + + def flush(self): + # reset wbuf before write/flush to preserve state on underlying failure + out = self._buffer.getvalue() + # prepend the message with the direction byte + out = struct.pack("b", self._get_write_direction()) + out + self._buffer = BytesIO() + + # N.B.: Doing this string concatenation is WAY cheaper than making + # two separate calls to the underlying socket object. Socket writes in + # Python turn out to be REALLY expensive, but it seems to do a pretty + # good job of managing string buffer operations without excessive + # copies + self._get_writer().write(struct.pack("!i", len(out)) + out) + + +class TBidirectionalClientTransport(TTransportBase, FramedWriter): + def __init__(self, host, port): + super(TBidirectionalClientTransport, self).__init__() + + self.host = host + self.port = port + + # the following properties will be initialized in `open()` + self._client_socket = None + self._reader = None + self._writer = None + self._server_transport = None + + def _get_writer(self): + raise self._writer + + def _get_write_direction(self): + return REQUEST + + def _read(self, sz): + """ + Reads a response from the multiplexed reader. + """ + return self._reader.read_response(sz) + + def get_server_transport(self): + if not self._server_transport: + raise Exception + + return self._server_transport + + def is_open(self): + # todo we may try to monitor reads and writes and put a flag if they fail + return self._client_socket + + def open(self): + self._client_socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM) + self._reader = MultiplexedSocketReader(self._client_socket) + self._writer = SocketWriter(self._client_socket) + + self._server_transport = TReversedServerTransport(self._reader.read_request, self._writer) + + def close(self): + # todo should we do something with buffer of the multiplexed reader + self._client_socket.shutdown(socket.SHUT_RDWR) + self._client_socket.close() + + +# todo use TFramedTransport to wrap transport into size-prefixed messages + +class TServerTransportBase(object): + """Base class for Thrift server transports.""" + + def listen(self): + pass + + def accept(self): + raise NotImplementedError + + def close(self): + pass + + +class TReversedServerTransport(TServerTransportBase): + + def __init__(self, read_fn, writer): + self._read_fn = read_fn + self._writer = writer + + def accept(self): + return TReversedServerAcceptedTransport(self._read_fn, self._writer) + + +class TReversedServerAcceptedTransport(FramedWriter): + + def __init__(self, read_fn, writer): + """ + :param read_fn: hi-level read function (reads solely requests from the input stream) + :param writer: low-level writer (it is expected to know only bytes) + """ + super(TReversedServerAcceptedTransport, self).__init__() + + self._read_fn = read_fn + self._writer = writer + + def _get_writer(self): + return self._writer + + def _get_write_direction(self): + # this side acts as a server and writes responses back to the client + return RESPONSE + + def read(self, sz): + return self._read_fn(sz) diff --git a/python/helpers/pydev/pydev_run_in_console.py b/python/helpers/pydev/pydev_run_in_console.py index e7d023609e16..85ffe45391cf 100644 --- a/python/helpers/pydev/pydev_run_in_console.py +++ b/python/helpers/pydev/pydev_run_in_console.py @@ -4,12 +4,12 @@ Entry point module to run a file in the interactive console. import os import sys import traceback -from pydevconsole import InterpreterInterface, process_exec_queue, start_console_server, init_mpl_in_console -from _pydev_imps._pydev_saved_modules import threading, _queue from _pydev_bundle import pydev_imports -from _pydevd_bundle.pydevd_utils import save_main_module from _pydev_bundle.pydev_console_utils import StdIn +from _pydev_imps._pydev_saved_modules import threading, _queue +from _pydevd_bundle.pydevd_utils import save_main_module +from pydevconsole import InterpreterInterface, process_exec_queue, start_console_server, init_mpl_in_console from pydevd_file_utils import get_fullname @@ -126,6 +126,8 @@ if __name__ == '__main__': sys.exit = skip_successful_exit connect_status_queue = _queue.Queue() + # ERROR `InterpreterInterface.__init__()` signature changed + # todo this method and file will be eliminated by Liza interpreter = InterpreterInterface(host, int(client_port), threading.currentThread(), connect_status_queue=connect_status_queue) server_thread = threading.Thread(target=start_console_server, @@ -134,6 +136,8 @@ if __name__ == '__main__': server_thread.setDaemon(True) server_thread.start() + # ERROR `StdIn.__init__()` signature changed + # todo this method and file will be eliminated by Liza sys.stdin = StdIn(interpreter, host, client_port, sys.stdin) init_mpl_in_console(interpreter) diff --git a/python/helpers/pydev/pydevconsole.py b/python/helpers/pydev/pydevconsole.py index 5b47c81dc7ee..c855fbd6f00a 100644 --- a/python/helpers/pydev/pydevconsole.py +++ b/python/helpers/pydev/pydevconsole.py @@ -2,6 +2,8 @@ Entry point module to start the interactive console. ''' from _pydev_imps._pydev_saved_modules import thread +from pydev_console.thrift_rpc import make_rpc_client, make_rpc_server + start_new_thread = thread.start_new_thread try: @@ -89,10 +91,8 @@ class InterpreterInterface(BaseInterpreterInterface): The methods in this class should be registered in the xml-rpc server. ''' - def __init__(self, host, client_port, mainThread, connect_status_queue=None): - BaseInterpreterInterface.__init__(self, mainThread, connect_status_queue) - self.client_port = client_port - self.host = host + def __init__(self, mainThread, connect_status_queue=None, rpc_client=None): + BaseInterpreterInterface.__init__(self, mainThread, connect_status_queue, rpc_client) self.namespace = {} self.interpreter = InteractiveConsole(self.namespace) self._input_error_printed = False @@ -334,9 +334,30 @@ def start_server(host, port, client_port, client_host = None): #note that this does not work in jython!!! (sys method can't be replaced). sys.exit = do_exit - interpreter = InterpreterInterface(client_host, client_port, threading.currentThread()) + from pydev_console.thrift_communication import console_thrift - start_new_thread(start_console_server,(host, port, interpreter)) + client_service = console_thrift.IDE + + client, server_transport = make_rpc_client(client_service, client_host, client_port) + + interpreter = InterpreterInterface(client_host, client_port, threading.currentThread(), client) + + # start_new_thread(start_console_server,(host, port, interpreter)) + # we do not need to start the server in a new thread because it does not need to accept a client connection, it already has it + + + # todo do we need the following: + # # Tell UMD the proper default namespace + # _set_globals_function(interpreter.get_namespace) + + server_service = console_thrift.PythonConsole + + # `InterpreterInterface` implements all methods required for the handler + server_handler = interpreter + + server = make_rpc_server(server_transport, server_service, server_handler) + # todo as `server.serve()` is excessive we may want to get rid of `server` as `TThreadedServer` + server.serve() process_exec_queue(interpreter) diff --git a/python/helpers/pydev/tests_pydevd/test_pydev_ipython_011.py b/python/helpers/pydev/tests_pydevd/test_pydev_ipython_011.py index dca346cc542e..b90007480bbd 100644 --- a/python/helpers/pydev/tests_pydevd/test_pydev_ipython_011.py +++ b/python/helpers/pydev/tests_pydevd/test_pydev_ipython_011.py @@ -1,14 +1,16 @@ -import sys -import unittest -import threading import os +import socket +import sys +import threading +import time +import unittest + +import pytest + +from _pydev_bundle.pydev_console_utils import StdIn from _pydev_bundle.pydev_imports import SimpleXMLRPCServer from _pydev_bundle.pydev_localhost import get_localhost -from _pydev_bundle.pydev_console_utils import StdIn -import socket -import time from _pydevd_bundle import pydevd_io -import pytest try: xrange @@ -249,6 +251,7 @@ class TestRunningCode(TestBase): client_thread = start_client_thread(self.client_port) orig_stdin = sys.stdin + # @alexander TODO `StdIn.__init__()` signature changed sys.stdin = StdIn(self, get_localhost(), self.client_port) try: filename = 'made_up_file.py' diff --git a/python/helpers/pydev/tests_pydevd/test_pydevconsole.py b/python/helpers/pydev/tests_pydevd/test_pydevconsole.py index 4b5069b2de97..41713ff65650 100644 --- a/python/helpers/pydev/tests_pydevd/test_pydevconsole.py +++ b/python/helpers/pydev/tests_pydevd/test_pydevconsole.py @@ -1,6 +1,7 @@ +import sys import threading import unittest -import sys + import pydevconsole from _pydev_bundle.pydev_imports import xmlrpclib, SimpleXMLRPCServer from _pydevd_bundle import pydevd_io @@ -32,6 +33,7 @@ class Test(unittest.TestCase): time.sleep(.3) #let's give it some time to start the threads from _pydev_bundle import pydev_localhost + # @alexander TODO `InterpreterInterface.__init__()` signature changed interpreter = pydevconsole.InterpreterInterface(pydev_localhost.get_localhost(), client_port, threading.currentThread()) (result,) = interpreter.hello("Hello pydevconsole") @@ -53,6 +55,7 @@ class Test(unittest.TestCase): from _pydev_bundle import pydev_localhost from _pydev_bundle.pydev_console_utils import CodeFragment + # @alexander TODO `InterpreterInterface.__init__()` signature changed interpreter = pydevconsole.InterpreterInterface(pydev_localhost.get_localhost(), client_port, threading.currentThread()) sys.stdout = pydevd_io.IOBuf() interpreter.add_exec(CodeFragment('class Foo:\n CONSTANT=1\n'))