4.3.2. Simulation Configuration

4.3.2.1. Base

class simbricks.orchestration.simulation.base.Simulator(simulation: sim_base.Simulation, executable: str, name: str = '')

Base class for all simulators.

T = ~T

filter_components_by_pred(pred: tp.Callable[[sys_conf.Component], bool], ty: type[T] = <class 'simbricks.orchestration.system.base.Component'>) → list[T]

filter_components_by_type(ty: type[T]) → list[T]

property extra_args: str

property wait_terminate: bool

toJSON() → dict

classmethod fromJSON(simulation: Simulation, json_obj: dict) → Self

static filter_sockets(sockets: list[Socket], filter_type: SockType = SockType.LISTEN) → list[Socket]

static split_sockets_by_type(sockets: list[Socket]) → tuple[list[Socket], list[Socket]]

static get_unique_latency_period_sync(channels: list[Channel]) → tuple[int, int, bool]

get_parameters_url(inst: Instantiation, socket: Socket, channel: Channel | None = None, sync: bool | None = None, latency: int | None = None, sync_period: int | None = None) → str

get_interface_url(inst: Instantiation, intf: Interface) → str

components() → set[Component]

resreq_cores() → int

Number of cores this simulator requires during execution.

This is used for scheduling multiple runs and experiments.

resreq_mem() → int

Number of memory in MB this simulator requires during execution.

This is used for scheduling multiple runs and experiments.

full_name() → str

Full name of the simulator.

add(comp: Component) → None

get_channels() → list[Channel]

static filter_channels_by_sys_type(channels: list[sim_chan.Channel], ty: type[T]) → list[T]

abstract run_cmd(inst: Instantiation) → str

Command to execute this simulator.

checkpoint_commands() → list[str]

cleanup_commands() → list[str]

abstract supported_socket_types(interface: Interface) → set[SockType]

sockets_cleanup(inst: Instantiation) → list[Socket]

sockets_wait(inst: Instantiation) → list[Socket]

start_delay() → int

supports_checkpointing() → bool

async prepare(inst: Instantiation) → None

id() → int

class simbricks.orchestration.simulation.base.DummySimulator(simulation: sim_base.Simulation, executable: str, name: str = '')

run_cmd(inst: Instantiation) → str

Command to execute this simulator.

supported_socket_types(interface: Interface) → set[SockType]

classmethod fromJSON(simulation: Simulation, json_obj: dict) → Self

T = ~T

add(comp: Component) → None

checkpoint_commands() → list[str]

cleanup_commands() → list[str]

components() → set[Component]

property extra_args: str

static filter_channels_by_sys_type(channels: list[sim_chan.Channel], ty: type[T]) → list[T]

filter_components_by_pred(pred: tp.Callable[[sys_conf.Component], bool], ty: type[T] = <class 'simbricks.orchestration.system.base.Component'>) → list[T]

filter_components_by_type(ty: type[T]) → list[T]

static filter_sockets(sockets: list[Socket], filter_type: SockType = SockType.LISTEN) → list[Socket]

full_name() → str

Full name of the simulator.

get_channels() → list[Channel]

get_interface_url(inst: Instantiation, intf: Interface) → str

get_parameters_url(inst: Instantiation, socket: Socket, channel: Channel | None = None, sync: bool | None = None, latency: int | None = None, sync_period: int | None = None) → str

static get_unique_latency_period_sync(channels: list[Channel]) → tuple[int, int, bool]

id() → int

async prepare(inst: Instantiation) → None

resreq_cores() → int

Number of cores this simulator requires during execution.

This is used for scheduling multiple runs and experiments.

resreq_mem() → int

Number of memory in MB this simulator requires during execution.

This is used for scheduling multiple runs and experiments.

sockets_cleanup(inst: Instantiation) → list[Socket]

sockets_wait(inst: Instantiation) → list[Socket]

static split_sockets_by_type(sockets: list[Socket]) → tuple[list[Socket], list[Socket]]

start_delay() → int

supports_checkpointing() → bool

toJSON() → dict

property wait_terminate: bool

name: str

class simbricks.orchestration.simulation.base.Simulation(name: str, system: System)

Base class for all simulation experiments.

Contains the simulators to be run and experiment-wide parameters.

name

This experiment’s name.

Can be used to run only a selection of experiments.

timeout: int | None

Timeout for experiment in seconds.

toJSON() → dict

Serializes a Simulation.

Note: The sys_sim_map is not serialized as the sim_list stores the required information implicitly as Simulators,

when serialized store a list of their corresponding components.

classmethod fromJSON(system: System, json_obj: dict, enforce_dummies: bool = False) → Self

Deserializes a Simulation.

Note: The sys_sim_map is not deserialized as the map is restored when deserializing the the sim_list.

This is possible as each Simulator stores when a list of their corresponding components when serialized.

add_sim(sim: Simulator)

add_spec_sim_map(sys: Component, sim: Simulator)

Add a mapping from specification to simulation instance

is_channel_instantiated(chan: Channel) → bool

update_channel_mapping(sys_chan: Channel, sim_chan: Channel) → None

retrieve_or_create_channel(chan: Channel) → Channel

get_channel(chan: Channel) → Channel

get_channel_by_id(id: int) → Channel

all_simulators() → list[Simulator]

get_simulator(id: int) → Simulator

id() → int

get_all_channels(lazy: bool = False) → list[Channel]

enable_synchronization(amount: int | None = None, ratio: Time | None = None) → None

resreq_mem() → int

Memory required to run all simulators in this experiment.

resreq_cores() → int

Number of Cores required to run all simulators in this experiment.

find_sim(comp: sys_conf.Component) → sim_base.Simulator

Returns the used simulator object for the system component.

async prepare(inst: Instantiation) → None

any_supports_checkpointing() → bool

4.3.2.2. Channel

class simbricks.orchestration.simulation.channel.Channel(chan: Channel)

sync_period: int

The synchronization period in nanoseconds. For SimBricks to function properly in sync mode, the sync period must not be larger than a channels latency.

toJSON()

classmethod fromJSON(simulation: Simulation, json_obj: dict) → Self

id() → int

set_sync_period(amount: int, ratio: Time = Time.Nanoseconds) → None

4.3.2.3. Host

class simbricks.orchestration.simulation.host.HostSim(simulation: Simulation, executable: str, name='')

toJSON() → dict

classmethod fromJSON(simulation: Simulation, json_obj: dict) → Self

full_name() → str

Full name of the simulator.

add(host: Host)

abstract supported_image_formats() → list[str]

abstract async copy_disk_image(inst: Instantiation, disk_image: DiskImage, ident: str) → str

async prepare(inst: Instantiation)

supported_socket_types(interface: Interface) → set[SockType]

T = ~T

checkpoint_commands() → list[str]

cleanup_commands() → list[str]

components() → set[Component]

property extra_args: str

static filter_channels_by_sys_type(channels: list[sim_chan.Channel], ty: type[T]) → list[T]

filter_components_by_pred(pred: tp.Callable[[sys_conf.Component], bool], ty: type[T] = <class 'simbricks.orchestration.system.base.Component'>) → list[T]

filter_components_by_type(ty: type[T]) → list[T]

static filter_sockets(sockets: list[Socket], filter_type: SockType = SockType.LISTEN) → list[Socket]

get_channels() → list[Channel]

get_interface_url(inst: Instantiation, intf: Interface) → str

get_parameters_url(inst: Instantiation, socket: Socket, channel: Channel | None = None, sync: bool | None = None, latency: int | None = None, sync_period: int | None = None) → str

static get_unique_latency_period_sync(channels: list[Channel]) → tuple[int, int, bool]

id() → int

resreq_cores() → int

Number of cores this simulator requires during execution.

This is used for scheduling multiple runs and experiments.

resreq_mem() → int

Number of memory in MB this simulator requires during execution.

This is used for scheduling multiple runs and experiments.

abstract run_cmd(inst: Instantiation) → str

Command to execute this simulator.

sockets_cleanup(inst: Instantiation) → list[Socket]

sockets_wait(inst: Instantiation) → list[Socket]

static split_sockets_by_type(sockets: list[Socket]) → tuple[list[Socket], list[Socket]]

start_delay() → int

supports_checkpointing() → bool

property wait_terminate: bool

name: str

class simbricks.orchestration.simulation.host.Gem5Sim(simulation: Simulation)

name: str

supports_checkpointing() → bool

resreq_cores() → int

Number of cores this simulator requires during execution.

This is used for scheduling multiple runs and experiments.

resreq_mem() → int

Number of memory in MB this simulator requires during execution.

This is used for scheduling multiple runs and experiments.

supported_image_formats() → list[str]

toJSON() → dict

classmethod fromJSON(simulation: Simulation, json_obj: dict) → Self

async copy_disk_image(inst: Instantiation, disk_image: DiskImage, ident: str)

async prepare(inst: Instantiation) → None

checkpoint_commands() → list[str]

cleanup_commands() → list[str]

run_cmd(inst: Instantiation) → str

Command to execute this simulator.

T = ~T

add(host: Host)

components() → set[Component]

property extra_args: str

static filter_channels_by_sys_type(channels: list[sim_chan.Channel], ty: type[T]) → list[T]

filter_components_by_pred(pred: tp.Callable[[sys_conf.Component], bool], ty: type[T] = <class 'simbricks.orchestration.system.base.Component'>) → list[T]

filter_components_by_type(ty: type[T]) → list[T]

static filter_sockets(sockets: list[Socket], filter_type: SockType = SockType.LISTEN) → list[Socket]

full_name() → str

Full name of the simulator.

get_channels() → list[Channel]

get_interface_url(inst: Instantiation, intf: Interface) → str

get_parameters_url(inst: Instantiation, socket: Socket, channel: Channel | None = None, sync: bool | None = None, latency: int | None = None, sync_period: int | None = None) → str

static get_unique_latency_period_sync(channels: list[Channel]) → tuple[int, int, bool]

id() → int

sockets_cleanup(inst: Instantiation) → list[Socket]

sockets_wait(inst: Instantiation) → list[Socket]

static split_sockets_by_type(sockets: list[Socket]) → tuple[list[Socket], list[Socket]]

start_delay() → int

supported_socket_types(interface: Interface) → set[SockType]

property wait_terminate: bool

class simbricks.orchestration.simulation.host.QemuSim(simulation: Simulation)

name: str

resreq_cores() → int

Number of cores this simulator requires during execution.

This is used for scheduling multiple runs and experiments.

resreq_mem() → int

Number of memory in MB this simulator requires during execution.

This is used for scheduling multiple runs and experiments.

supported_image_formats() → list[str]

toJSON() → dict

classmethod fromJSON(simulation: Simulation, json_obj: dict) → Self

async copy_disk_image(inst: Instantiation, disk_image: DiskImage, ident: str) → str

checkpoint_commands() → list[str]

cleanup_commands() → list[str]

run_cmd(inst: Instantiation) → str

Command to execute this simulator.

T = ~T

add(host: Host)

components() → set[Component]

property extra_args: str

static filter_channels_by_sys_type(channels: list[sim_chan.Channel], ty: type[T]) → list[T]

filter_components_by_pred(pred: tp.Callable[[sys_conf.Component], bool], ty: type[T] = <class 'simbricks.orchestration.system.base.Component'>) → list[T]

filter_components_by_type(ty: type[T]) → list[T]

static filter_sockets(sockets: list[Socket], filter_type: SockType = SockType.LISTEN) → list[Socket]

full_name() → str

Full name of the simulator.

get_channels() → list[Channel]

get_interface_url(inst: Instantiation, intf: Interface) → str

get_parameters_url(inst: Instantiation, socket: Socket, channel: Channel | None = None, sync: bool | None = None, latency: int | None = None, sync_period: int | None = None) → str

static get_unique_latency_period_sync(channels: list[Channel]) → tuple[int, int, bool]

id() → int

async prepare(inst: Instantiation)

sockets_cleanup(inst: Instantiation) → list[Socket]

sockets_wait(inst: Instantiation) → list[Socket]

static split_sockets_by_type(sockets: list[Socket]) → tuple[list[Socket], list[Socket]]

start_delay() → int

supported_socket_types(interface: Interface) → set[SockType]

supports_checkpointing() → bool

property wait_terminate: bool

4.3.2.4. PCIDev

class simbricks.orchestration.simulation.pcidev.PCIDevSim(simulation: Simulation, executable: str, name: str)

Base class for PCIe device simulators.

full_name() → str

Full name of the simulator.

supported_socket_types(interface: Interface) → set[SockType]

T = ~T

add(comp: Component) → None

checkpoint_commands() → list[str]

cleanup_commands() → list[str]

components() → set[Component]

property extra_args: str

static filter_channels_by_sys_type(channels: list[sim_chan.Channel], ty: type[T]) → list[T]

filter_components_by_pred(pred: tp.Callable[[sys_conf.Component], bool], ty: type[T] = <class 'simbricks.orchestration.system.base.Component'>) → list[T]

filter_components_by_type(ty: type[T]) → list[T]

static filter_sockets(sockets: list[Socket], filter_type: SockType = SockType.LISTEN) → list[Socket]

classmethod fromJSON(simulation: Simulation, json_obj: dict) → Self

get_channels() → list[Channel]

get_interface_url(inst: Instantiation, intf: Interface) → str

get_parameters_url(inst: Instantiation, socket: Socket, channel: Channel | None = None, sync: bool | None = None, latency: int | None = None, sync_period: int | None = None) → str

static get_unique_latency_period_sync(channels: list[Channel]) → tuple[int, int, bool]

id() → int

async prepare(inst: Instantiation) → None

resreq_cores() → int

Number of cores this simulator requires during execution.

This is used for scheduling multiple runs and experiments.

resreq_mem() → int

Number of memory in MB this simulator requires during execution.

This is used for scheduling multiple runs and experiments.

abstract run_cmd(inst: Instantiation) → str

Command to execute this simulator.

sockets_cleanup(inst: Instantiation) → list[Socket]

sockets_wait(inst: Instantiation) → list[Socket]

static split_sockets_by_type(sockets: list[Socket]) → tuple[list[Socket], list[Socket]]

start_delay() → int

supports_checkpointing() → bool

toJSON() → dict

property wait_terminate: bool

name: str

class simbricks.orchestration.simulation.pcidev.FEMUSim(simulation: Simulation)

name: str

classmethod fromJSON(simulation: Simulation, json_obj: dict) → Self

add(nic: NVMeSSD)

run_cmd(inst: Instantiation) → str

Command to execute this simulator.

T = ~T

checkpoint_commands() → list[str]

cleanup_commands() → list[str]

components() → set[Component]

property extra_args: str

static filter_channels_by_sys_type(channels: list[sim_chan.Channel], ty: type[T]) → list[T]

filter_components_by_pred(pred: tp.Callable[[sys_conf.Component], bool], ty: type[T] = <class 'simbricks.orchestration.system.base.Component'>) → list[T]

filter_components_by_type(ty: type[T]) → list[T]

static filter_sockets(sockets: list[Socket], filter_type: SockType = SockType.LISTEN) → list[Socket]

full_name() → str

Full name of the simulator.

get_channels() → list[Channel]

get_interface_url(inst: Instantiation, intf: Interface) → str

get_parameters_url(inst: Instantiation, socket: Socket, channel: Channel | None = None, sync: bool | None = None, latency: int | None = None, sync_period: int | None = None) → str

static get_unique_latency_period_sync(channels: list[Channel]) → tuple[int, int, bool]

id() → int

async prepare(inst: Instantiation) → None

resreq_cores() → int

Number of cores this simulator requires during execution.

This is used for scheduling multiple runs and experiments.

resreq_mem() → int

Number of memory in MB this simulator requires during execution.

This is used for scheduling multiple runs and experiments.

sockets_cleanup(inst: Instantiation) → list[Socket]

sockets_wait(inst: Instantiation) → list[Socket]

static split_sockets_by_type(sockets: list[Socket]) → tuple[list[Socket], list[Socket]]

start_delay() → int

supported_socket_types(interface: Interface) → set[SockType]

supports_checkpointing() → bool

toJSON() → dict

property wait_terminate: bool

class simbricks.orchestration.simulation.pcidev.NICSim(simulation: Simulation, executable: str, name: str = '')

Base class for NIC simulators.

toJSON() → dict

classmethod fromJSON(simulation: Simulation, json_obj: dict) → Self

full_name() → str

Full name of the simulator.

add(nic: SimplePCIeNIC)

run_cmd(inst: Instantiation) → str

Command to execute this simulator.

T = ~T

checkpoint_commands() → list[str]

cleanup_commands() → list[str]

components() → set[Component]

property extra_args: str

static filter_channels_by_sys_type(channels: list[sim_chan.Channel], ty: type[T]) → list[T]

filter_components_by_pred(pred: tp.Callable[[sys_conf.Component], bool], ty: type[T] = <class 'simbricks.orchestration.system.base.Component'>) → list[T]

filter_components_by_type(ty: type[T]) → list[T]

static filter_sockets(sockets: list[Socket], filter_type: SockType = SockType.LISTEN) → list[Socket]

get_channels() → list[Channel]

get_interface_url(inst: Instantiation, intf: Interface) → str

get_parameters_url(inst: Instantiation, socket: Socket, channel: Channel | None = None, sync: bool | None = None, latency: int | None = None, sync_period: int | None = None) → str

static get_unique_latency_period_sync(channels: list[Channel]) → tuple[int, int, bool]

id() → int

async prepare(inst: Instantiation) → None

resreq_cores() → int

Number of cores this simulator requires during execution.

This is used for scheduling multiple runs and experiments.

resreq_mem() → int

Number of memory in MB this simulator requires during execution.

This is used for scheduling multiple runs and experiments.

sockets_cleanup(inst: Instantiation) → list[Socket]

sockets_wait(inst: Instantiation) → list[Socket]

static split_sockets_by_type(sockets: list[Socket]) → tuple[list[Socket], list[Socket]]

start_delay() → int

supported_socket_types(interface: Interface) → set[SockType]

supports_checkpointing() → bool

property wait_terminate: bool

name: str

class simbricks.orchestration.simulation.pcidev.I40eNicSim(simulation: Simulation)

name: str

toJSON() → dict

classmethod fromJSON(simulation: Simulation, json_obj: dict) → Self

add(nic: IntelI40eNIC)

run_cmd(inst: Instantiation) → str

Command to execute this simulator.

T = ~T

checkpoint_commands() → list[str]

cleanup_commands() → list[str]

components() → set[Component]

property extra_args: str

static filter_channels_by_sys_type(channels: list[sim_chan.Channel], ty: type[T]) → list[T]

filter_components_by_pred(pred: tp.Callable[[sys_conf.Component], bool], ty: type[T] = <class 'simbricks.orchestration.system.base.Component'>) → list[T]

filter_components_by_type(ty: type[T]) → list[T]

static filter_sockets(sockets: list[Socket], filter_type: SockType = SockType.LISTEN) → list[Socket]

full_name() → str

Full name of the simulator.

get_channels() → list[Channel]

get_interface_url(inst: Instantiation, intf: Interface) → str

get_parameters_url(inst: Instantiation, socket: Socket, channel: Channel | None = None, sync: bool | None = None, latency: int | None = None, sync_period: int | None = None) → str

static get_unique_latency_period_sync(channels: list[Channel]) → tuple[int, int, bool]

id() → int

async prepare(inst: Instantiation) → None

resreq_cores() → int

Number of cores this simulator requires during execution.

This is used for scheduling multiple runs and experiments.

resreq_mem() → int

Number of memory in MB this simulator requires during execution.

This is used for scheduling multiple runs and experiments.

sockets_cleanup(inst: Instantiation) → list[Socket]

sockets_wait(inst: Instantiation) → list[Socket]

static split_sockets_by_type(sockets: list[Socket]) → tuple[list[Socket], list[Socket]]

start_delay() → int

supported_socket_types(interface: Interface) → set[SockType]

supports_checkpointing() → bool

property wait_terminate: bool

mac: str | None

log_file: str | None

class simbricks.orchestration.simulation.pcidev.E1000NIC(simulation: Simulation)

name: str

toJSON() → dict

classmethod fromJSON(simulation: Simulation, json_obj: dict) → Self

add(nic: IntelE1000NIC)

run_cmd(inst: Instantiation) → str

Command to execute this simulator.

T = ~T

checkpoint_commands() → list[str]

cleanup_commands() → list[str]

components() → set[Component]

property extra_args: str

static filter_channels_by_sys_type(channels: list[sim_chan.Channel], ty: type[T]) → list[T]

filter_components_by_pred(pred: tp.Callable[[sys_conf.Component], bool], ty: type[T] = <class 'simbricks.orchestration.system.base.Component'>) → list[T]

filter_components_by_type(ty: type[T]) → list[T]

static filter_sockets(sockets: list[Socket], filter_type: SockType = SockType.LISTEN) → list[Socket]

full_name() → str

Full name of the simulator.

get_channels() → list[Channel]

get_interface_url(inst: Instantiation, intf: Interface) → str

get_parameters_url(inst: Instantiation, socket: Socket, channel: Channel | None = None, sync: bool | None = None, latency: int | None = None, sync_period: int | None = None) → str

static get_unique_latency_period_sync(channels: list[Channel]) → tuple[int, int, bool]

id() → int

async prepare(inst: Instantiation) → None

resreq_cores() → int

Number of cores this simulator requires during execution.

This is used for scheduling multiple runs and experiments.

resreq_mem() → int

Number of memory in MB this simulator requires during execution.

This is used for scheduling multiple runs and experiments.

sockets_cleanup(inst: Instantiation) → list[Socket]

sockets_wait(inst: Instantiation) → list[Socket]

static split_sockets_by_type(sockets: list[Socket]) → tuple[list[Socket], list[Socket]]

start_delay() → int

supported_socket_types(interface: Interface) → set[SockType]

supports_checkpointing() → bool

property wait_terminate: bool

mac: str | None

log_file: str | None