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     def __init__(self, file_object, async_map, forwarding_func):
  46         asyncore.dispatcher.__init__(self, sock=file_object, map=async_map)
  47
  48         self.header_contents = b""
  49         self.packet_bytes_remaining = 0
  50         self.reading_header = True
  51         self.ibuffer = b''
  52         self.forwarding_func = forwarding_func
  53         if forwarding_func is None:
  54             # This whole class is useless if we do nothing with the
  55             # unpickled results.
  56             raise Exception("forwarding function must be set")
  57
  58         # Initiate all connections by sending an ack.  This allows
  59         # the initiators of the socket to await this to ensure
  60         # that this end is up and running (and therefore already
  61         # into the async map).
  62         ack_bytes = b'*'
  63         file_object.send(ack_bytes)
  64
  65     def deserialize_payload(self):
  66         """Unpickles the collected input buffer bytes and forwards."""
  67         if len(self.ibuffer) > 0:
  68             self.forwarding_func(cPickle.loads(self.ibuffer))
  69             self.ibuffer = b''
  70
  71     def consume_header_bytes(self, data):
  72         """Consumes header bytes from the front of data.
  73         @param data the incoming data stream bytes
  74         @return any data leftover after consuming header bytes.
  75         """
  76         # We're done if there is no content.
  77         if not data or (len(data) == 0):
  78             return None
  79
  80         full_header_len = 4
  81
  82         assert len(self.header_contents) < full_header_len
  83
  84         bytes_avail = len(data)
  85         bytes_needed = full_header_len - len(self.header_contents)
  86         header_bytes_avail = min(bytes_needed, bytes_avail)
  87         self.header_contents += data[:header_bytes_avail]
  88         if len(self.header_contents) == full_header_len:
  89             import struct
  90             # End of header.
  91             self.packet_bytes_remaining = struct.unpack(
  92                 "!I", self.header_contents)[0]
  93             self.header_contents = b""
  94             self.reading_header = False
  95             return data[header_bytes_avail:]
  96
  97         # If we made it here, we've exhausted the data and
  98         # we're still parsing header content.
  99         return None
 100
 101     def consume_payload_bytes(self, data):
 102         """Consumes payload bytes from the front of data.
 103         @param data the incoming data stream bytes
 104         @return any data leftover after consuming remaining payload bytes.
 105         """
 106         if not data or (len(data) == 0):
 107             # We're done and there's nothing to do.
 108             return None
 109
 110         data_len = len(data)
 111         if data_len <= self.packet_bytes_remaining:
 112             # We're consuming all the data provided.
 113             self.ibuffer += data
 114             self.packet_bytes_remaining -= data_len
 115
 116             # If we're no longer waiting for payload bytes,
 117             # we flip back to parsing header bytes and we
 118             # unpickle the payload contents.
 119             if self.packet_bytes_remaining < 1:
 120                 self.reading_header = True
 121                 self.deserialize_payload()
 122
 123             # We're done, no more data left.
 124             return None
 125         else:
 126             # We're only consuming a portion of the data since
 127             # the data contains more than the payload amount.
 128             self.ibuffer += data[:self.packet_bytes_remaining]
 129             data = data[self.packet_bytes_remaining:]
 130
 131             # We now move on to reading the header.
 132             self.reading_header = True
 133             self.packet_bytes_remaining = 0
 134
 135             # And we can deserialize the payload.
 136             self.deserialize_payload()
 137
 138             # Return the remaining data.
 139             return data
 140
 141     def handle_read(self):
 142         # Read some data from the socket.
 143         try:
 144             data = self.recv(8192)
 145             # print('driver socket READ: %d bytes' % len(data))
 146         except socket.error as socket_error:
 147             print(
 148                 "\nINFO: received socket error when reading data "
 149                 "from test inferior:\n{}".format(socket_error))
 150             raise
 151         except Exception as general_exception:
 152             print(
 153                 "\nERROR: received non-socket error when reading data "
 154                 "from the test inferior:\n{}".format(general_exception))
 155             raise
 156
 157         # Consume the message content.
 158         while data and (len(data) > 0):
 159             # If we're reading the header, gather header bytes.
 160             if self.reading_header:
 161                 data = self.consume_header_bytes(data)
 162             else:
 163                 data = self.consume_payload_bytes(data)
 164
 165     def handle_close(self):
 166         # print("socket reader: closing port")
 167         self.close()
 168
 169
 170 class UnpicklingForwardingListenerChannel(asyncore.dispatcher):
 171     """Provides a socket listener asyncore channel for unpickling/forwarding.
 172
 173     This channel will listen on a socket port (use 0 for host-selected).  Any
 174     client that connects will have an UnpicklingForwardingReaderChannel handle
 175     communication over the connection.
 176
 177     The dotest parallel test runners, when collecting test results, open the
 178     test results side channel over a socket.  This channel handles connections
 179     from inferiors back to the test runner.  Each worker fires up a listener
 180     for each inferior invocation.  This simplifies the asyncore.loop() usage,
 181     one of the reasons for implementing with asyncore.  This listener shuts
 182     down once a single connection is made to it.
 183     """
 184     def __init__(self, async_map, host, port, backlog_count, forwarding_func):
 185         asyncore.dispatcher.__init__(self, map=async_map)
 186         self.create_socket(socket.AF_INET, socket.SOCK_STREAM)
 187         self.set_reuse_addr()
 188         self.bind((host, port))
 189         self.address = self.socket.getsockname()
 190         self.listen(backlog_count)
 191         self.handler = None
 192         self.async_map = async_map
 193         self.forwarding_func = forwarding_func
 194         if forwarding_func is None:
 195             # This whole class is useless if we do nothing with the
 196             # unpickled results.
 197             raise Exception("forwarding function must be set")
 198
 199     def handle_accept(self):
 200         (sock, addr) = self.socket.accept()
 201         if sock and addr:
 202             # print('Incoming connection from %s' % repr(addr))
 203             self.handler = UnpicklingForwardingReaderChannel(
 204                 sock, self.async_map, self.forwarding_func)
 205
 206     def handle_close(self):
 207         self.close()