|
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 | | -# |
| 1 | +""" |
| 2 | +QEMU Console Socket Module: |
| 3 | +
|
| 4 | +This python module implements a ConsoleSocket object, |
| 5 | +which can drain a socket and optionally dump the bytes to file. |
| 6 | +""" |
10 | 7 | # Copyright 2020 Linaro |
11 | 8 | # |
12 | 9 | # Authors: |
|
15 | 12 | # This code is licensed under the GPL version 2 or later. See |
16 | 13 | # the COPYING file in the top-level directory. |
17 | 14 | # |
18 | | -import asyncore |
| 15 | + |
19 | 16 | import socket |
20 | 17 | import threading |
21 | | -import io |
22 | | -import os |
23 | | -import sys |
24 | 18 | from collections import deque |
25 | 19 | import time |
26 | | -import traceback |
27 | 20 |
|
28 | | -class ConsoleSocket(asyncore.dispatcher): |
29 | 21 |
|
30 | | - def __init__(self, address, file=None): |
| 22 | +class ConsoleSocket(socket.socket): |
| 23 | + """ |
| 24 | + ConsoleSocket represents a socket attached to a char device. |
| 25 | +
|
| 26 | + Optionally (if drain==True), drains the socket and places the bytes |
| 27 | + into an in memory buffer for later processing. |
| 28 | +
|
| 29 | + Optionally a file path can be passed in and we will also |
| 30 | + dump the characters to this file for debugging purposes. |
| 31 | + """ |
| 32 | + def __init__(self, address, file=None, drain=False): |
31 | 33 | self._recv_timeout_sec = 300 |
| 34 | + self._sleep_time = 0.5 |
32 | 35 | 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 | + socket.socket.__init__(self, socket.AF_UNIX, socket.SOCK_STREAM) |
| 37 | + self.connect(address) |
36 | 38 | self._logfile = None |
37 | 39 | if file: |
38 | 40 | self._logfile = open(file, "w") |
39 | | - asyncore.dispatcher.__init__(self, sock=self._sock) |
40 | 41 | self._open = True |
41 | | - self._thread_start() |
| 42 | + if drain: |
| 43 | + self._drain_thread = self._thread_start() |
| 44 | + else: |
| 45 | + self._drain_thread = None |
42 | 46 |
|
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() |
| 47 | + def _drain_fn(self): |
| 48 | + """Drains the socket and runs while the socket is open.""" |
| 49 | + while self._open: |
| 50 | + try: |
| 51 | + self._drain_socket() |
| 52 | + except socket.timeout: |
| 53 | + # The socket is expected to timeout since we set a |
| 54 | + # short timeout to allow the thread to exit when |
| 55 | + # self._open is set to False. |
| 56 | + time.sleep(self._sleep_time) |
51 | 57 |
|
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 | + def _thread_start(self): |
| 59 | + """Kick off a thread to drain the socket.""" |
| 60 | + # Configure socket to not block and timeout. |
| 61 | + # This allows our drain thread to not block |
| 62 | + # on recieve and exit smoothly. |
| 63 | + socket.socket.setblocking(self, False) |
| 64 | + socket.socket.settimeout(self, 1) |
| 65 | + drain_thread = threading.Thread(target=self._drain_fn) |
| 66 | + drain_thread.daemon = True |
| 67 | + drain_thread.start() |
| 68 | + return drain_thread |
58 | 69 |
|
59 | 70 | def close(self): |
60 | 71 | """Close the base object and wait for the thread to terminate""" |
61 | 72 | if self._open: |
62 | 73 | 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 |
| 74 | + if self._drain_thread is not None: |
| 75 | + thread, self._drain_thread = self._drain_thread, None |
66 | 76 | thread.join() |
| 77 | + socket.socket.close(self) |
67 | 78 | if self._logfile: |
68 | 79 | self._logfile.close() |
69 | 80 | self._logfile = None |
70 | 81 |
|
71 | | - def handle_read(self): |
| 82 | + def _drain_socket(self): |
72 | 83 | """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 |
| 84 | + data = socket.socket.recv(self, 1) |
| 85 | + # latin1 is needed since there are some chars |
| 86 | + # we are receiving that cannot be encoded to utf-8 |
| 87 | + # such as 0xe2, 0x80, 0xA6. |
| 88 | + string = data.decode("latin1") |
83 | 89 | if self._logfile: |
84 | 90 | self._logfile.write("{}".format(string)) |
85 | 91 | self._logfile.flush() |
86 | 92 | for c in string: |
87 | 93 | self._buffer.extend(c) |
88 | 94 |
|
89 | | - def recv(self, n=1, sleep_delay_s=0.1): |
90 | | - """Return chars from in memory buffer""" |
| 95 | + def recv(self, bufsize=1): |
| 96 | + """Return chars from in memory buffer. |
| 97 | + Maintains the same API as socket.socket.recv. |
| 98 | + """ |
| 99 | + if self._drain_thread is None: |
| 100 | + # Not buffering the socket, pass thru to socket. |
| 101 | + return socket.socket.recv(self, bufsize) |
91 | 102 | start_time = time.time() |
92 | | - while len(self._buffer) < n: |
93 | | - time.sleep(sleep_delay_s) |
| 103 | + while len(self._buffer) < bufsize: |
| 104 | + time.sleep(self._sleep_time) |
94 | 105 | elapsed_sec = time.time() - start_time |
95 | 106 | if elapsed_sec > self._recv_timeout_sec: |
96 | 107 | raise socket.timeout |
97 | | - chars = ''.join([self._buffer.popleft() for i in range(n)]) |
| 108 | + chars = ''.join([self._buffer.popleft() for i in range(bufsize)]) |
98 | 109 | # We choose to use latin1 to remain consistent with |
99 | 110 | # handle_read() and give back the same data as the user would |
100 | 111 | # receive if they were reading directly from the |
101 | 112 | # socket w/o our intervention. |
102 | 113 | return chars.encode("latin1") |
103 | 114 |
|
104 | | - def set_blocking(self): |
105 | | - """Maintain compatibility with socket API""" |
106 | | - pass |
| 115 | + def setblocking(self, value): |
| 116 | + """When not draining we pass thru to the socket, |
| 117 | + since when draining we control socket blocking. |
| 118 | + """ |
| 119 | + if self._drain_thread is None: |
| 120 | + socket.socket.setblocking(self, value) |
107 | 121 |
|
108 | 122 | def settimeout(self, seconds): |
109 | | - """Set current timeout on recv""" |
110 | | - self._recv_timeout_sec = seconds |
| 123 | + """When not draining we pass thru to the socket, |
| 124 | + since when draining we control the timeout. |
| 125 | + """ |
| 126 | + if seconds is not None: |
| 127 | + self._recv_timeout_sec = seconds |
| 128 | + if self._drain_thread is None: |
| 129 | + socket.socket.settimeout(self, seconds) |
0 commit comments