src/systemc/tests/tlm/endian_conv/testall.py

"""
/*****************************************************************************

  Licensed to Accellera Systems Initiative Inc. (Accellera) under one or
  more contributor license agreements.  See the NOTICE file distributed
  with this work for additional information regarding copyright ownership.
  Accellera licenses this file to you under the Apache License, Version 2.0
  (the "License"); you may not use this file except in compliance with the
  License.  You may obtain a copy of the License at

    http://www.apache.org/licenses/LICENSE-2.0

  Unless required by applicable law or agreed to in writing, software
  distributed under the License is distributed on an "AS IS" BASIS,
  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or
  implied.  See the License for the specific language governing
  permissions and limitations under the License.

 *****************************************************************************/

Python Script to test Endianness Conversion Functions for OSCI TLM-2

There is a simple testbench programme in C++ which runs a single
transaction through a single conversion function, to a simple target
memory and back.  This script will run the testbench many times and test for
- incomplete execution, seg-faults, etc
- distributability of conversion function: each transaction should have
the same effect on initiator/target memory as a set of smaller transactions
that sum to it
- equivalence: all conversion functions should have the same functional
effect as each other

The approach is to provide the initial state of the initiator and
target memory (as strings) and to capture their final states, so that
only the effect of the transaction on the data buffers is measured.

Script works out for itself which conversion functions are legal for a
given transaction and applies all of them.  Note that where data is
wider than bus, only one conversion function is legal and testing is
somewhat limited.

Testing space (select a transaction at random from the space):
- with and without byte-enables (generated at random for each data word
and can be all-zero)
- data widths of (1,2,4,8,16)
- bus widths of (1,2,4,8,16), lower priority for those smaller than data
width
- transaction lengths of (1..32) x data width, higher probability for
lower values
- base address (0..1023) at bus_width steps
- offset address (0..bus width) with a higher priority for 0
- address in initiator buffer uniform random
- read or write
- byte-enable length may be smaller than transasction length
- may be a streaming burst

Transaction breakdown
- individual words (always)
- one random breakdown with each segment containing between 1 and N-1
words, where N is the length in words
- one breakdown with two segments interleaved, using (additional) byte
enables to suppress the part where the other segment is active

Data buffer definition:  starts at 0, randomly filled
with lower case letters and numbers.  Size 2 kB.  Addresses are limited to
1 kB.
"""


import random
import string


class transaction:
  """ contains read_not_write, address, length, byte_enable,
      bus_width, data_width, data_pointer, stream_width """
  def __init__(self, **a):  self.__dict__ = a
  def __str__(self):
    if self.read_not_write:  a = "R: "
    else:  a = "W: "
    a += "addr = %d, len = %d, bus = %d, word = %d, data = %d" % \
      (self.address, self.length, self.bus_width, self.data_width, \
       self.data_pointer)
    if self.byte_enable:  a += ", be = " + self.byte_enable
    else:  a += ", be = x"
    a += ", sw = %d" % (self.stream_width)
    return a


def txn_generator(nr):
  pr_read = 0.5
  pr_byte_enable = 0.5
  pr_enabled = 0.5
  bus_widths = [1, 2, 4, 8, 16]
  data_widths = [1, 2, 4, 8, 16] + [1, 2, 4, 8] + [1, 2, 4] + [1, 2]
  lengths = list(range(1,33)) + list(range(1,17)) + list(range(1,9)) + list(range(1,5)) + list(range(1,3))
  pr_short_be = 0.2
  pr_stream = 0.1
  nr_generated = 0
  while nr_generated < nr:
    # create a random transaction
    bus_width = random.choice(bus_widths)
    while True:
      data_width = random.choice(data_widths)
      if data_width <= bus_width:  break
      if random.random() < 0.25:  break
    length = random.choice(lengths)
    addr_base = random.choice(list(range(0,1024,bus_width)))
    addr_offset = random.choice(list(range(bus_width))+[0]*int(bus_width//2))
    txn = transaction(
      bus_width = bus_width,
      data_width = data_width,
      read_not_write = random.random() < pr_read,
      length = length * data_width,
      address = addr_base + addr_offset,
      byte_enable = False,
      stream_width = length * data_width,
      data_pointer = random.randint(0,1023)
    )
    if random.random() < pr_byte_enable:
      belen = length
      if random.random() < pr_short_be:
        belen = min(random.choice(lengths), length)
      bep = ["0" * data_width, "1" * data_width]
      txn.byte_enable = "".join([random.choice(bep) for x in range(belen)])
    if random.random() < pr_stream and length > 1:
      strlen = length
      while True:
        strlen -= 1
        if strlen == 1 or \
          (random.random() < 0.5 and (length/strlen)*strlen == length):
          break
      txn.stream_width = strlen * data_width
    nr_generated += 1
    yield txn

# test code for transaction generator
if False:
  for t in txn_generator(20):
    print(t)
  raise Exception
# end test code


class memory_state_cl:
  buffer_size = 2048
  repeats = 10 * buffer_size // 36
  population = (string.ascii_lowercase + string.digits) * int(repeats)
  def __init__(self):
    self.initiator = "".join(
      random.sample(memory_state_cl.population, memory_state_cl.buffer_size))
    self.target =  "".join(
      random.sample(memory_state_cl.population, memory_state_cl.buffer_size))
  def copy(self):
    r = memory_state_cl()
    r.initiator = self.initiator
    r.target = self.target
    return r
  def __eq__(self, golden):
    return self.initiator==golden.initiator and self.target==golden.target
  def __ne__(self, golden):
    return self.initiator!=golden.initiator or self.target!=golden.target
  def __str__(self):
    return "initiator = " + self.initiator + "\n" + "target = " + self.target


# all fragmentation generators
def __FRAG__null(txn):
  yield txn

def __FRAG__word(txn):
  curr_address = txn.address
  reset_address = curr_address + txn.stream_width
  if txn.byte_enable:
    full_byte_enable = txn.byte_enable * (1+txn.length/len(txn.byte_enable))
  be_pos = 0
  d_pos = txn.data_pointer
  end = txn.length + d_pos
  while d_pos < end:
    new_txn = transaction(
      bus_width = txn.bus_width,
      data_width = txn.data_width,
      read_not_write = txn.read_not_write,
      length = txn.data_width,
      address = curr_address,
      byte_enable = False,
      stream_width = txn.data_width,
      data_pointer = d_pos
    )
    curr_address += txn.data_width
    if curr_address == reset_address:  curr_address = txn.address
    d_pos += txn.data_width
    if txn.byte_enable:
      new_txn.byte_enable = full_byte_enable[be_pos:be_pos+txn.data_width]
      be_pos += txn.data_width
    yield new_txn

def __FRAG__stream(txn):
  if txn.byte_enable:
    full_byte_enable = txn.byte_enable * (1+txn.length/len(txn.byte_enable))
  be_pos = 0
  bytes_done = 0
  while bytes_done < txn.length:
    new_txn = transaction(
      bus_width = txn.bus_width,
      data_width = txn.data_width,
      read_not_write = txn.read_not_write,
      length = txn.stream_width,
      address = txn.address,
      byte_enable = False,
      stream_width = txn.stream_width,
      data_pointer = bytes_done + txn.data_pointer
    )
    if txn.byte_enable:
      new_txn.byte_enable = full_byte_enable[be_pos:be_pos+txn.stream_width]
      be_pos += txn.stream_width
    yield new_txn
    bytes_done += txn.stream_width

def __FRAG__random(stream_txn):
  for txn in __FRAG__stream(stream_txn):
    # txn has full byte enables and no stream feature guaranteed
    pr_nofrag = 0.5
    end_address = txn.address + txn.length
    curr_address = txn.address
    be_pos = 0
    d_pos = txn.data_pointer
    while curr_address < end_address:
      new_txn = transaction(
        bus_width = txn.bus_width,
        data_width = txn.data_width,
        read_not_write = txn.read_not_write,
        length = txn.data_width,
        address = curr_address,
        byte_enable = txn.byte_enable,
        stream_width = txn.data_width,
        data_pointer = d_pos
      )
      curr_address += txn.data_width
      d_pos += txn.data_width
      if txn.byte_enable:
        new_txn.byte_enable = txn.byte_enable[be_pos:be_pos+txn.data_width]
        be_pos += txn.data_width
      while random.random() < pr_nofrag and curr_address < end_address:
        new_txn.length += txn.data_width
        new_txn.stream_width += txn.data_width
        curr_address += txn.data_width
        d_pos += txn.data_width
        if txn.byte_enable:
          new_txn.byte_enable += txn.byte_enable[be_pos:be_pos+txn.data_width]
          be_pos += txn.data_width
      yield new_txn

def __FRAG__randinterleave(stream_txn):
  for txn in __FRAG__stream(stream_txn):
    # txn has full byte enables and no stream feature guaranteed
    pr_frag = 0.5
    txns = [ transaction(
      bus_width = txn.bus_width,
      data_width = txn.data_width,
      read_not_write = txn.read_not_write,
      length = txn.length,
      address = txn.address,
      byte_enable = "",
      stream_width = txn.length,
      data_pointer = txn.data_pointer
    ), transaction(
      bus_width = txn.bus_width,
      data_width = txn.data_width,
      read_not_write = txn.read_not_write,
      length = txn.length,
      address = txn.address,
      byte_enable = "",
      stream_width = txn.length,
      data_pointer = txn.data_pointer
    ) ]
    curr = 0
    be_pos = 0
    on = "1" * txn.data_width
    off = "0" * txn.data_width
    while be_pos < txn.length:
      if txn.byte_enable:  bew = txn.byte_enable[be_pos:be_pos+txn.data_width]
      else:  bew = on
      txns[curr].byte_enable += bew
      txns[1-curr].byte_enable += off
      be_pos += txn.data_width
      if random.random() < pr_frag:  curr = 1-curr
    yield txns[0]
    yield txns[1]

fragmenters = [globals()[n] for n in globals().keys() if n[:8]=="__FRAG__"]

# test code for fragmenters
if False:
  for t in txn_generator(1):
    print(t)
    print()
    for u in fragmenters[4](t):
      print(u)
  raise Exception
# end test code


# conversion functions are determined by an index (shared with C++) and
# a function that tests if they can be applied to a transaction
def __CHCK__generic(txn):
  __CHCK__generic.nr = 0
  return True

def __CHCK__word(txn):
  __CHCK__word.nr = 1
  if txn.data_width > txn.bus_width:  return False
  if txn.stream_width < txn.length:  return False
  if txn.byte_enable and len(txn.byte_enable) < txn.length:  return False
  return True

def __CHCK__aligned(txn):
  __CHCK__aligned.nr = 2
  if txn.data_width > txn.bus_width:  return False
  if txn.stream_width < txn.length:  return False
  if txn.byte_enable and len(txn.byte_enable) < txn.length:  return False
  base_addr = txn.address / txn.bus_width
  if base_addr * txn.bus_width != txn.address:  return False
  nr_bus_words = txn.length / txn.bus_width
  if nr_bus_words * txn.bus_width != txn.length:  return False
  return True

def __CHCK__single(txn):
  __CHCK__single.nr = 3
  if txn.length != txn.data_width:  return False
  base_addr = txn.address / txn.bus_width
  end_base_addr = (txn.address + txn.length - 1) / txn.bus_width
  if base_addr != end_base_addr:  return False
  return True

def __CHCK__local_single(txn):
  __CHCK__local_single.nr = 4
  if txn.length != txn.data_width:  return False
  return True

all_converters = [globals()[n] for n in globals().keys() if n[:8]=="__CHCK__"]
for x in all_converters:  x.usage = 0


class TesterFailure(Exception):  pass
class SystemCFailure(Exception):  pass
class ConverterDifference(Exception):  pass
class FragmenterDifference(Exception):  pass

from subprocess import Popen, PIPE

# test a single fragment in multiple ways
def test_a_fragment(f, ms):
  # f is the (fragment of a) transaction
  # ms is the memory state to use at start of test

  # run the same fragment through all applicable conversion functions
  # and check they all do the same thing
  # use the same sub-process for all of them

  # build complete stdin
  convs = [c for c in all_converters if c(f)]
  if len(convs) == 0:  raise TesterFailure(f.str())
  txtin = "\n".join(
    [("%s\n%s\nconverter = %d\n" % (f, ms, c.nr)) for c in convs])

  # run and get stdout
  txtout = "no output"
  try:
    sp = Popen("../build-unix/test_endian_conv.exe", stdin=PIPE, stdout=PIPE)
    txtout = sp.communicate(txtin)[0]
    tmp = txtout.splitlines()
    initiators = [l.split()[-1] for l in tmp if l[:14] == "  initiator = "]
    targets = [l.split()[-1] for l in tmp if l[:11] == "  target = "]
  except:
    raise SystemCFailure("\n" + txtin + txtout)
  if sp.returncode != 0:  raise SystemCFailure("\n" + txtin + txtout)
  if len(initiators) != len(convs):  raise SystemCFailure("\n" + txtin + txtout)
  if len(targets) != len(convs):  raise SystemCFailure("\n" + txtin + txtout)
  for c in convs:  c.usage += 1

  ms_out = memory_state_cl()
  ms_out.initiator = initiators[0]
  ms_out.target = targets[0]
  for i in range(1,len(convs)):
    if initiators[i]!=ms_out.initiator or targets[i]!=ms_out.target:
      raise ConverterDifference("""
%s
start memory:
%s
converter = %d
golden memory:
%s
actual memory:
%s""" % (f, ms, i, golden_ms, ms_out))

  return ms_out


# main loop

from sys import argv

print("Testing Endianness Conversion Functions")
print("March 2008")
print("OSCI TLM-2")

try:  nr_txns_to_test = int(argv[1])
except:
  print("No command line input for number of tests, using default")
  nr_txns_to_test = 1000

print("Number to test:", nr_txns_to_test)

# generate and test a number of transactions
for txn in txn_generator(nr_txns_to_test):

  # each transaction has a random initial memory state
  initial_memory = memory_state_cl()

  # iterate over all defined fragmentation functions
  first_time = True
  for fragmenter in fragmenters:

    # all versions of the transaction start in the same place
    memory_state = initial_memory.copy()

    # now iterate over the fragments of the transaction, accumulating
    # the memory state
    for partial_txn in fragmenter(txn):
      memory_state = test_a_fragment(partial_txn, memory_state)

    if first_time:
      golden_memory_state = memory_state.copy()
      first_time = False
    else:
      if memory_state != golden_memory_state:  raise FragmenterDifference("""
fragmenter: %s
transaction:
%s
start memory:
%s
golden memory:
%s
actual memory:
%s""" % (fragmenter, txn, initial_memory, golden_memory_state, memory_state))

  print("."),
print()


print("Conversion functions usage frequency:")
for c in all_converters:
  print(c.nr, c.__name__, c.usage)