PY-18029 Python console Thrift communication implemented on Python side

A fix-up required for Python console tests!

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
 

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")

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)

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)

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)

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)
 

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'

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'))