Re: [PATCH v8 11/12] python/qemu: Add ConsoleSocket for optional use in QEMUMachine

[Robert Foley]

Hi Philippe,

On Sun, 31 May 2020 at 07:22, Philippe Mathieu-Daudé <philmd@redhat.com> wrote:
>
> On 5/29/20 10:34 PM, Robert Foley wrote:
> > We add the ConsoleSocket object, which has a socket interface
> > and which will consume all arriving characters on the
> > socket, placing them into an in memory buffer.
> > This will also provide those chars via recv() as
> > would a regular socket.
> > ConsoleSocket also has the option of dumping
> > the console bytes to a log file.
> >
> > We also give QEMUMachine the option of using ConsoleSocket
> > to drain and to use for logging console to a file.
> > By default QEMUMachine does not use ConsoleSocket.
> >
> > This is added in preparation for use by basevm.py in a later commit.
> > This is a workaround we found was needed for basevm.py since
> > there is a known issue where QEMU will hang waiting
> > for console characters to be consumed.
> >
> > Cc: Eduardo Habkost <ehabkost@redhat.com>
> > Cc: Cleber Rosa <crosa@redhat.com>
> > Signed-off-by: Robert Foley <robert.foley@linaro.org>
> > Reviewed-by: Peter Puhov <peter.puhov@linaro.org>
> > Acked-by: Alex Bennée <alex.bennee@linaro.org>
> > ---
> >  python/qemu/console_socket.py | 118 ++++++++++++++++++++++++++++++++++
> >  python/qemu/machine.py        |  23 +++++--
> >  2 files changed, 137 insertions(+), 4 deletions(-)
> >  create mode 100644 python/qemu/console_socket.py
> >
> > diff --git a/python/qemu/console_socket.py b/python/qemu/console_socket.py
> > new file mode 100644
> > index 0000000000..2fe1c20d64
> > --- /dev/null
> > +++ b/python/qemu/console_socket.py
> > @@ -0,0 +1,118 @@
> > +#!/usr/bin/env python3
> > +#
> > +# This python module implements a ConsoleSocket object which is
> > +# designed always drain the socket itself, and place
> > +# the bytes into a in memory buffer for later processing.
> > +#
> > +# Optionally a file path can be passed in and we will also
> > +# dump the characters to this file for debug.
> > +#
> > +# Copyright 2020 Linaro
> > +#
> > +# Authors:
> > +#  Robert Foley <robert.foley@linaro.org>
> > +#
> > +# This code is licensed under the GPL version 2 or later.  See
> > +# the COPYING file in the top-level directory.
> > +#
> > +import asyncore
> > +import socket
> > +import threading
> > +import io
> > +import os
> > +import sys
> > +from collections import deque
> > +import time
> > +import traceback
> > +
> > +class ConsoleSocket(asyncore.dispatcher):
> > +
> > +    def __init__(self, address, file=None):
> > +        self._recv_timeout_sec = 300
> > +        self._buffer = deque()
> > +        self._asyncore_thread = None
> > +        self._sock = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
> > +        self._sock.connect(address)
> > +        self._logfile = None
> > +        if file:
> > +            self._logfile = open(file, "w")
> > +        asyncore.dispatcher.__init__(self, sock=self._sock)
> > +        self._thread_start()
> > +        self._open = True
>
> Please set _open before starting the thread (invert last 2 lines).
>
> > +
> > +    def _thread_start(self):
> > +        """Kick off a thread to wait on the asyncore.loop"""
> > +        if self._asyncore_thread is not None:
> > +            return
> > +        self._asyncore_thread = threading.Thread(target=asyncore.loop,
> > +                                                 kwargs={'timeout':1})
> > +        self._asyncore_thread.daemon = True
> > +        self._asyncore_thread.start()
> > +
> > +    def handle_close(self):
> > +        """redirect close to base class"""
> > +        # Call the base class close, but not self.close() since
> > +        # handle_close() occurs in the context of the thread which
> > +        # self.close() attempts to join.
> > +        asyncore.dispatcher.close(self)
> > +
> > +    def close(self):
> > +        """Close the base object and wait for the thread to terminate"""
> > +        if self._open:
> > +            self._open = False
> > +            asyncore.dispatcher.close(self)
> > +            if self._asyncore_thread is not None:
> > +                thread, self._asyncore_thread = self._asyncore_thread, None
> > +                thread.join()
> > +            if self._logfile:
> > +                self._logfile.close()
> > +                self._logfile = None
> > +
> > +    def handle_read(self):
> > +        """process arriving characters into in memory _buffer"""
> > +        try:
> > +            data = asyncore.dispatcher.recv(self, 1)
> > +            # latin1 is needed since there are some chars
> > +            # we are receiving that cannot be encoded to utf-8
> > +            # such as 0xe2, 0x80, 0xA6.
> > +            string = data.decode("latin1")
> > +        except:
> > +            print("Exception seen.")
> > +            traceback.print_exc()
> > +            return
> > +        if self._logfile:
> > +            self._logfile.write("{}".format(string))
> > +            self._logfile.flush()
> > +        for c in string:
> > +            self._buffer.append(c)
>
>            self._buffer.extend(string) ?
>
> > +
> > +    def recv(self, n=1):
>
> Please make a sleep_delay_s argument:
>
>        def recv(self, n=1, sleep_delay_s=0.1):
>
> > +        """Return chars from in memory buffer"""
> > +        start_time = time.time()
> > +        while len(self._buffer) < n:
> > +            time.sleep(0.1)
>
>                time.sleep(sleep_delay_s)
>
> > +            elapsed_sec = time.time() - start_time
> > +            if elapsed_sec > self._recv_timeout_sec:
> > +                raise socket.timeout
> > +        chars = ''.join([self._buffer.popleft() for i in range(n)])
> > +        # We choose to use latin1 to remain consistent with
> > +        # handle_read() and give back the same data as the user would
> > +        # receive if they were reading directly from the
> > +        # socket w/o our intervention.
> > +        return chars.encode("latin1")
> > +
> > +    def set_blocking(self):
> > +        """Maintain compatibility with socket API"""
> > +        pass
> > +
> > +    def settimeout(self, seconds):
> > +        """Set current timeout on recv"""
> > +        self._recv_timeout_sec = seconds
> > +
> > +class ByteBuffer(deque):
> > +    """Simple in memory buffer with read/write interface"""
> > +    def write(self, bytes):
> > +        for i in bytes:
> > +            self.append(i)
>
>            self.extend(bytes) ?
>
> However 'bytes' seems a reserved keyword.

These are all good points.  However, I noticed that this class is
no longer used in this patch, so I will plan to remove it.

I will apply these and the other suggestions above too in the next version.

Thanks,
-Rob


>
> > +    def read(self, n):
> > +        return ''.join([self.popleft() for i in range(n)])
> > diff --git a/python/qemu/machine.py b/python/qemu/machine.py
> > index b9a98e2c86..ee9c337f55 100644
> > --- a/python/qemu/machine.py
> > +++ b/python/qemu/machine.py
> > @@ -24,6 +24,7 @@ import subprocess
> >  import shutil
> >  import socket
> >  import tempfile
> > +from qemu.console_socket import ConsoleSocket
> >
> >  from . import qmp
> >
> > @@ -71,7 +72,8 @@ class QEMUMachine(object):
> >
> >      def __init__(self, binary, args=None, wrapper=None, name=None,
> >                   test_dir="/var/tmp", monitor_address=None,
> > -                 socket_scm_helper=None, sock_dir=None):
> > +                 socket_scm_helper=None, sock_dir=None,
> > +                 drain_console=False, console_log=None):
> >          '''
> >          Initialize a QEMUMachine
> >
> > @@ -82,6 +84,9 @@ class QEMUMachine(object):
> >          @param test_dir: where to create socket and log file
> >          @param monitor_address: address for QMP monitor
> >          @param socket_scm_helper: helper program, required for 
> > send_fd_scm()
> > +        @param sock_dir: where to create socket (overrides test_dir for 
> > sock)
> > +        @param console_log: (optional) path to console log file
> > +        @param drain_console: (optional) True to drain console socket to 
> > buffer
> >          @note: Qemu process is not started until launch() is used.
> >          '''
> >          if args is None:
> > @@ -118,6 +123,12 @@ class QEMUMachine(object):
> >          self._console_address = None
> >          self._console_socket = None
> >          self._remove_files = []
> > +        self._console_log_path = console_log
> > +        if self._console_log_path:
> > +            # In order to log the console, buffering needs to be enabled.
> > +            self._drain_console = True
> > +        else:
> > +            self._drain_console = drain_console
> >
> >          # just in case logging wasn't configured by the main script:
> >          logging.basicConfig()
> > @@ -568,7 +579,11 @@ class QEMUMachine(object):
> >          Returns a socket connected to the console
> >          """
> >          if self._console_socket is None:
> > -            self._console_socket = socket.socket(socket.AF_UNIX,
> > -                                                 socket.SOCK_STREAM)
> > -            self._console_socket.connect(self._console_address)
> > +            if self._drain_console:
> > +                self._console_socket = ConsoleSocket(self._console_address,
> > +                                                    
> > file=self._console_log_path)
> > +            else:
> > +                self._console_socket = socket.socket(socket.AF_UNIX,
> > +                                                     socket.SOCK_STREAM)
> > +                self._console_socket.connect(self._console_address)
> >          return self._console_socket
> >
>
> Tested-by: Philippe Mathieu-Daudé <philmd@redhat.com>
>