[PATCH v1 0/3] python/qemu: follow-up changes for ConsoleSocket

[PATCH v1 0/3] python/qemu: follow-up changes for ConsoleSocket

[Robert Foley]

For v1, we added a few minor changes, and also added one new patch 
in tests/vm to add a shutdown timeout.  This fixes an issue we saw in 
testing the aarch64 VMs with TCG.

This patch series introduces a few follow-up changes after the introduction of 
ConsoleSocket.

The first patch introduces cleanup changes for pylint and flake8.

The second patch allows machine.py to use a single type for the console_socket,
a ConsoleSocket.
Since machine.py will use ConsoleSocket for both the draining and non-draining
cases, we changed ConsoleSocket to handle the case where it does not drain the
socket at all and essentially behaves like a socket.

Robert Foley (3):
  python/qemu: Cleanup changes to ConsoleSocket
  python/qemu: Change ConsoleSocket to optionally drain socket.
  tests/vm: add shutdown timeout in basevm.py

 python/qemu/console_socket.py | 137 +++++++++++++++++++---------------
 python/qemu/machine.py        |  14 ++--
 python/qemu/pylintrc          |   2 +-
 tests/vm/basevm.py            |  15 ++--
 4 files changed, 94 insertions(+), 74 deletions(-)

-- 
2.17.1

[PATCH v1 1/3] python/qemu: Cleanup changes to ConsoleSocket

[Robert Foley]

The changes to console_socket.py and machine.py are to
cleanup for pylint and flake8.

Reviewed-by: Alex Bennée <alex.bennee@linaro.org>
Signed-off-by: Robert Foley <robert.foley@linaro.org>
---
 python/qemu/console_socket.py | 57 ++++++++++++++++++-----------------
 python/qemu/machine.py        |  7 +++--
 python/qemu/pylintrc          |  2 +-
 3 files changed, 34 insertions(+), 32 deletions(-)

diff --git a/python/qemu/console_socket.py b/python/qemu/console_socket.py
index 830cb7c628..09986bc215 100644
--- a/python/qemu/console_socket.py
+++ b/python/qemu/console_socket.py
@@ -1,12 +1,9 @@
-#!/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.
-#
+"""
+QEMU Console Socket Module:
+
+This python module implements a ConsoleSocket object,
+which can drain a socket and optionally dump the bytes to file.
+"""
 # Copyright 2020 Linaro
 #
 # Authors:
@@ -15,20 +12,27 @@
 # 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):
+    """
+    ConsoleSocket represents a socket attached to a char device.
 
+    Drains the socket and places the bytes into an 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 debugging purposes.
+    """
     def __init__(self, address, file=None):
         self._recv_timeout_sec = 300
+        self._sleep_time = 0.5
         self._buffer = deque()
         self._asyncore_thread = None
         self._sock = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
@@ -70,31 +74,28 @@ class ConsoleSocket(asyncore.dispatcher):
 
     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
+        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")
         if self._logfile:
             self._logfile.write("{}".format(string))
             self._logfile.flush()
         for c in string:
             self._buffer.extend(c)
 
-    def recv(self, n=1, sleep_delay_s=0.1):
-        """Return chars from in memory buffer"""
+    def recv(self, buffer_size=1):
+        """Return chars from in memory buffer.
+           Maintains the same API as socket.socket.recv.
+        """
         start_time = time.time()
-        while len(self._buffer) < n:
-            time.sleep(sleep_delay_s)
+        while len(self._buffer) < buffer_size:
+            time.sleep(self._sleep_time)
             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)])
+        chars = ''.join([self._buffer.popleft() for i in range(buffer_size)])
         # 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
diff --git a/python/qemu/machine.py b/python/qemu/machine.py
index 80c4d4a8b6..9956360a79 100644
--- a/python/qemu/machine.py
+++ b/python/qemu/machine.py
@@ -27,7 +27,7 @@ import socket
 import tempfile
 from typing import Optional, Type
 from types import TracebackType
-from qemu.console_socket import ConsoleSocket
+from . import console_socket
 
 from . import qmp
 
@@ -674,8 +674,9 @@ class QEMUMachine:
         """
         if self._console_socket is None:
             if self._drain_console:
-                self._console_socket = ConsoleSocket(self._console_address,
-                                                    file=self._console_log_path)
+                self._console_socket = console_socket.ConsoleSocket(
+                    self._console_address,
+                    file=self._console_log_path)
             else:
                 self._console_socket = socket.socket(socket.AF_UNIX,
                                                      socket.SOCK_STREAM)
diff --git a/python/qemu/pylintrc b/python/qemu/pylintrc
index 5d6ae7367d..3f69205000 100644
--- a/python/qemu/pylintrc
+++ b/python/qemu/pylintrc
@@ -33,7 +33,7 @@ good-names=i,
            Run,
            _,
            fd,
-
+           c,
 [VARIABLES]
 
 [STRING]
-- 
2.17.1

[PATCH v1 2/3] python/qemu: Change ConsoleSocket to optionally drain socket.

[Robert Foley]

The primary purpose of this change is to clean up
machine.py's console_socket property to return a single type,
a ConsoleSocket.

ConsoleSocket now derives from a socket, which means that
in the default case (of not draining), machine.py
will see the same behavior as it did prior to ConsoleSocket.

Signed-off-by: Robert Foley <robert.foley@linaro.org>
---
 python/qemu/console_socket.py | 92 +++++++++++++++++++++--------------
 python/qemu/machine.py        | 13 ++---
 2 files changed, 59 insertions(+), 46 deletions(-)

diff --git a/python/qemu/console_socket.py b/python/qemu/console_socket.py
index 09986bc215..70869fbbdc 100644
--- a/python/qemu/console_socket.py
+++ b/python/qemu/console_socket.py
@@ -13,68 +13,75 @@ which can drain a socket and optionally dump the bytes to file.
 # the COPYING file in the top-level directory.
 #
 
-import asyncore
 import socket
 import threading
 from collections import deque
 import time
 
 
-class ConsoleSocket(asyncore.dispatcher):
+class ConsoleSocket(socket.socket):
     """
     ConsoleSocket represents a socket attached to a char device.
 
-    Drains the socket and places the bytes into an in memory buffer
-    for later processing.
+    Optionally (if drain==True), drains the socket and places the bytes
+    into an 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 debugging purposes.
     """
-    def __init__(self, address, file=None):
+    def __init__(self, address, file=None, drain=False):
         self._recv_timeout_sec = 300
         self._sleep_time = 0.5
         self._buffer = deque()
-        self._asyncore_thread = None
-        self._sock = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
-        self._sock.connect(address)
+        socket.socket.__init__(self, socket.AF_UNIX, socket.SOCK_STREAM)
+        self.connect(address)
         self._logfile = None
         if file:
             self._logfile = open(file, "w")
-        asyncore.dispatcher.__init__(self, sock=self._sock)
         self._open = True
-        self._thread_start()
+        if drain:
+            self._drain_thread = self._thread_start()
+        else:
+            self._drain_thread = None
 
-    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 _drain_fn(self):
+        """Drains the socket and runs while the socket is open."""
+        while self._open:
+            try:
+                self._drain_socket()
+            except socket.timeout:
+                # The socket is expected to timeout since we set a
+                # short timeout to allow the thread to exit when
+                # self._open is set to False.
+                time.sleep(self._sleep_time)
 
-    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 _thread_start(self):
+        """Kick off a thread to drain the socket."""
+        # Configure socket to not block and timeout.
+        # This allows our drain thread to not block
+        # on recieve and exit smoothly.
+        socket.socket.setblocking(self, False)
+        socket.socket.settimeout(self, 1)
+        drain_thread = threading.Thread(target=self._drain_fn)
+        drain_thread.daemon = True
+        drain_thread.start()
+        return drain_thread
 
     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
+            if self._drain_thread is not None:
+                thread, self._drain_thread = self._drain_thread, None
                 thread.join()
+            socket.socket.close(self)
             if self._logfile:
                 self._logfile.close()
                 self._logfile = None
 
-    def handle_read(self):
+    def _drain_socket(self):
         """process arriving characters into in memory _buffer"""
-        data = asyncore.dispatcher.recv(self, 1)
+        data = socket.socket.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.
@@ -85,27 +92,38 @@ class ConsoleSocket(asyncore.dispatcher):
         for c in string:
             self._buffer.extend(c)
 
-    def recv(self, buffer_size=1):
+    def recv(self, bufsize=1):
         """Return chars from in memory buffer.
            Maintains the same API as socket.socket.recv.
         """
+        if self._drain_thread is None:
+            # Not buffering the socket, pass thru to socket.
+            return socket.socket.recv(self, bufsize)
         start_time = time.time()
-        while len(self._buffer) < buffer_size:
+        while len(self._buffer) < bufsize:
             time.sleep(self._sleep_time)
             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(buffer_size)])
+        chars = ''.join([self._buffer.popleft() for i in range(bufsize)])
         # 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 setblocking(self, value):
+        """When not draining we pass thru to the socket,
+           since when draining we control socket blocking.
+        """
+        if self._drain_thread is None:
+            socket.socket.setblocking(self, value)
 
     def settimeout(self, seconds):
-        """Set current timeout on recv"""
-        self._recv_timeout_sec = seconds
+        """When not draining we pass thru to the socket,
+           since when draining we control the timeout.
+        """
+        if seconds is not None:
+            self._recv_timeout_sec = seconds
+        if self._drain_thread is None:
+            socket.socket.settimeout(self, seconds)
diff --git a/python/qemu/machine.py b/python/qemu/machine.py
index 9956360a79..c5f777b9e7 100644
--- a/python/qemu/machine.py
+++ b/python/qemu/machine.py
@@ -23,7 +23,6 @@ import os
 import subprocess
 import shutil
 import signal
-import socket
 import tempfile
 from typing import Optional, Type
 from types import TracebackType
@@ -673,12 +672,8 @@ class QEMUMachine:
         Returns a socket connected to the console
         """
         if self._console_socket is None:
-            if self._drain_console:
-                self._console_socket = 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)
+            self._console_socket = console_socket.ConsoleSocket(
+                self._console_address,
+                file=self._console_log_path,
+                drain=self._drain_console)
         return self._console_socket
-- 
2.17.1

[PATCH v1 3/3] tests/vm: add shutdown timeout in basevm.py

[Robert Foley]

We are adding the shutdown timeout to solve an issue
we now see where the aarch64 VMs timeout on shutdown
under TCG.

There is a new 3 second timeout in machine.py,
which we override in basevm.py when shutting down.

Signed-off-by: Robert Foley <robert.foley@linaro.org>
---
 tests/vm/basevm.py | 15 ++++++++++-----
 1 file changed, 10 insertions(+), 5 deletions(-)

diff --git a/tests/vm/basevm.py b/tests/vm/basevm.py
index 7acb48b876..3fac20e929 100644
--- a/tests/vm/basevm.py
+++ b/tests/vm/basevm.py
@@ -80,6 +80,8 @@ class BaseVM(object):
     arch = "#arch"
     # command to halt the guest, can be overridden by subclasses
     poweroff = "poweroff"
+    # Time to wait for shutdown to finish.
+    shutdown_timeout_default = 30
     # enable IPv6 networking
     ipv6 = True
     # This is the timeout on the wait for console bytes.
@@ -87,7 +89,7 @@ class BaseVM(object):
     # Scale up some timeouts under TCG.
     # 4 is arbitrary, but greater than 2,
     # since we found we need to wait more than twice as long.
-    tcg_ssh_timeout_multiplier = 4
+    tcg_timeout_multiplier = 4
     def __init__(self, args, config=None):
         self._guest = None
         self._genisoimage = args.genisoimage
@@ -141,9 +143,12 @@ class BaseVM(object):
         if args.jobs and args.jobs > 1:
             self._args += ["-smp", "%d" % args.jobs]
         if kvm_available(self.arch):
+            self._shutdown_timeout = self.shutdown_timeout_default
             self._args += ["-enable-kvm"]
         else:
             logging.info("KVM not available, not using -enable-kvm")
+            self._shutdown_timeout = \
+                self.shutdown_timeout_default * self.tcg_timeout_multiplier
         self._data_args = []
 
         if self._config['qemu_args'] != None:
@@ -423,7 +428,7 @@ class BaseVM(object):
     def wait_ssh(self, wait_root=False, seconds=300, cmd="exit 0"):
         # Allow more time for VM to boot under TCG.
         if not kvm_available(self.arch):
-            seconds *= self.tcg_ssh_timeout_multiplier
+            seconds *= self.tcg_timeout_multiplier
         starttime = datetime.datetime.now()
         endtime = starttime + datetime.timedelta(seconds=seconds)
         cmd_success = False
@@ -441,14 +446,14 @@ class BaseVM(object):
             raise Exception("Timeout while waiting for guest ssh")
 
     def shutdown(self):
-        self._guest.shutdown()
+        self._guest.shutdown(timeout=self._shutdown_timeout)
 
     def wait(self):
-        self._guest.wait()
+        self._guest.wait(timeout=self._shutdown_timeout)
 
     def graceful_shutdown(self):
         self.ssh_root(self.poweroff)
-        self._guest.wait()
+        self._guest.wait(timeout=self._shutdown_timeout)
 
     def qmp(self, *args, **kwargs):
         return self._guest.qmp(*args, **kwargs)
-- 
2.17.1