packages/Python/lldbsuite/test_event/dotest_channels.py

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