src/third_party/v8/test/debugging/wasm/gdb-server/gdb_rsp.py

# Copyright 2020 the V8 project authors. All rights reserved.
# Use of this source code is governed by a BSD-style license that can be
# found in the LICENSE file.

import re
import socket
import struct
import subprocess
import time
import xml.etree.ElementTree
import sys

# Python2 has both int and long, Python3 only has int, which is the same as
# Python2 long.
try:
  long
except NameError:
  long = int

# Python2 has xrange and range. Python3 range is Python2 xrange.
try:
  xrange
except NameError:
  xrange = range

SOCKET_ADDR = ('localhost', 8765)

SIGTRAP = 5
SIGSEGV = 11
RETURNCODE_KILL = -9

ARCH = 'wasm32'
REG_DEFS = {
    ARCH: [('pc', 'Q'), ],
}


def EnsurePortIsAvailable(addr=SOCKET_ADDR):
  # As a validity check, check that the TCP port is available by binding to it
  # ourselves (and then unbinding).  Otherwise, we could end up talking to an
  # old instance of the GDB stub that is still hanging around, or to some
  # unrelated service that uses the same port number. Of course, there is still
  # a race condition because an unrelated process could bind the port after we
  # unbind.
  sock = socket.socket()
  sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, True)
  sock.bind(addr)
  sock.close()

def RspChecksum(data):
  checksum = 0
  for char in data:
    checksum = (checksum + ord(char)) & 0xff
  return checksum

class GdbRspConnection(object):

  def __init__(self, addr=SOCKET_ADDR):
    self._socket = self._Connect(addr)

  def _Connect(self, addr):
    # We have to poll because we do not know when the GDB stub has successfully
    # done bind() on the TCP port. This is inherently unreliable.
    timeout_in_seconds = 10
    poll_time_in_seconds = 0.1
    for i in xrange(int(timeout_in_seconds / poll_time_in_seconds)):
      # On Mac OS X, we have to create a new socket FD for each retry.
      sock = socket.socket()
      # Do not delay sending small packets. This significantly speeds up debug
      # stub tests.
      sock.setsockopt(socket.IPPROTO_TCP, socket.TCP_NODELAY, True)
      try:
        sock.connect(addr)
      except socket.error:
        # Retry after a delay.
        time.sleep(poll_time_in_seconds)
      else:
        return sock
    raise Exception('Could not connect to the debug stub in %i seconds'
                    % timeout_in_seconds)

  def _GetReply(self):
    reply = ''
    message_finished = re.compile('#[0-9a-fA-F]{2}')
    while True:
      data = self._socket.recv(1024)
      if len(data) == 0:
        raise AssertionError('EOF on socket reached with '
                             'incomplete reply message: %r' % reply)
      reply += data
      if message_finished.match(reply[-3:]):
        break
    match = re.match('\+?\$([^#]*)#([0-9a-fA-F]{2})$', reply)
    if match is None:
      raise AssertionError('Unexpected reply message: %r' % reply)
    reply_body = match.group(1)
    checksum = match.group(2)
    expected_checksum = '%02x' % RspChecksum(reply_body)
    if checksum != expected_checksum:
      raise AssertionError('Bad RSP checksum: %r != %r' %
                           (checksum, expected_checksum))
    # Send acknowledgement.
    self._socket.send('+')
    return reply_body

  # Send an rsp message, but don't wait for or expect a reply.
  def RspSendOnly(self, data):
    msg = '$%s#%02x' % (data, RspChecksum(data))
    return self._socket.send(msg)

  def RspRequest(self, data):
    self.RspSendOnly(data)
    reply = self._GetReply()
    return reply

  def Close(self):
    self._socket.close()


def PopenDebugStub(command):
  EnsurePortIsAvailable()
  return subprocess.Popen(command)


def KillProcess(process):
  if process.returncode is not None:
    # kill() won't work if we've already wait()'ed on the process.
    return
  try:
    process.kill()
  except OSError:
    if sys.platform == 'win32':
      # If process is already terminated, kill() throws
      # "WindowsError: [Error 5] Access is denied" on Windows.
      pass
    else:
      raise
  process.wait()


class LaunchDebugStub(object):
  def __init__(self, command):
    self._proc = PopenDebugStub(command)

  def __enter__(self):
    try:
      return GdbRspConnection()
    except:
      KillProcess(self._proc)
      raise

  def __exit__(self, exc_type, exc_value, traceback):
    KillProcess(self._proc)


def AssertEquals(x, y):
  if x != y:
    raise AssertionError('%r != %r' % (x, y))

def DecodeHex(data):
  assert len(data) % 2 == 0, data
  return bytes(bytearray([int(data[index * 2 : (index + 1) * 2], 16) for index in xrange(len(data) // 2)]))

def EncodeHex(data):
  return ''.join('%02x' % ord(byte) for byte in data)

def DecodeUInt64Array(data):
  assert len(data) % 16 == 0, data
  result = []
  for index in xrange(len(data) // 16):
    value = 0
    for digit in xrange(7, -1, -1):
      value = value * 256 + int(data[index * 16 + digit * 2 : index * 16 + (digit + 1) * 2], 16)
    result.append(value)
  return result

def AssertReplySignal(reply, signal):
  AssertEquals(ParseThreadStopReply(reply)['signal'], signal)

def ParseThreadStopReply(reply):
  match = re.match('T([0-9a-f]{2})thread-pcs:([0-9a-f]+);thread:([0-9a-f]+);$', reply)
  if not match:
    raise AssertionError('Bad thread stop reply: %r' % reply)
  return {'signal': int(match.group(1), 16),
          'thread_pc': int(match.group(2), 16),
          'thread_id': int(match.group(3), 16)}

def CheckInstructionPtr(connection, expected_ip):
  ip_value = DecodeRegs(connection.RspRequest('g'))['pc']
  AssertEquals(ip_value, expected_ip)

def DecodeRegs(reply):
  defs = REG_DEFS[ARCH]
  names = [reg_name for reg_name, reg_fmt in defs]
  fmt = ''.join([reg_fmt for reg_name, reg_fmt in defs])

  values = struct.unpack_from(fmt, DecodeHex(reply))
  return dict(zip(names, values))

def GetLoadedModules(connection):
  modules = {}
  reply = connection.RspRequest('qXfer:libraries:read')
  AssertEquals(reply[0], 'l')
  library_list = xml.etree.ElementTree.fromstring(reply[1:])
  AssertEquals(library_list.tag, 'library-list')
  for library in library_list:
    AssertEquals(library.tag, 'library')
    section = library.find('section')
    address = section.get('address')
    assert long(address) > 0
    modules[long(address)] = library.get('name')
  return modules

def GetLoadedModuleAddress(connection):
  modules = GetLoadedModules(connection)
  assert len(modules) > 0
  return modules.keys()[0]

def ReadCodeMemory(connection, address, size):
  reply = connection.RspRequest('m%x,%x' % (address, size))
  assert not reply.startswith('E'), reply
  return DecodeHex(reply)