python/qemu/console_socket.py

   1 #!/usr/bin/env python3
   2 #
   3 # This python module implements a ConsoleSocket object which is
   4 # designed always drain the socket itself, and place
   5 # the bytes into a in memory buffer for later processing.
   6 #
   7 # Optionally a file path can be passed in and we will also
   8 # dump the characters to this file for debug.
   9 #
  10 # Copyright 2020 Linaro
  11 #
  12 # Authors:
  13 #  Robert Foley <robert.foley@linaro.org>
  14 #
  15 # This code is licensed under the GPL version 2 or later.  See
  16 # the COPYING file in the top-level directory.
  17 #
  18 import asyncore
  19 import socket
  20 import threading
  21 import io
  22 import os
  23 import sys
  24 from collections import deque
  25 import time
  26 import traceback
  27
  28 class ConsoleSocket(asyncore.dispatcher):
  29
  30     def __init__(self, address, file=None):
  31         self._recv_timeout_sec = 300
  32         self._buffer = deque()
  33         self._asyncore_thread = None
  34         self._sock = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
  35         self._sock.connect(address)
  36         self._logfile = None
  37         if file:
  38             self._logfile = open(file, "w")
  39         asyncore.dispatcher.__init__(self, sock=self._sock)
  40         self._open = True
  41         self._thread_start()
  42
  43     def _thread_start(self):
  44         """Kick off a thread to wait on the asyncore.loop"""
  45         if self._asyncore_thread is not None:
  46             return
  47         self._asyncore_thread = threading.Thread(target=asyncore.loop,
  48                                                  kwargs={'timeout':1})
  49         self._asyncore_thread.daemon = True
  50         self._asyncore_thread.start()
  51
  52     def handle_close(self):
  53         """redirect close to base class"""
  54         # Call the base class close, but not self.close() since
  55         # handle_close() occurs in the context of the thread which
  56         # self.close() attempts to join.
  57         asyncore.dispatcher.close(self)
  58
  59     def close(self):
  60         """Close the base object and wait for the thread to terminate"""
  61         if self._open:
  62             self._open = False
  63             asyncore.dispatcher.close(self)
  64             if self._asyncore_thread is not None:
  65                 thread, self._asyncore_thread = self._asyncore_thread, None
  66                 thread.join()
  67             if self._logfile:
  68                 self._logfile.close()
  69                 self._logfile = None
  70
  71     def handle_read(self):
  72         """process arriving characters into in memory _buffer"""
  73         try:
  74             data = asyncore.dispatcher.recv(self, 1)
  75             # latin1 is needed since there are some chars
  76             # we are receiving that cannot be encoded to utf-8
  77             # such as 0xe2, 0x80, 0xA6.
  78             string = data.decode("latin1")
  79         except:
  80             print("Exception seen.")
  81             traceback.print_exc()
  82             return
  83         if self._logfile:
  84             self._logfile.write("{}".format(string))
  85             self._logfile.flush()
  86         for c in string:
  87             self._buffer.extend(c)
  88
  89     def recv(self, n=1, sleep_delay_s=0.1):
  90         """Return chars from in memory buffer"""
  91         start_time = time.time()
  92         while len(self._buffer) < n:
  93             time.sleep(sleep_delay_s)
  94             elapsed_sec = time.time() - start_time
  95             if elapsed_sec > self._recv_timeout_sec:
  96                 raise socket.timeout
  97         chars = ''.join([self._buffer.popleft() for i in range(n)])
  98         # We choose to use latin1 to remain consistent with
  99         # handle_read() and give back the same data as the user would
 100         # receive if they were reading directly from the
 101         # socket w/o our intervention.
 102         return chars.encode("latin1")
 103
 104     def set_blocking(self):
 105         """Maintain compatibility with socket API"""
 106         pass
 107
 108     def settimeout(self, seconds):
 109         """Set current timeout on recv"""
 110         self._recv_timeout_sec = seconds