python/qemu: Add ConsoleSocket for optional use in QEMUMachine
[qemu.git] / 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