1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
|
"""
The LLVM Compiler Infrastructure
This file is distributed under the University of Illinois Open Source
License. See LICENSE.TXT for details.
Sync lldb and related source from a local machine to a remote machine.
This facilitates working on the lldb sourcecode on multiple machines
and multiple OS types, verifying changes across all.
This module provides asyncore channels used within the LLDB test
framework.
"""
from __future__ import print_function
from __future__ import absolute_import
# System modules
import asyncore
import socket
# Third-party modules
from six.moves import cPickle
# LLDB modules
class UnpicklingForwardingReaderChannel(asyncore.dispatcher):
"""Provides an unpickling, forwarding asyncore dispatch channel reader.
Inferior dotest.py processes with side-channel-based test results will
send test result event data in a pickled format, one event at a time.
This class supports reconstructing the pickled data and forwarding it
on to its final destination.
The channel data is written in the form:
{num_payload_bytes}#{payload_bytes}
The bulk of this class is devoted to reading and parsing out
the payload bytes.
"""
def __init__(self, file_object, async_map, forwarding_func):
asyncore.dispatcher.__init__(self, sock=file_object, map=async_map)
self.header_contents = b""
self.packet_bytes_remaining = 0
self.reading_header = True
self.ibuffer = b''
self.forwarding_func = forwarding_func
if forwarding_func is None:
# This whole class is useless if we do nothing with the
# unpickled results.
raise Exception("forwarding function must be set")
# Initiate all connections by sending an ack. This allows
# the initiators of the socket to await this to ensure
# that this end is up and running (and therefore already
# into the async map).
ack_bytes = b'*'
file_object.send(ack_bytes)
def deserialize_payload(self):
"""Unpickles the collected input buffer bytes and forwards."""
if len(self.ibuffer) > 0:
self.forwarding_func(cPickle.loads(self.ibuffer))
self.ibuffer = b''
def consume_header_bytes(self, data):
"""Consumes header bytes from the front of data.
@param data the incoming data stream bytes
@return any data leftover after consuming header bytes.
"""
# We're done if there is no content.
if not data or (len(data) == 0):
return None
full_header_len = 4
assert len(self.header_contents) < full_header_len
bytes_avail = len(data)
bytes_needed = full_header_len - len(self.header_contents)
header_bytes_avail = min(bytes_needed, bytes_avail)
self.header_contents += data[:header_bytes_avail]
if len(self.header_contents) == full_header_len:
import struct
# End of header.
self.packet_bytes_remaining = struct.unpack(
"!I", self.header_contents)[0]
self.header_contents = b""
self.reading_header = False
return data[header_bytes_avail:]
# If we made it here, we've exhausted the data and
# we're still parsing header content.
return None
def consume_payload_bytes(self, data):
"""Consumes payload bytes from the front of data.
@param data the incoming data stream bytes
@return any data leftover after consuming remaining payload bytes.
"""
if not data or (len(data) == 0):
# We're done and there's nothing to do.
return None
data_len = len(data)
if data_len <= self.packet_bytes_remaining:
# We're consuming all the data provided.
self.ibuffer += data
self.packet_bytes_remaining -= data_len
# If we're no longer waiting for payload bytes,
# we flip back to parsing header bytes and we
# unpickle the payload contents.
if self.packet_bytes_remaining < 1:
self.reading_header = True
self.deserialize_payload()
# We're done, no more data left.
return None
else:
# We're only consuming a portion of the data since
# the data contains more than the payload amount.
self.ibuffer += data[:self.packet_bytes_remaining]
data = data[self.packet_bytes_remaining:]
# We now move on to reading the header.
self.reading_header = True
self.packet_bytes_remaining = 0
# And we can deserialize the payload.
self.deserialize_payload()
# Return the remaining data.
return data
def handle_read(self):
# Read some data from the socket.
try:
data = self.recv(8192)
# print('driver socket READ: %d bytes' % len(data))
except socket.error as socket_error:
print(
"\nINFO: received socket error when reading data "
"from test inferior:\n{}".format(socket_error))
raise
except Exception as general_exception:
print(
"\nERROR: received non-socket error when reading data "
"from the test inferior:\n{}".format(general_exception))
raise
# Consume the message content.
while data and (len(data) > 0):
# If we're reading the header, gather header bytes.
if self.reading_header:
data = self.consume_header_bytes(data)
else:
data = self.consume_payload_bytes(data)
def handle_close(self):
# print("socket reader: closing port")
self.close()
class UnpicklingForwardingListenerChannel(asyncore.dispatcher):
"""Provides a socket listener asyncore channel for unpickling/forwarding.
This channel will listen on a socket port (use 0 for host-selected). Any
client that connects will have an UnpicklingForwardingReaderChannel handle
communication over the connection.
The dotest parallel test runners, when collecting test results, open the
test results side channel over a socket. This channel handles connections
from inferiors back to the test runner. Each worker fires up a listener
for each inferior invocation. This simplifies the asyncore.loop() usage,
one of the reasons for implementing with asyncore. This listener shuts
down once a single connection is made to it.
"""
def __init__(self, async_map, host, port, backlog_count, forwarding_func):
asyncore.dispatcher.__init__(self, map=async_map)
self.create_socket(socket.AF_INET, socket.SOCK_STREAM)
self.set_reuse_addr()
self.bind((host, port))
self.address = self.socket.getsockname()
self.listen(backlog_count)
self.handler = None
self.async_map = async_map
self.forwarding_func = forwarding_func
if forwarding_func is None:
# This whole class is useless if we do nothing with the
# unpickled results.
raise Exception("forwarding function must be set")
def handle_accept(self):
(sock, addr) = self.socket.accept()
if sock and addr:
# print('Incoming connection from %s' % repr(addr))
self.handler = UnpicklingForwardingReaderChannel(
sock, self.async_map, self.forwarding_func)
def handle_close(self):
self.close()
|
