Writing Traffic Generators in Python - Writing Traffic Generators in Python - 2023.1 English - UG1393

Vitis Unified Software Platform Documentation: Application Acceleration Development (UG1393)

Document_ID
UG1393
Release_Date
2023-07-17
Version
2023.1 English

You must also include a traffic generator process while simulating your application to generate data traffic on the I/O traffic generators, or to capture output data from the emulation process. The AMD provided Python or C++ library can be used to create the traffic generator code as described below. Also, an application can communicate to multiple I/O interface. It is not necessary to have each instance of I/O utilities to be in a separate process/thread. In case your application demands it, you might consider the non-blocking version APIs (details provided in the following section).

The following is the list of Python traffic generator APIs along with their usage.

Table 1. List of Python APIs used in external traffic generator
API Behavior
ipc_axis_master_util("<sender_io_name>") To instantiate the master
ipc_axis_slave_util("<receiver_io_name>") To instantiate the slave
b_transport(packet) Blocking API to send the AXI4-Stream packets
sample_transaction() Blocking API to receive the AXI4-Stream packets
nb_transport(packet) Non-blocking API call to send the AXI4-Stream packets
nb_sample_transaction() Non-blocking API call to receive the AXI4-Stream packets
end_of_simulation()

End and exit the simulation using this API

Preferable use in External TG with sw_emu/hw_emu

disconnect()

To disconnect the socket connections (DataIn, DataOut etc)

Preferable use in External TG with AI Enginesim/x86sim

The following is a list of Python traffic generator APIs along with their usage.

ipc_axis_master_util (<sender_io_name>) 

Used to instantiate the sender port in the Python script.

API usage
Creates a user object using the API for sender:
sender_obj = ipc_axis_master_util (<sender_io_name>) 
Parameters
Name of the external port that sends data:
<sender_io_name>
Example
#Instantiating AXI Master Utilities 
data_in = ipc_axis_master_util("DataIn")
DataIn is the external port name and data_in is the user object.

ipc_axis_slave_util(<receiver_io_name>)

Used to instantiate the receiver port in the Python script.

API usage
Creates a user object using the API for the receiver:
receiver_obj = ipc_axis_slave_util (<receiver_io_name>)
Parameters
Name of the external port that receives the data:
receiver_io_name
Example
#Instantiating AXI Slave Utilities 
data_out = ipc_axis_slave_util("DataOut")
In the example, DataOut is the external port and data_out is the user object.

b_transport(packet)

The blocking API to send the AXI stream.
API usage
Calls the API using the object created
sender_obj.b_transport(<packet>) 
Parameters
AXI stream packet:
packet
This AXI stream has the following attributes that need to be set.
The data that needs to be sent:
data
The TLAST value to mark the end of the packet. This is the user's choice based on the application.
tlast
If not specified, the default value is true (i.e. every data beat is a transaction). If tlast = false, no TLAST is driven. If tlast = true (specified by the user based on the application), the last data beat has TLAST=true.
Some other optional AXI stream packet fields are:
tuser
tkeep
Example
#Create packet
data_packet = xtlm_ipc.axi_stream_packet() // API to generate the stream packet
data = generate_data() # custom user code to generate the data
data_packet.data = data 

data_packet.tlast = True #AXI Stream Fields
#Optional AXI Stream Parameters
data_packet.tuser = optional data
data_packet.tkeep = optional data 

#Send Transaction
data_in.b_transport(data_packet)

sample_transaction()

Blocking API to receive the AXI stream packets.
API Usage
Calls the API using the receiver_obj created:
packet = receiver_obj.sample_transaction()
Example
#Sample the packets
data_packet = data_out.sample_transaction() # Will wait until data is received.
Here, data_out is the receiver object and data_packet is the received packet.

ipc_axis_master_nb_util (<sender_io_name>) 

Instantiates the sender port for non-blocking in the python script.
API Usage
Creates a user object using the API for sender:
sender_obj = ipc_axis_master_nb_util (<sender_io_name>) 
Parameters
Name of the external port that sends the data:
<sender_io_name>
Example
#Instantiating AXI Master Utilities 
data_in = ipc_axis_master_nb_util("DataIn")

ipc_axis_slave_nb_util(<receiver_io_name>)

Used to instantiate the receiver port for non-blocking in the Python script

API Usage
Creates a user object using the API for receiver
receiver_obj = ipc_axis_slave_util (<receiver_io_name>)
Parameters
Name of the external port that sends the data
receiver_io_name
Example
#Instantiating AXI Slave Utilities
data_out = ipc_axis_slave_nb_util("DataOut")

nb_transport(<packet>)

Non-blocking API used to send AXI stream packets.

API Usage
Calls the API using sender_obj:
sender_obj.nb_transport(<packet>) 
Parameters
AXI stream packet:
packet
This AXI stream packet is created by the user which has the following attributes:
The data that needs to be sent:
data
The TLAST value to mark the end of the packet. This is the user's choice based on the application. If not specified, the default value is true (i.e. every data beat is a transaction). If tlast = false, no TLAST is driven. If tlast = true (specified by the user based on the application), the last data beat has TLAST=true.
tlast
Some other optional AXI stream packet fields are:
tuser
tkeep
Example
#Create packet
data_packet = xtlm_ipc.axi_stream_packet()
data = generate_data() # custom user code to generate the desired data
data_packet.data_length = "DATA_LENGTH"
data_packet.data = data
data_packet.tlast = True #to mark end of packet. This is user's choice based on application
data_in.nb_transport(data_packet)

nb_sample_transaction()

A non-blocking API to receive the AXI stream packets.

API Usage
packet = receiver_obj.nb_sample_transaction() 
Example
#Sample the packets
# Will return data if it is available
# Will return empty list if nothing is available
# Will not wait or stop the execution until data is present to sample
data_packet = data_out.nb_sample_transaction()

Here, data_out is the receiver object and data_packet is the received packet.

no_of_available_txns

For non-blocking, this API is used to get to get number of transactions received.

API Usage
Calls the API using the user object for the receiver.
Example
data_out = ipc_axis_slave_nb_util("ReceiverName")
data = []

# Checks if any data is present to receive
if(slave.no_of_available_txns != 0) :
data = data_out.nb_sample_transaction() # fill data with respective received data

end_of_simulation()

Ends and exits the simulation.

API Usage
Call the API using the sender object
sender_obj.end_of_simulation()
Example
data_in.end_of_simulation()

Preferably used with with sw_emu/hw_emu.

disconnect()

To disconnect the socket connections (DataIn, DataOut etc)

API Usage
Calls the API using sender or receiver object:
sender_obj.disconect()
receiver_obj.disconnect()
Example
data_in.disconnect() 
data_out.disconnect()

Preferably used with x86sim/AIEsim.

A full system-level example is available for external IO in Python: https://github.com/Xilinx/Vitis_Accel_Examples/tree/master/emulation/external_io_py