Skip to content

GitLab

Unverified Commit 23867991 authored by Jakob Görgen's avatar Jakob Görgen
Browse files

new symphony folder with new folder structure and packages + removed old orchestration framework

parent a14a0365
Hide whitespace changes
Inline Side-by-side
Showing with 706 additions and 1946 deletions
experiments/simbricks/orchestration/runtime/distributed.py deleted 100644 → 0
View file @ a14a0365
# Copyright 2021 Max Planck Institute for Software Systems, and
# National University of Singapore
#
# Permission is hereby granted, free of charge, to any person obtaining
# a copy of this software and associated documentation files (the
# "Software"), to deal in the Software without restriction, including
# without limitation the rights to use, copy, modify, merge, publish,
# distribute, sublicense, and/or sell copies of the Software, and to
# permit persons to whom the Software is furnished to do so, subject to
# the following conditions:
#
# The above copyright notice and this permission notice shall be
# included in all copies or substantial portions of the Software.
#
# THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND,
# EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF
# MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT.
# IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY
# CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN AN ACTION OF CONTRACT,
# TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION WITH THE
# SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.
import asyncio
import typing as tp
from simbricks.orchestration import proxy
from simbricks.orchestration.exectools import Executor
from simbricks.orchestration.experiments import (
DistributedExperiment, Experiment
)
from simbricks.orchestration.runners import ExperimentDistributedRunner
from simbricks.orchestration.runtime.common import Run, Runtime
class DistributedSimpleRuntime(Runtime):
def __init__(self, executors, verbose=False) -> None:
super().__init__()
self.runnable: tp.List[Run] = []
self.complete: tp.List[Run] = []
self.verbose = verbose
self.executors = executors
self._running: asyncio.Task
def add_run(self, run: Run) -> None:
if not isinstance(run.experiment, DistributedExperiment):
raise RuntimeError('Only distributed experiments supported')
self.runnable.append(run)
async def do_run(self, run: Run) -> None:
runner = ExperimentDistributedRunner(
self.executors,
# we ensure the correct type in add_run()
tp.cast(DistributedExperiment, run.experiment),
run.env,
self.verbose
)
if self.profile_int:
runner.profile_int = self.profile_int
try:
for executor in self.executors:
await run.prep_dirs(executor)
await runner.prepare()
except asyncio.CancelledError:
# it is safe to just exit here because we are not running any
# simulators yet
return
run.output = await runner.run() # already handles CancelledError
self.complete.append(run)
# if the log is huge, this step takes some time
if self.verbose:
print(
f'Writing collected output of run {run.name()} to JSON file ...'
)
run.output.dump(run.outpath)
async def start(self) -> None:
for run in self.runnable:
if self._interrupted:
return
self._running = asyncio.create_task(self.do_run(run))
await self._running
def interrupt_handler(self) -> None:
if self._running:
self._running.cancel()
def auto_dist(
e: Experiment,
execs: tp.List[Executor],
proxy_type: str = 'sockets'
) -> DistributedExperiment:
"""
Converts an Experiment into a DistributedExperiment.
Assigns network to executor zero, and then round-robin assignment of hosts
to executors, while also assigning all nics for a host to the same executor.
"""
if len(execs) < 2:
raise RuntimeError('auto_dist needs at least two hosts')
elif len(execs) > 2:
print('Warning: currently auto_dist only uses the first two hosts')
if proxy_type == 'sockets':
proxy_listener_c = proxy.SocketsNetProxyListener
proxy_connecter_c = proxy.SocketsNetProxyConnecter
elif proxy_type == 'rdma':
proxy_listener_c = proxy.RDMANetProxyListener
proxy_connecter_c = proxy.RDMANetProxyConnecter
else:
raise RuntimeError('Unknown proxy type specified')
# Create the distributed experiment
de = DistributedExperiment(e.name, 2)
de.timeout = e.timeout
de.checkpoint = e.checkpoint
de.no_simbricks = e.no_simbricks
de.metadata = e.metadata.copy()
# create listening proxy on host 0
lp = proxy_listener_c()
lp.name = 'listener'
de.add_proxy(lp)
de.assign_sim_host(lp, 0)
# assign networks to first host
for net in e.networks:
de.add_network(net)
de.assign_sim_host(net, 0)
# create connecting proxy on host 1
cp = proxy_connecter_c(lp)
cp.name = 'connecter'
de.add_proxy(cp)
de.assign_sim_host(cp, 1)
# round-robin assignment for hosts
k = 0
for h in e.hosts:
de.add_host(h)
de.assign_sim_host(h, k)
for nic in h.nics:
de.assign_sim_host(nic, k)
if k != 0:
cp.add_nic(nic)
k = (k + 1) % 2
for nic in e.nics:
de.add_nic(nic)
return de
experiments/simbricks/orchestration/runtime/local.py deleted 100644 → 0
View file @ a14a0365
# Copyright 2021 Max Planck Institute for Software Systems, and
# National University of Singapore
#
# Permission is hereby granted, free of charge, to any person obtaining
# a copy of this software and associated documentation files (the
# "Software"), to deal in the Software without restriction, including
# without limitation the rights to use, copy, modify, merge, publish,
# distribute, sublicense, and/or sell copies of the Software, and to
# permit persons to whom the Software is furnished to do so, subject to
# the following conditions:
#
# The above copyright notice and this permission notice shall be
# included in all copies or substantial portions of the Software.
#
# THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND,
# EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF
# MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT.
# IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY
# CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN AN ACTION OF CONTRACT,
# TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION WITH THE
# SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.
import asyncio
import typing as tp
from simbricks.orchestration import exectools
from simbricks.orchestration.runners import ExperimentSimpleRunner
from simbricks.orchestration.runtime.common import Run, Runtime
class LocalSimpleRuntime(Runtime):
"""Execute runs locally in sequence."""
def __init__(
self,
verbose=False,
executor: exectools.Executor = exectools.LocalExecutor()
):
super().__init__()
self.runnable: tp.List[Run] = []
self.complete: tp.List[Run] = []
self.verbose = verbose
self.executor = executor
self._running: tp.Optional[asyncio.Task] = None
def add_run(self, run: Run) -> None:
self.runnable.append(run)
async def do_run(self, run: Run) -> None:
"""Actually executes `run`."""
try:
runner = ExperimentSimpleRunner(
self.executor, run.experiment, run.env, self.verbose
)
if self.profile_int:
runner.profile_int = self.profile_int
await run.prep_dirs(self.executor)
await runner.prepare()
except asyncio.CancelledError:
# it is safe to just exit here because we are not running any
# simulators yet
return
run.output = await runner.run() # handles CancelledError
self.complete.append(run)
# if the log is huge, this step takes some time
if self.verbose:
print(
f'Writing collected output of run {run.name()} to JSON file ...'
)
run.output.dump(run.outpath)
async def start(self) -> None:
"""Execute the runs defined in `self.runnable`."""
for run in self.runnable:
if self._interrupted:
return
self._running = asyncio.create_task(self.do_run(run))
await self._running
def interrupt_handler(self) -> None:
if self._running:
self._running.cancel()
# class LocalParallelRuntime(Runtime):
# """Execute runs locally in parallel on multiple cores."""
# def __init__(
# self,
# cores: int,
# mem: tp.Optional[int] = None,
# verbose=False,
# executor: exectools.Executor = exectools.LocalExecutor()
# ):
# super().__init__()
# self.runs_noprereq: tp.List[Run] = []
# """Runs with no prerequesite runs."""
# self.runs_prereq: tp.List[Run] = []
# """Runs with prerequesite runs."""
# self.complete: tp.Set[Run] = set()
# self.cores = cores
# self.mem = mem
# self.verbose = verbose
# self.executor = executor
# self._pending_jobs: tp.Set[asyncio.Task] = set()
# self._starter_task: asyncio.Task
# def add_run(self, run: Run) -> None:
# if run.experiment.resreq_cores() > self.cores:
# raise RuntimeError('Not enough cores available for run')
# if self.mem is not None and run.experiment.resreq_mem() > self.mem:
# raise RuntimeError('Not enough memory available for run')
# if run.prereq is None:
# self.runs_noprereq.append(run)
# else:
# self.runs_prereq.append(run)
# async def do_run(self, run: Run) -> tp.Optional[Run]:
# """Actually executes `run`."""
# try:
# runner = ExperimentSimpleRunner(
# self.executor, run.experiment, run.env, self.verbose
# )
# if self.profile_int:
# runner.profile_int = self.profile_int
# await run.prep_dirs(executor=self.executor)
# await runner.prepare()
# except asyncio.CancelledError:
# # it is safe to just exit here because we are not running any
# # simulators yet
# return None
# print('starting run ', run.name())
# run.output = await runner.run() # already handles CancelledError
# # if the log is huge, this step takes some time
# if self.verbose:
# print(
# f'Writing collected output of run {run.name()} to JSON file ...'
# )
# run.output.dump(run.outpath)
# print('finished run ', run.name())
# return run
# async def wait_completion(self) -> None:
# """Wait for any run to terminate and return."""
# assert self._pending_jobs
# done, self._pending_jobs = await asyncio.wait(
# self._pending_jobs, return_when=asyncio.FIRST_COMPLETED
# )
# for run in done:
# run = await run
# self.complete.add(run)
# self.cores_used -= run.experiment.resreq_cores()
# self.mem_used -= run.experiment.resreq_mem()
# def enough_resources(self, run: Run) -> bool:
# """Check if enough cores and mem are available for the run."""
# exp = run.experiment # pylint: disable=redefined-outer-name
# if self.cores is not None:
# enough_cores = (self.cores - self.cores_used) >= exp.resreq_cores()
# else:
# enough_cores = True
# if self.mem is not None:
# enough_mem = (self.mem - self.mem_used) >= exp.resreq_mem()
# else:
# enough_mem = True
# return enough_cores and enough_mem
# def prereq_ready(self, run: Run) -> bool:
# """Check if the prerequesite run for `run` has completed."""
# if run.prereq is None:
# return True
# return run.prereq in self.complete
# async def do_start(self) -> None:
# """Asynchronously execute the runs defined in `self.runs_noprereq +
# self.runs_prereq."""
# #self.completions = asyncio.Queue()
# self.cores_used = 0
# self.mem_used = 0
# runs = self.runs_noprereq + self.runs_prereq
# for run in runs:
# # if necessary, wait for enough memory or cores
# while not self.enough_resources(run):
# print('waiting for resources')
# await self.wait_completion()
# # if necessary, wait for prerequesite runs to complete
# while not self.prereq_ready(run):
# print('waiting for prereq')
# await self.wait_completion()
# self.cores_used += run.experiment.resreq_cores()
# self.mem_used += run.experiment.resreq_mem()
# job = asyncio.create_task(self.do_run(run))
# self._pending_jobs.add(job)
# # wait for all runs to finish
# await asyncio.gather(*self._pending_jobs)
# async def start(self) -> None:
# """Execute all defined runs."""
# self._starter_task = asyncio.create_task(self.do_start())
# try:
# await self._starter_task
# except asyncio.CancelledError:
# for job in self._pending_jobs:
# job.cancel()
# # wait for all runs to finish
# await asyncio.gather(*self._pending_jobs)
# def interrupt_handler(self) -> None:
# self._starter_task.cancel()
experiments/simbricks/orchestration/runtime/slurm.py deleted 100644 → 0
View file @ a14a0365
# Copyright 2021 Max Planck Institute for Software Systems, and
# National University of Singapore
#
# Permission is hereby granted, free of charge, to any person obtaining
# a copy of this software and associated documentation files (the
# "Software"), to deal in the Software without restriction, including
# without limitation the rights to use, copy, modify, merge, publish,
# distribute, sublicense, and/or sell copies of the Software, and to
# permit persons to whom the Software is furnished to do so, subject to
# the following conditions:
#
# The above copyright notice and this permission notice shall be
# included in all copies or substantial portions of the Software.
#
# THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND,
# EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF
# MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT.
# IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY
# CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN AN ACTION OF CONTRACT,
# TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION WITH THE
# SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.
import asyncio
import os
import pathlib
import pickle
import re
import typing as tp
from simbricks.orchestration.runtime.common import Run, Runtime
class SlurmRuntime(Runtime):
def __init__(self, slurmdir, args, verbose=False, cleanup=True) -> None:
super().__init__()
self.runnable: tp.List[Run] = []
self.slurmdir = slurmdir
self.args = args
self.verbose = verbose
self.cleanup = cleanup
self._start_task: asyncio.Task
def add_run(self, run: Run) -> None:
self.runnable.append(run)
def prep_run(self, run: Run) -> str:
exp = run.experiment
e_idx = exp.name + f'-{run.index}' + '.exp'
exp_path = os.path.join(self.slurmdir, e_idx)
log_idx = exp.name + f'-{run.index}' + '.log'
exp_log = os.path.join(self.slurmdir, log_idx)
sc_idx = exp.name + f'-{run.index}' + '.sh'
exp_script = os.path.join(self.slurmdir, sc_idx)
print(exp_path)
print(exp_log)
print(exp_script)
# write out pickled run
with open(exp_path, 'wb', encoding='utf-8') as f:
run.prereq = None # we don't want to pull in the prereq too
pickle.dump(run, f)
# create slurm batch script
with open(exp_script, 'w', encoding='utf-8') as f:
f.write('#!/bin/sh\n')
f.write(f'#SBATCH -o {exp_log} -e {exp_log}\n')
#f.write('#SBATCH -c %d\n' % (exp.resreq_cores(),))
f.write(f'#SBATCH --mem={exp.resreq_mem()}M\n')
f.write(f'#SBATCH --job-name="{run.name()}"\n')
f.write('#SBATCH --exclude=spyder[01-05],spyder16\n')
f.write('#SBATCH -c 32\n')
f.write('#SBATCH --nodes=1\n')
if exp.timeout is not None:
h = int(exp.timeout / 3600)
m = int((exp.timeout % 3600) / 60)
s = int(exp.timeout % 60)
f.write(f'#SBATCH --time={h:02d}:{m:02d}:{s:02d}\n')
extra = ''
if self.verbose:
extra = '--verbose'
f.write(f'python3 run.py {extra} --pickled {exp_path}\n')
f.write('status=$?\n')
if self.cleanup:
f.write(f'rm -rf {run.env.workdir}\n')
f.write('exit $status\n')
return exp_script
async def _do_start(self) -> None:
pathlib.Path(self.slurmdir).mkdir(parents=True, exist_ok=True)
jid_re = re.compile(r'Submitted batch job ([0-9]+)')
for run in self.runnable:
if run.prereq is None:
dep_cmd = ''
else:
dep_cmd = '--dependency=afterok:' + str(run.prereq.job_id)
script = self.prep_run(run)
stream = os.popen(f'sbatch {dep_cmd} {script}')
output = stream.read()
result = stream.close()
if result is not None:
raise RuntimeError('running sbatch failed')
m = jid_re.search(output)
if m is None:
raise RuntimeError('cannot retrieve id of submitted job')
run.job_id = int(m.group(1))
async def start(self) -> None:
self._start_task = asyncio.create_task(self._do_start())
try:
await self._start_task
except asyncio.CancelledError:
# stop all runs that have already been scheduled
# (existing slurm job id)
job_ids = []
for run in self.runnable:
if run.job_id:
job_ids.append(str(run.job_id))
scancel_process = await asyncio.create_subprocess_shell(
f"scancel {' '.join(job_ids)}"
)
await scancel_process.wait()
def interrupt_handler(self) -> None:
self._start_task.cancel()
experiments/simbricks/orchestration/simulator_utils.py deleted 100644 → 0
View file @ a14a0365
# Copyright 2021 Max Planck Institute for Software Systems, and
# National University of Singapore
#
# Permission is hereby granted, free of charge, to any person obtaining
# a copy of this software and associated documentation files (the
# "Software"), to deal in the Software without restriction, including
# without limitation the rights to use, copy, modify, merge, publish,
# distribute, sublicense, and/or sell copies of the Software, and to
# permit persons to whom the Software is furnished to do so, subject to
# the following conditions:
#
# The above copyright notice and this permission notice shall be
# included in all copies or substantial portions of the Software.
#
# THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND,
# EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF
# MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT.
# IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY
# CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN AN ACTION OF CONTRACT,
# TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION WITH THE
# SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.
"""Provides helper functions for assembling multiple host simulators."""
import typing as tp
from simbricks.orchestration.experiments import Experiment
from simbricks.orchestration.nodeconfig import AppConfig, NodeConfig
from simbricks.orchestration.simulators import (
HostSim, I40eMultiNIC, NetSim, NICSim
)
def create_basic_hosts(
e: Experiment,
num: int,
name_prefix: str,
net: NetSim,
nic_class: tp.Type[NICSim],
host_class: tp.Type[HostSim],
nc_class: tp.Type[NodeConfig],
app_class: tp.Type[AppConfig],
ip_start: int = 1,
ip_prefix: int = 24
) -> tp.List[HostSim]:
"""
Creates and configures multiple hosts to be simulated using the given
parameters.
Args:
num: number of hosts to create
"""
hosts: tp.List[HostSim] = []
for i in range(0, num):
nic = nic_class()
#nic.name = '%s.%d' % (name_prefix, i)
nic.set_network(net)
node_config = nc_class()
node_config.prefix = ip_prefix
ip = ip_start + i
node_config.ip = f'10.0.{int(ip / 256)}.{ip % 256}'
node_config.app = app_class()
host = host_class(node_config)
host.name = f'{name_prefix}.{i}'
host.add_nic(nic)
e.add_nic(nic)
e.add_host(host)
hosts.append(host)
return hosts
def create_multinic_hosts(
e: Experiment,
num: int,
name_prefix: str,
net: NetSim,
host_class: tp.Type[HostSim],
nc_class: tp.Type[NodeConfig],
app_class: tp.Type[AppConfig],
ip_start: int = 1,
ip_prefix: int = 24
) -> tp.List[HostSim]:
"""
Creates and configures multiple hosts to be simulated using the given
parameters. These hosts use multiple NICs.
Args:
num: number of hosts to create
"""
hosts: tp.List[HostSim] = []
mn = I40eMultiNIC()
mn.name = name_prefix
e.add_nic(mn)
for i in range(0, num):
nic = mn.create_subnic()
#nic.name = '%s.%d' % (name_prefix, i)
nic.set_network(net)
node_config = nc_class()
node_config.prefix = ip_prefix
ip = ip_start + i
node_config.ip = f'10.0.{int(ip / 256)}.{ip % 256}'
node_config.app = app_class()
host = host_class(node_config)
host.name = f'{name_prefix}.{i}'
host.add_nic(nic)
e.add_host(host)
hosts.append(host)
return hosts
def create_dctcp_hosts(
e: Experiment,
num: int,
name_prefix: str,
net: NetSim,
nic_class: tp.Type[NICSim],
host_class: tp.Type[HostSim],
nc_class: tp.Type[NodeConfig],
app_class: tp.Type[AppConfig],
cpu_freq: str,
mtu: int,
ip_start: int = 1
) -> tp.List[HostSim]:
"""
Creates and configures multiple hosts to be simulated in a DCTCP experiment
using the given parameters.
Args:
num: number of hosts to create
cpu_freq: CPU frequency to simulate, e.g. '5GHz'
"""
hosts = []
for i in range(0, num):
nic = nic_class()
#nic.name = '%s.%d' % (name_prefix, i)
nic.set_network(net)
node_config = nc_class()
node_config.mtu = mtu
node_config.ip = f'192.168.64.{ip_start + i}'
node_config.app = app_class()
host = host_class(node_config)
host.name = f'{name_prefix}.{i}'
host.cpu_freq = cpu_freq
host.add_nic(nic)
e.add_nic(nic)
e.add_host(host)
hosts.append(host)
return hosts
def create_tcp_cong_hosts(
e: Experiment,
num: int,
name_prefix: str,
net: NetSim,
nic_class: tp.Type[NICSim],
host_class: tp.Union[tp.Type[HostSim], tp.Callable[[NodeConfig], HostSim]],
nc_class: tp.Type[NodeConfig],
app_class: tp.Type[AppConfig],
cpu_freq: str,
mtu: int,
congestion_control: str,
ip_start: int = 1
):
"""
Creates and configures multiple hosts to be simulated in a TCP congestion
control experiment using the given parameters.
Args:
num: number of hosts to create
cpu_freq: CPU frequency to simulate, e.g. '5GHz'
"""
hosts = []
for i in range(0, num):
nic = nic_class()
#nic.name = '%s.%d' % (name_prefix, i)
nic.set_network(net)
node_config = nc_class()
node_config.mtu = mtu
node_config.tcp_congestion_control = congestion_control
node_config.ip = f'192.168.64.{ip_start + i}'
node_config.app = app_class()
host = host_class(node_config)
host.name = f'{name_prefix}.{i}'
host.cpu_freq = cpu_freq
host.add_nic(nic)
e.add_nic(nic)
e.add_host(host)
hosts.append(host)
return hosts
experiments/simbricks/orchestration/simulators.py deleted 100644 → 0
View file @ a14a0365
# Copyright 2021 Max Planck Institute for Software Systems, and
# National University of Singapore
#
# Permission is hereby granted, free of charge, to any person obtaining
# a copy of this software and associated documentation files (the
# "Software"), to deal in the Software without restriction, including
# without limitation the rights to use, copy, modify, merge, publish,
# distribute, sublicense, and/or sell copies of the Software, and to
# permit persons to whom the Software is furnished to do so, subject to
# the following conditions:
#
# The above copyright notice and this permission notice shall be
# included in all copies or substantial portions of the Software.
#
# THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND,
# EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF
# MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT.
# IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY
# CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN AN ACTION OF CONTRACT,
# TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION WITH THE
# SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.
# Allow own class to be used as type for a method's argument
from __future__ import annotations
import math
import typing as tp
from simbricks.orchestration import e2e_components as e2e
from simbricks.orchestration.e2e_topologies import E2ETopology
from simbricks.orchestration.experiment.experiment_environment import ExpEnv
from simbricks.orchestration.nodeconfig import NodeConfig
class Simulator(object):
"""Base class for all simulators."""
def __init__(self) -> None:
self.extra_deps: tp.List[Simulator] = []
self.name = ''
def resreq_cores(self) -> int:
"""
Number of cores this simulator requires during execution.
This is used for scheduling multiple runs and experiments.
"""
return 1
def resreq_mem(self) -> int:
"""
Number of memory in MB this simulator requires during execution.
This is used for scheduling multiple runs and experiments.
"""
return 64
def full_name(self) -> str:
"""Full name of the simulator."""
return ''
# pylint: disable=unused-argument
def prep_cmds(self, env: ExpEnv) -> tp.List[str]:
"""Commands to prepare execution of this simulator."""
return []
# pylint: disable=unused-argument
def run_cmd(self, env: ExpEnv) -> tp.Optional[str]:
"""Command to execute this simulator."""
return None
def dependencies(self) -> tp.List[Simulator]:
"""Other simulators to execute before this one."""
return []
# Sockets to be cleaned up
# pylint: disable=unused-argument
def sockets_cleanup(self, env: ExpEnv) -> tp.List[str]:
return []
# sockets to wait for indicating the simulator is ready
# pylint: disable=unused-argument
def sockets_wait(self, env: ExpEnv) -> tp.List[str]:
return []
def start_delay(self) -> int:
return 5
def wait_terminate(self, env: ExpEnv) -> bool:
return False
class PCIDevSim(Simulator):
"""Base class for PCIe device simulators."""
def __init__(self) -> None:
super().__init__()
self.sync_mode = 0
"""
Synchronization mode.
0 is running unsynchronized, 1 synchronized. Depending on the concrete
simulator, there may be additional modes.
"""
self.start_tick = 0
"""
The timestamp at which to start the simulation.
This is useful when the simulator is only attached at a later point in
time and needs to synchronize with connected simulators. For example,
this could be used when taking checkpoints to only attach certain
simulators after the checkpoint has been taken.
"""
self.sync_period = 500
"""Period in nanoseconds of sending synchronization messages from this
device to connected components."""
self.pci_latency = 500
"""Latency in nanoseconds for sending messages to components connected
via PCI."""
def full_name(self) -> str:
return 'dev.' + self.name
def is_nic(self) -> bool:
return False
def sockets_cleanup(self, env: ExpEnv) -> tp.List[str]:
return [env.dev_pci_path(self), env.dev_shm_path(self)]
def sockets_wait(self, env: ExpEnv) -> tp.List[str]:
return [env.dev_pci_path(self)]
class NICSim(PCIDevSim):
"""Base class for NIC simulators."""
def __init__(self) -> None:
super().__init__()
self.network: tp.Optional[NetSim] = None
self.mac: tp.Optional[str] = None
self.eth_latency = 500
"""Ethernet latency in nanoseconds from this NIC to the network
component."""
def set_network(self, net: NetSim) -> None:
"""Connect this NIC to a network simulator."""
self.network = net
net.connect_nic(self)
def basic_args(self, env: ExpEnv, extra: tp.Optional[str] = None) -> str:
cmd = (
f'{env.dev_pci_path(self)} {env.nic_eth_path(self)}'
f' {env.dev_shm_path(self)} {self.sync_mode} {self.start_tick}'
f' {self.sync_period} {self.pci_latency} {self.eth_latency}'
)
if self.mac is not None:
cmd += ' ' + (''.join(reversed(self.mac.split(':'))))
if extra is not None:
cmd += ' ' + extra
return cmd
def basic_run_cmd(
self, env: ExpEnv, name: str, extra: tp.Optional[str] = None
) -> str:
cmd = f'{env.repodir}/sims/nic/{name} {self.basic_args(env, extra)}'
return cmd
def full_name(self) -> str:
return 'nic.' + self.name
def is_nic(self) -> bool:
return True
def sockets_cleanup(self, env: ExpEnv) -> tp.List[str]:
return super().sockets_cleanup(env) + [env.nic_eth_path(self)]
def sockets_wait(self, env: ExpEnv) -> tp.List[str]:
return super().sockets_wait(env) + [env.nic_eth_path(self)]
class NetSim(Simulator):
"""Base class for network simulators."""
def __init__(self) -> None:
super().__init__()
self.opt = ''
self.sync_mode = 0
"""
Synchronization mode.
0 is running unsynchronized, 1 synchronized. Depending on the concrete
simulator, there may be additional modes.
"""
self.sync_period = 500
"""Synchronization period in nanoseconds from this network to connected
components."""
self.eth_latency = 500
"""Ethernet latency in nanoseconds from this network to connected
components."""
self.nics: tp.List[NICSim] = []
self.hosts_direct: tp.List[HostSim] = []
self.net_listen: tp.List[tp.Tuple[NetSim, str]] = []
self.net_connect: tp.List[tp.Tuple[NetSim, str]] = []
self.wait = False
def full_name(self) -> str:
return 'net.' + self.name
def connect_nic(self, nic: NICSim) -> None:
self.nics.append(nic)
def connect_network(self, net: NetSim, suffix='') -> None:
"""Connect this network to the listening peer `net`"""
net.net_listen.append((self, suffix))
self.net_connect.append((net, suffix))
def connect_sockets(self, env: ExpEnv) -> tp.List[tp.Tuple[Simulator, str]]:
sockets = []
for n in self.nics:
sockets.append((n, env.nic_eth_path(n)))
for (n, suffix) in self.net_connect:
sockets.append((n, env.n2n_eth_path(n, self, suffix)))
for h in self.hosts_direct:
sockets.append((h, env.net2host_eth_path(self, h)))
return sockets
def listen_sockets(self, env: ExpEnv) -> tp.List[tp.Tuple[NetSim, str]]:
listens = []
for (net, suffix) in self.net_listen:
listens.append((net, env.n2n_eth_path(self, net, suffix)))
return listens
def dependencies(self) -> tp.List[Simulator]:
return self.nics + self.hosts_direct + [n for n, _ in self.net_connect]
def sockets_cleanup(self, env: ExpEnv) -> tp.List[str]:
return [s for (_, s) in self.listen_sockets(env)]
def sockets_wait(self, env: ExpEnv) -> tp.List[str]:
return [s for (_, s) in self.listen_sockets(env)]
def wait_terminate(self, env: ExpEnv) -> bool:
return self.wait
def init_network(self) -> None:
pass
# FIXME: Class hierarchy is broken here as an ugly hack
class MemDevSim(NICSim):
"""Base class for memory device simulators."""
def __init__(self) -> None:
super().__init__()
self.mem_latency = 500
self.addr = 0xe000000000000000
self.size = 1024 * 1024 * 1024 # 1GB
self.as_id = 0
def full_name(self) -> str:
return 'mem.' + self.name
def sockets_cleanup(self, env: ExpEnv) -> tp.List[str]:
return [env.dev_mem_path(self), env.dev_shm_path(self)]
def sockets_wait(self, env: ExpEnv) -> tp.List[str]:
return [env.dev_mem_path(self)]
class NetMemSim(NICSim):
"""Base class for netork memory simulators."""
def __init__(self) -> None:
super().__init__()
self.addr = 0xe000000000000000
self.size = 1024 * 1024 * 1024 # 1GB
self.as_id = 0
def full_name(self) -> str:
return 'netmem.' + self.name
def sockets_cleanup(self, env: ExpEnv) -> tp.List[str]:
return [env.nic_eth_path(self), env.dev_shm_path(self)]
def sockets_wait(self, env: ExpEnv) -> tp.List[str]:
return [env.nic_eth_path(self)]
class HostSim(Simulator):
"""Base class for host simulators."""
def __init__(self, node_config: NodeConfig) -> None:
super().__init__()
self.node_config = node_config
"""System configuration for this simulated host."""
self.wait = False
"""
`True` - Wait for this simulator to finish execution. `False` - Don't
wait and instead shutdown the simulator as soon as all other awaited
simulators have completed execution.
"""
self.sleep = 0
self.cpu_freq = '4GHz'
self.sync_mode = 0
"""
Synchronization mode.
0 is running unsynchronized, 1 synchronized. Depending on the concrete
simulator, there may be additional modes.
"""
self.sync_period = 500
"""Period in nanoseconds of sending synchronization messages from this
device to connected components."""
self.pci_latency = 500
"""Latency in nanoseconds for sending messages to components connected
via PCIe."""
self.mem_latency = 500
"""Latency in nanoseconds for sending messages to components connected
via Ethernet."""
self.pcidevs: tp.List[PCIDevSim] = []
self.net_directs: tp.List[NetSim] = []
self.memdevs: tp.List[MemDevSim] = []
@property
def nics(self) -> tp.List[NICSim]:
return [
tp.cast(NICSim, pcidev)
for pcidev in self.pcidevs
if pcidev.is_nic()
]
def full_name(self) -> str:
return 'host.' + self.name
def add_nic(self, dev: NICSim) -> None:
"""Add a NIC to this host."""
self.add_pcidev(dev)
def add_pcidev(self, dev: PCIDevSim) -> None:
"""Add a PCIe device to this host."""
dev.name = self.name + '.' + dev.name
self.pcidevs.append(dev)
def add_memdev(self, dev: MemDevSim) -> None:
dev.name = self.name + '.' + dev.name
self.memdevs.append(dev)
def add_netdirect(self, net: NetSim) -> None:
"""Add a direct connection to a network to this host."""
net.hosts_direct.append(self)
self.net_directs.append(net)
def dependencies(self) -> tp.List[PCIDevSim]:
deps = []
for dev in self.pcidevs:
deps.append(dev)
if isinstance(dev, NICSim):
deps.append(dev.network)
for dev in self.memdevs:
deps.append(dev)
return deps
def wait_terminate(self, env: ExpEnv) -> bool:
return self.wait
class QemuHost(HostSim):
"""Qemu host simulator."""
def __init__(self, node_config: NodeConfig) -> None:
super().__init__(node_config)
self.sync = False
""""Whether to synchronize with attached simulators."""
def resreq_cores(self) -> int:
if self.sync:
return 1
else:
return self.node_config.cores + 1
def resreq_mem(self) -> int:
return 8192
def prep_cmds(self, env: ExpEnv) -> tp.List[str]:
return [
f'{env.qemu_img_path} create -f qcow2 -o '
f'backing_file="{env.hd_path(self.node_config.disk_image)}" '
f'{env.hdcopy_path(self)}'
]
def run_cmd(self, env: ExpEnv) -> str:
accel = ',accel=kvm:tcg' if not self.sync else ''
if self.node_config.kcmd_append:
kcmd_append = ' ' + self.node_config.kcmd_append
else:
kcmd_append = ''
cmd = (
f'{env.qemu_path} -machine q35{accel} -serial mon:stdio '
'-cpu Skylake-Server -display none -nic none '
f'-kernel {env.qemu_kernel_path} '
f'-drive file={env.hdcopy_path(self)},if=ide,index=0,media=disk '
f'-drive file={env.cfgtar_path(self)},if=ide,index=1,media=disk,'
'driver=raw '
'-append "earlyprintk=ttyS0 console=ttyS0 root=/dev/sda1 '
f'init=/home/ubuntu/guestinit.sh rw{kcmd_append}" '
f'-m {self.node_config.memory} -smp {self.node_config.cores} '
)
if self.sync:
unit = self.cpu_freq[-3:]
if unit.lower() == 'ghz':
base = 0
elif unit.lower() == 'mhz':
base = 3
else:
raise ValueError('cpu frequency specified in unsupported unit')
num = float(self.cpu_freq[:-3])
shift = base - int(math.ceil(math.log(num, 2)))
cmd += f' -icount shift={shift},sleep=off '
for dev in self.pcidevs:
cmd += f'-device simbricks-pci,socket={env.dev_pci_path(dev)}'
if self.sync:
cmd += ',sync=on'
cmd += f',pci-latency={self.pci_latency}'
cmd += f',sync-period={self.sync_period}'
else:
cmd += ',sync=off'
cmd += ' '
# qemu does not currently support net direct ports
assert len(self.net_directs) == 0
# qemu does not currently support mem device ports
assert len(self.memdevs) == 0
return cmd
class QemuIcountHost(QemuHost):
"""QEMU host simulator that uses instruction counting for
synchronization."""
def __init__(self, node_config: NodeConfig) -> None:
super().__init__(node_config)
self.sync = True
class Gem5Host(HostSim):
"""Gem5 host simulator."""
def __init__(self, node_config: NodeConfig) -> None:
node_config.sim = 'gem5'
super().__init__(node_config)
self.cpu_type_cp = 'X86KvmCPU'
self.cpu_type = 'TimingSimpleCPU'
self.sys_clock = '1GHz'
self.extra_main_args = []
self.extra_config_args = []
self.variant = 'fast'
def resreq_cores(self) -> int:
return 1
def resreq_mem(self) -> int:
return 4096
def prep_cmds(self, env: ExpEnv) -> tp.List[str]:
return [f'mkdir -p {env.gem5_cpdir(self)}']
def run_cmd(self, env: ExpEnv) -> str:
cpu_type = self.cpu_type
if env.create_cp:
cpu_type = self.cpu_type_cp
cmd = f'{env.gem5_path(self.variant)} --outdir={env.gem5_outdir(self)} '
cmd += ' '.join(self.extra_main_args)
cmd += (
f' {env.gem5_py_path} --caches --l2cache '
'--l1d_size=32kB --l1i_size=32kB --l2_size=32MB '
'--l1d_assoc=8 --l1i_assoc=8 --l2_assoc=16 '
f'--cacheline_size=64 --cpu-clock={self.cpu_freq}'
f' --sys-clock={self.sys_clock} '
f'--checkpoint-dir={env.gem5_cpdir(self)} '
f'--kernel={env.gem5_kernel_path} '
f'--disk-image={env.hd_raw_path(self.node_config.disk_image)} '
f'--disk-image={env.cfgtar_path(self)} '
f'--cpu-type={cpu_type} --mem-size={self.node_config.memory}MB '
f'--num-cpus={self.node_config.cores} '
'--mem-type=DDR4_2400_16x4 '
)
if self.node_config.kcmd_append:
cmd += f'--command-line-append="{self.node_config.kcmd_append}" '
if env.create_cp:
cmd += '--max-checkpoints=1 '
if env.restore_cp:
cmd += '-r 1 '
for dev in self.pcidevs:
cmd += (
f'--simbricks-pci=connect:{env.dev_pci_path(dev)}'
f':latency={self.pci_latency}ns'
f':sync_interval={self.sync_period}ns'
)
if cpu_type == 'TimingSimpleCPU':
cmd += ':sync'
cmd += ' '
for dev in self.memdevs:
cmd += (
f'--simbricks-mem={dev.size}@{dev.addr}@{dev.as_id}@'
f'connect:{env.dev_mem_path(dev)}'
f':latency={self.mem_latency}ns'
f':sync_interval={self.sync_period}ns'
)
if cpu_type == 'TimingSimpleCPU':
cmd += ':sync'
cmd += ' '
for net in self.net_directs:
cmd += (
'--simbricks-eth-e1000=listen'
f':{env.net2host_eth_path(net, self)}'
f':{env.net2host_shm_path(net, self)}'
f':latency={net.eth_latency}ns'
f':sync_interval={net.sync_period}ns'
)
if cpu_type == 'TimingSimpleCPU':
cmd += ':sync'
cmd += ' '
cmd += ' '.join(self.extra_config_args)
return cmd
def wait_terminate(self, env: ExpEnv) -> bool:
return True if env.create_cp else self.wait
class Gem5KvmHost(Gem5Host):
def __init__(self, node_config: NodeConfig) -> None:
super().__init__(node_config)
self.cpu_type = 'X86KvmCPU'
class SimicsHost(HostSim):
"""Simics host simulator."""
def __init__(self, node_config: NodeConfig) -> None:
super().__init__(node_config)
node_config.sim = 'simics'
self.cpu_class = 'x86-cooper-lake'
"""
Simics CPU class.
Can be obtained by running `list-classes substr = processor_` inside
Simics.
"""
self.cpu_freq = 4000 # TODO Don't hide attribute in super class
"""CPU frequency in MHz."""
self.timing = False
"""
Whether to run Simics in a more precise timing mode.
This adds a cache model.
"""
self.append_cmdline: tp.List[str] = []
"""Additional parameters to append on the command-line when invoking
Simics."""
self.interactive = False
"""
Whether to launch Simics in interactive GUI mode.
This is helpful for debugging, e.g. enabling log messages in the mid of
the simulation.
"""
self.debug_messages = False
"""Whether to enable debug messages of SimBricks adapter devices."""
def resreq_cores(self) -> int:
return 2
def resreq_mem(self) -> int:
return self.node_config.memory
def run_cmd(self, env: ExpEnv) -> str:
if self.node_config.kcmd_append:
raise RuntimeError(
'Appending kernel command-line not yet implemented.'
)
if self.interactive and not env.create_cp:
cmd = f'{env.simics_gui_path} -q '
else:
cmd = f'{env.simics_path} -q -batch-mode -werror '
if env.restore_cp:
# restore checkpoint
cmd += f'-e \'read-configuration {env.simics_cpfile(self)}\' '
else:
# initialize simulated machine
cmd += (
'-e \'run-command-file '
f'{env.simics_qsp_modern_core_path} '
f'disk0_image = {env.hd_raw_path(self.node_config.disk_image)} '
f'disk1_image = {env.cfgtar_path(self)} '
f'cpu_comp_class = {self.cpu_class} '
f'freq_mhz = {self.cpu_freq} '
f'num_cores = {self.node_config.cores} '
f'num_threads = {self.node_config.threads} '
f'memory_megs = {self.node_config.memory} '
'create_network = FALSE\' '
)
if env.create_cp:
# stop simulation when encountering special checkpoint string on
# serial console
cmd += (
'-e \'bp.console_string.break board.serconsole.con '
'"ready to checkpoint"\' '
)
# run simulation
cmd += '-e run '
# create checkpoint
cmd += f'-e \'write-configuration {env.simics_cpfile(self)}\' '
return cmd
if self.timing:
# Add the cache model. Note that the caches aren't warmed up during
# the boot process. The reason is that when later adding the memory
# devices, we change the mapped memory. The cycle staller doesn't
# like this and will SEGFAULT.
#
# The cache model doesn't store any memory contents and therefore
# doesn't answer any memory transactions. It only inserts CPU stall
# cycles on each cache level and can be queried for statistics as
# well as which addresses are cached.
#
# Read penalties are based on https://www.7-cpu.com/cpu/Skylake.html
cmd += (
'-e \'new-cycle-staller name = cs0 '
'stall-interval = 10000\' '
)
cmd += (
'-e \'new-simple-cache-tool name = cachetool '
'cycle-staller = cs0 -connect-all\' '
)
cmd += (
'-e \'cachetool.add-l1i-cache name = l1i line-size = 64 '
'sets = 64 ways = 8\' '
)
cmd += (
'-e \'cachetool.add-l1d-cache name = l1d line-size = 64 '
'sets = 64 ways = 8 -ip-read-prefetcher '
'prefetch-additional = 1 read-penalty = 4\' '
)
cmd += (
'-e \'cachetool.add-l2-cache name = l2 line-size = 64 '
'sets = 8192 ways = 4 -prefetch-adjacent '
'prefetch-additional = 4 read-penalty = 12\' '
)
cmd += (
'-e \'cachetool.add-l3-cache name = l3 line-size = 64 '
'sets = 32768 ways = 16 read-penalty = 42\' '
)
# Only simulate one cycle per CPU and then switch to the next. This is
# necessary for the synchronization of the SimBricks adapter with all
# the CPUs to work properly.
cmd += '-e \'set-time-quantum 1\' '
if self.memdevs:
cmd += '-e \'load-module simbricks_mem_comp\' '
for memdev in self.memdevs:
cmd += (
f'-e \'$mem = (new-simbricks-mem-comp '
f'socket = "{env.dev_mem_path(memdev)}" '
f'mem_latency = {self.mem_latency} '
f'sync_period = {self.sync_period})\' '
)
cmd += (
f'-e \'board.mb.dram_space.add-map $mem.simbricks_mem_dev '
f'{memdev.addr:#x} {memdev.size:#x}\' '
)
if self.debug_messages:
cmd += '-e \'$mem.log-level 3\' '
if self.pcidevs:
cmd += '-e \'load-module simbricks_pcie_comp\' '
i = 0
for pcidev in self.pcidevs:
cmd += (
f'-e \'$pci = (create-simbricks-pcie-comp '
f'socket = "{env.dev_pci_path(pcidev)}" '
f'pci_latency = {self.pci_latency} '
f'sync_period = {self.sync_period})\' '
)
cmd += f'-e \'connect board.mb.nb.pci_slot[{i}] $pci.pci_bus\' '
cmd += '-e instantiate-components '
if self.debug_messages:
cmd += '-e \'$pci.log-level 3\' '
i += 1
for param in self.append_cmdline:
cmd += f'{param} '
# The simulation keeps running when the host powers off. A log message
# indicates the event when the machine is powering off. We place a
# breakpoint on that log message, which will terminate Simics due to the
# use of `-batch-mode`.
cmd += (
'-e \'bp.log.break object=board.mb.sb.lpc.bank.acpi_io_regs '
'substr="Sleep state is unimplemented" type=unimpl\' '
)
return cmd + '-e run'
def wait_terminate(self, env: ExpEnv) -> bool:
return True if env.create_cp else self.wait
class CorundumVerilatorNIC(NICSim):
def __init__(self) -> None:
super().__init__()
self.clock_freq = 250 # MHz
def resreq_mem(self) -> int:
# this is a guess
return 512
def run_cmd(self, env: ExpEnv) -> str:
return self.basic_run_cmd(
env, '/corundum/corundum_verilator', str(self.clock_freq)
)
class CorundumBMNIC(NICSim):
def run_cmd(self, env: ExpEnv) -> str:
return self.basic_run_cmd(env, '/corundum_bm/corundum_bm')
class I40eNIC(NICSim):
def run_cmd(self, env: ExpEnv) -> str:
return self.basic_run_cmd(env, '/i40e_bm/i40e_bm')
class E1000NIC(NICSim):
def __init__(self) -> None:
super().__init__()
self.debug = False
def run_cmd(self, env: ExpEnv) -> str:
cmd = self.basic_run_cmd(env, '/e1000_gem5/e1000_gem5')
if self.debug:
cmd = 'env E1000_DEBUG=1 ' + cmd
return cmd
class MultiSubNIC(NICSim):
def __init__(self, mn: Simulator) -> None:
super().__init__()
self.multinic = mn
def full_name(self) -> str:
return self.multinic.full_name() + '.' + self.name
def dependencies(self) -> tp.List[Simulator]:
return super().dependencies() + [self.multinic]
def start_delay(self) -> int:
return 0
class I40eMultiNIC(Simulator):
def __init__(self) -> None:
super().__init__()
self.subnics: tp.List[NICSim] = []
def create_subnic(self) -> MultiSubNIC:
sn = MultiSubNIC(self)
self.subnics.append(sn)
return sn
def full_name(self) -> str:
return 'multinic.' + self.name
def run_cmd(self, env: ExpEnv) -> str:
args = ''
first = True
for sn in self.subnics:
if not first:
args += ' -- '
first = False
args += sn.basic_args(env)
return f'{env.repodir}/sims/nic/i40e_bm/i40e_bm {args}'
def sockets_cleanup(self, env: ExpEnv) -> tp.List[str]:
ss = []
for sn in self.subnics:
ss += sn.sockets_cleanup(env)
return ss
def sockets_wait(self, env: ExpEnv) -> tp.List[str]:
ss = []
for sn in self.subnics:
ss += sn.sockets_wait(env)
return ss
class WireNet(NetSim):
def run_cmd(self, env: ExpEnv) -> str:
connects = self.connect_sockets(env)
assert len(connects) == 2
cmd = (
f'{env.repodir}/sims/net/wire/net_wire {connects[0][1]}'
f' {connects[1][1]} {self.sync_mode} {self.sync_period}'
f' {self.eth_latency}'
)
if len(env.pcap_file) > 0:
cmd += ' ' + env.pcap_file
return cmd
class SwitchNet(NetSim):
def __init__(self) -> None:
super().__init__()
self.sync = True
"""Whether to synchronize with attached simulators."""
def run_cmd(self, env: ExpEnv) -> str:
cmd = env.repodir + '/sims/net/switch/net_switch'
cmd += f' -S {self.sync_period} -E {self.eth_latency}'
if not self.sync:
cmd += ' -u'
if len(env.pcap_file) > 0:
cmd += ' -p ' + env.pcap_file
for (_, n) in self.connect_sockets(env):
cmd += ' -s ' + n
for (_, n) in self.listen_sockets(env):
cmd += ' -h ' + n
return cmd
def sockets_cleanup(self, env: ExpEnv) -> tp.List[str]:
# cleanup here will just have listening eth sockets, switch also creates
# shm regions for each with a "-shm" suffix
cleanup = []
for s in super().sockets_cleanup(env):
cleanup.append(s)
cleanup.append(s + '-shm')
return cleanup
class MemSwitchNet(NetSim):
def __init__(self) -> None:
super().__init__()
self.sync = True
"""AS_ID,VADDR_START,VADDR_END,MEMNODE_MAC,PHYS_START."""
self.mem_map = []
def run_cmd(self, env: ExpEnv) -> str:
cmd = env.repodir + '/sims/mem/memswitch/memswitch'
cmd += f' -S {self.sync_period} -E {self.eth_latency}'
if not self.sync:
cmd += ' -u'
if len(env.pcap_file) > 0:
cmd += ' -p ' + env.pcap_file
for (_, n) in self.connect_sockets(env):
cmd += ' -s ' + n
for (_, n) in self.listen_sockets(env):
cmd += ' -h ' + n
for m in self.mem_map:
cmd += ' -m ' + f' {m[0]},{m[1]},{m[2]},'
cmd += (''.join(reversed(m[3].split(':'))))
cmd += f',{m[4]}'
return cmd
def sockets_cleanup(self, env: ExpEnv) -> tp.List[str]:
# cleanup here will just have listening eth sockets, switch also creates
# shm regions for each with a "-shm" suffix
cleanup = []
for s in super().sockets_cleanup(env):
cleanup.append(s)
cleanup.append(s + '-shm')
return cleanup
class TofinoNet(NetSim):
def __init__(self) -> None:
super().__init__()
self.tofino_log_path = '/tmp/model.ldjson'
self.sync = True
def run_cmd(self, env: ExpEnv) -> str:
cmd = f'{env.repodir}/sims/net/tofino/tofino'
cmd += (
f' -S {self.sync_period} -E {self.eth_latency}'
f' -t {self.tofino_log_path}'
)
if not self.sync:
cmd += ' -u'
for (_, n) in self.connect_sockets(env):
cmd += ' -s ' + n
return cmd
class BMV2Net(NetSim):
def __init__(self):
super().__init__()
self.sync = False
self.p4_path = 'sims/net/p4/basic.json'
def run_cmd(self, env):
cmd = (
f'{env.repodir}/sims/external/bmv2'
f'/targets/simple_switch/simple_switch'
)
cmd += (
f' --use-simbricks --sync-interval {self.sync_period}'
f' --link-latency {self.eth_latency}'
)
if self.sync:
cmd += ' --sync-eth'
port = 0
for _, n in self.connect_sockets(env):
cmd += f' -i {port}@{n}'
port += 1
if self.p4_path.startswith('/'):
cmd += f' {self.p4_path}'
else:
cmd += f' {env.repodir}/{self.p4_path}'
print(cmd)
return cmd
class NS3E2ENet(NetSim):
def __init__(self) -> None:
super().__init__()
self.first_run = True
self.e2e_components: tp.List[tp.Union[e2e.E2ETopologyNode,
e2e.E2ETopologyChannel]] = []
self.e2e_topologies: tp.List[E2ETopology] = []
self.e2e_global = e2e.E2EGlobalConfig()
self.e2e_ns3_logging = e2e.E2ENs3Logging()
self.use_file = True
def add_component(
self,
component: tp.Union[e2e.E2ETopologyNode,
e2e.E2ETopologyChannel,
E2ETopology]
):
if isinstance(component, E2ETopology):
self.e2e_topologies.append(component)
else:
self.e2e_components.append(component)
if isinstance(component, e2e.E2ETopologyNode):
component.network = self
def resolve_socket_paths(
self,
env: ExpEnv,
e2e_sim: tp.Union[e2e.E2ENetworkSimbricks, e2e.E2ESimbricksHost],
listen: bool = False
) -> None:
if e2e_sim.simbricks_component is None:
raise RuntimeError(
'E2E Simbricks adapter does not contain a simulator'
)
if e2e_sim.adapter_type == e2e.SimbricksAdapterType.NIC:
e2e_sim.unix_socket = env.nic_eth_path(e2e_sim.simbricks_component)
elif e2e_sim.adapter_type == e2e.SimbricksAdapterType.NETWORK:
if not isinstance(e2e_sim, e2e.E2ENetworkSimbricks):
raise RuntimeError(
f'Expected {e2e_sim.id} to be of type E2ENetworkSimbricks'
)
p_suf = ''
if e2e_sim.peer:
p_suf = min(e2e_sim.name, e2e_sim.peer.name)
if listen:
e2e_sim.unix_socket = env.n2n_eth_path(
self, e2e_sim.simbricks_component, p_suf
)
else:
e2e_sim.unix_socket = env.n2n_eth_path(
e2e_sim.simbricks_component, self, p_suf
)
elif e2e_sim.adapter_type == e2e.SimbricksAdapterType.HOST:
e2e_sim.unix_socket = env.net2host_eth_path(
self, e2e_sim.simbricks_component
)
def init_network(self) -> None:
# add all components from topologies to the network
for topo in self.e2e_topologies:
topo.add_to_network(self)
for component in self.e2e_components:
component.resolve_paths()
# add all connected networks
for c in component.components:
if isinstance(c, e2e.E2ENetworkSimbricks) and not c.listen:
p_suf = ''
if c.peer:
p_suf = min(c.name, c.peer.name)
self.connect_network(c.simbricks_component, p_suf)
elif isinstance(c, e2e.E2ESimbricksHost):
c.simbricks_component.set_network(self)
def run_cmd(self, env):
# resolve all socket paths
for component in self.e2e_components:
for c in component.components:
if isinstance(c, e2e.E2ESimbricksHost):
self.resolve_socket_paths(env, c)
elif isinstance(c, e2e.E2ENetworkSimbricks):
self.resolve_socket_paths(env, c, c.listen)
params: tp.List[str] = []
params.append(self.e2e_global.ns3_config())
params.append(self.e2e_ns3_logging.ns3_config())
for component in self.e2e_components:
params.append(component.ns3_config())
params_str = f'{" ".join(params)} {self.opt}'
if self.use_file:
file_path = env.ns3_e2e_params_file(self)
with open(file_path, 'w', encoding='utf-8') as f:
f.write(params_str)
cmd = (
f'{env.repodir}/sims/external/ns-3'
f'/simbricks-run.sh e2e-cc-example --ConfigFile={file_path}'
)
else:
cmd = (
f'{env.repodir}/sims/external/ns-3'
f'/simbricks-run.sh e2e-cc-example {params_str}'
)
print(cmd)
return cmd
class NS3DumbbellNet(NetSim):
def run_cmd(self, env: ExpEnv) -> str:
ports = ''
for (n, s) in self.connect_sockets(env):
if 'server' in n.name:
ports += f'--SimbricksPortLeft={s} '
else:
ports += f'--SimbricksPortRight={s} '
cmd = (
f'{env.repodir}/sims/external/ns-3'
f'/simbricks-run.sh simbricks-dumbbell-example {ports} {self.opt}'
)
print(cmd)
return cmd
class NS3BridgeNet(NetSim):
def run_cmd(self, env: ExpEnv) -> str:
ports = ''
for (_, n) in self.connect_sockets(env):
ports += '--SimbricksPort=' + n + ' '
cmd = (
f'{env.repodir}/sims/external/ns-3'
f'/simbricks-run.sh simbricks-bridge-example {ports} {self.opt}'
)
print(cmd)
return cmd
class NS3SequencerNet(NetSim):
def run_cmd(self, env: ExpEnv) -> str:
ports = ''
for (n, s) in self.connect_sockets(env):
if 'client' in n.name:
ports += '--ClientPort=' + s + ' '
elif 'replica' in n.name:
ports += '--ServerPort=' + s + ' '
elif 'sequencer' in n.name:
ports += '--ServerPort=' + s + ' '
else:
raise KeyError('Wrong NIC type')
cmd = (
f'{env.repodir}/sims/external/ns-3'
f'/simbricks-run.sh sequencer-single-switch-example'
f' {ports} {self.opt}'
)
return cmd
class FEMUDev(PCIDevSim):
def run_cmd(self, env: ExpEnv) -> str:
cmd = (
f'{env.repodir}/sims/external/femu/femu-simbricks'
f' {env.dev_pci_path(self)} {env.dev_shm_path(self)}'
)
return cmd
class BasicMemDev(MemDevSim):
def run_cmd(self, env: ExpEnv) -> str:
cmd = (
f'{env.repodir}/sims/mem/basicmem/basicmem'
f' {self.size} {self.addr} {self.as_id}'
f' {env.dev_mem_path(self)} {env.dev_shm_path(self)}'
f' {self.sync_mode} {self.start_tick} {self.sync_period}'
f' {self.mem_latency}'
)
return cmd
class MemNIC(MemDevSim):
def run_cmd(self, env: ExpEnv) -> str:
cmd = (
f'{env.repodir}/sims/mem/memnic/memnic'
f' {env.dev_mem_path(self)} {env.nic_eth_path(self)}'
f' {env.dev_shm_path(self)}'
)
if self.mac is not None:
cmd += ' ' + (''.join(reversed(self.mac.split(':'))))
cmd += f' {self.sync_mode} {self.start_tick} {self.sync_period}'
cmd += f' {self.mem_latency} {self.eth_latency}'
return cmd
def sockets_cleanup(self, env: ExpEnv) -> tp.List[str]:
return super().sockets_cleanup(env) + [env.nic_eth_path(self)]
def sockets_wait(self, env: ExpEnv) -> tp.List[str]:
return super().sockets_wait(env) + [env.nic_eth_path(self)]
class NetMem(NetMemSim):
def run_cmd(self, env: ExpEnv) -> str:
cmd = (
f'{env.repodir}/sims/mem/netmem/netmem'
f' {self.size} {self.addr} {self.as_id}'
f' {env.nic_eth_path(self)}'
f' {env.dev_shm_path(self)}'
)
if self.mac is not None:
cmd += ' ' + (''.join(reversed(self.mac.split(':'))))
cmd += f' {self.sync_mode} {self.start_tick} {self.sync_period}'
cmd += f' {self.eth_latency}'
return cmd
symphony/.gitignore 0 → 100644
View file @ 23867991
**__pycache__
auth.json
\ No newline at end of file
symphony/LICENSE.md 0 → 100644
View file @ 23867991
MIT License
Copyright (c) 2024 - present Max Planck Institute for Software Systems, National
University of Singapore, and SimBricks contributors.
Permission is hereby granted, free of charge, to any person obtaining a copy
of this software and associated documentation files (the "Software"), to deal
in the Software without restriction, including without limitation the rights
to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
copies of the Software, and to permit persons to whom the Software is
furnished to do so, subject to the following conditions:
The above copyright notice and this permission notice shall be included in all
copies or substantial portions of the Software.
THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
SOFTWARE.
symphony/cli/pyproject.toml 0 → 100644
View file @ 23867991
# Copyright 2024 Max Planck Institute for Software Systems, and
# National University of Singapore
#
# Permission is hereby granted, free of charge, to any person obtaining
# a copy of this software and associated documentation files (the
# "Software"), to deal in the Software without restriction, including
# without limitation the rights to use, copy, modify, merge, publish,
# distribute, sublicense, and/or sell copies of the Software, and to
# permit persons to whom the Software is furnished to do so, subject to
# the following conditions:
#
# The above copyright notice and this permission notice shall be
# included in all copies or substantial portions of the Software.
#
# THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND,
# EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF
# MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT.
# IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY
# CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN AN ACTION OF CONTRACT,
# TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION WITH THE
# SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.
[build-system]
requires = ["poetry-core>=1.0.0"]
build-backend = "poetry.core.masonry.api"
[tool.poetry]
name = "simbricks-cli"
version = "0.0.1"
description = "simbricks cli tool"
authors = [ "Jakob Görgen <jakob@simbricks.io>", ]
packages = [
{ include = "simbricks" }
]
[tool.poetry.dependencies]
typer = "0.13.1"
\ No newline at end of file
symphony/cli/requirements.txt 0 → 100644
View file @ 23867991
# Copyright 2024 Max Planck Institute for Software Systems, and
# National University of Singapore
#
# Permission is hereby granted, free of charge, to any person obtaining
# a copy of this software and associated documentation files (the
# "Software"), to deal in the Software without restriction, including
# without limitation the rights to use, copy, modify, merge, publish,
# distribute, sublicense, and/or sell copies of the Software, and to
# permit persons to whom the Software is furnished to do so, subject to
# the following conditions:
#
# The above copyright notice and this permission notice shall be
# included in all copies or substantial portions of the Software.
#
# THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND,
# EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF
# MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT.
# IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY
# CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN AN ACTION OF CONTRACT,
# TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION WITH THE
# SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.
# external dependencies
typer==0.13.1
# local dependencies
-e ../orchestration
experiments/simbricks/orchestration/utils/__init__.py → symphony/cli/simbricks/cli/__init__.py
View file @ 23867991
# Copyright 2022 Max Planck Institute for Software Systems, and # Copyright 2024 Max Planck Institute for Software Systems, and
# National University of Singapore # National University of Singapore
# #
# Permission is hereby granted, free of charge, to any person obtaining # Permission is hereby granted, free of charge, to any person obtaining
......
symphony/cli/simbricks/cli/__main__.py 0 → 100644
View file @ 23867991
# Copyright 2024 Max Planck Institute for Software Systems, and
# National University of Singapore
#
# Permission is hereby granted, free of charge, to any person obtaining
# a copy of this software and associated documentation files (the
# "Software"), to deal in the Software without restriction, including
# without limitation the rights to use, copy, modify, merge, publish,
# distribute, sublicense, and/or sell copies of the Software, and to
# permit persons to whom the Software is furnished to do so, subject to
# the following conditions:
#
# The above copyright notice and this permission notice shall be
# included in all copies or substantial portions of the Software.
#
# THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND,
# EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF
# MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT.
# IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY
# CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN AN ACTION OF CONTRACT,
# TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION WITH THE
# SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.
from typer import Typer, Option
from typing_extensions import Annotated
from simbricks.cli.commands import audit, namespaces, runs
from simbricks.cli import state
from simbricks.cli.utils import async_cli
app = Typer()
app.add_typer(namespaces.app, name="ns")
app.add_typer(runs.app, name="runs")
app.add_typer(audit.app, name="audit")
@app.callback()
@async_cli()
async def main(
ns: Annotated[str, Option(help='Namespace to operate in.')] = 'foo/bar/baz',
):
state.namespace = ns
if __name__ == "__main__":
app()
\ No newline at end of file
experiments/simbricks/orchestration/runtime/__init__.py → symphony/cli/simbricks/cli/commands/audit.py
View file @ 23867991
# Copyright 2021 Max Planck Institute for Software Systems, and # Copyright 2024 Max Planck Institute for Software Systems, and
# National University of Singapore # National University of Singapore
# #
# Permission is hereby granted, free of charge, to any person obtaining # Permission is hereby granted, free of charge, to any person obtaining
...@@ -20,11 +20,18 @@ ...@@ -20,11 +20,18 @@
# TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION WITH THE # TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION WITH THE
# SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE. # SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.
from simbricks.orchestration.runtime.common import Run, Runtime from pathlib import Path
from simbricks.orchestration.runtime.distributed import ( from typer import Typer, Option
DistributedSimpleRuntime, auto_dist from typing_extensions import Annotated
) from ..state import state
from simbricks.orchestration.runtime.local import ( from ..utils import async_cli
LocalParallelRuntime, LocalSimpleRuntime
app = Typer(
help="Auditing user actions."
) )
from simbricks.orchestration.runtime.slurm import SlurmRuntime
@app.command()
@async_cli()
async def audit():
"""List all user actions."""
pass
\ No newline at end of file
symphony/cli/simbricks/cli/commands/namespaces.py 0 → 100644
View file @ 23867991
# Copyright 2024 Max Planck Institute for Software Systems, and
# National University of Singapore
#
# Permission is hereby granted, free of charge, to any person obtaining
# a copy of this software and associated documentation files (the
# "Software"), to deal in the Software without restriction, including
# without limitation the rights to use, copy, modify, merge, publish,
# distribute, sublicense, and/or sell copies of the Software, and to
# permit persons to whom the Software is furnished to do so, subject to
# the following conditions:
#
# The above copyright notice and this permission notice shall be
# included in all copies or substantial portions of the Software.
#
# THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND,
# EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF
# MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT.
# IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY
# CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN AN ACTION OF CONTRACT,
# TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION WITH THE
# SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.
from pathlib import Path
from typer import Typer, Option
from typing_extensions import Annotated
from ..state import state
from ..utils import async_cli
app = Typer(
help="Managing SimBricks namespaces."
)
@app.command()
@async_cli()
async def ls():
"""List available namespaces."""
print(f"Listing Namespaces in {state.namespace}:")
@app.command()
@async_cli()
async def create(name: str):
"""Create a new namespace."""
print(f"Creating namespace {name} in {state.namespace}")
\ No newline at end of file
symphony/cli/simbricks/cli/commands/runs.py 0 → 100644
View file @ 23867991
# Copyright 2024 Max Planck Institute for Software Systems, and
# National University of Singapore
#
# Permission is hereby granted, free of charge, to any person obtaining
# a copy of this software and associated documentation files (the
# "Software"), to deal in the Software without restriction, including
# without limitation the rights to use, copy, modify, merge, publish,
# distribute, sublicense, and/or sell copies of the Software, and to
# permit persons to whom the Software is furnished to do so, subject to
# the following conditions:
#
# The above copyright notice and this permission notice shall be
# included in all copies or substantial portions of the Software.
#
# THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND,
# EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF
# MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT.
# IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY
# CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN AN ACTION OF CONTRACT,
# TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION WITH THE
# SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.
import asyncio
from pathlib import Path
import simbricks.utils
from typer import Typer, Argument, Option
from typing_extensions import Annotated
from ..state import state
from ..utils import async_cli
from rich.console import Console
from rich.table import Table
app = Typer(
help="Managing SimBricks runs."
)
@app.command()
@async_cli()
async def ls():
"""List runs."""
runs = await state.simbricks_client().get_runs()
table = Table()
table.add_column('Id')
table.add_column('Instantiation')
table.add_column('State')
for r in runs:
table.add_row(str(r['id']), str(r['instantiation_id']),
r['state'])
console = Console()
console.print(table)
@app.command()
@async_cli()
async def show(run_id: int):
"""Show individual run."""
run = await state.simbricks_client().get_run(run_id)
print(run)
async def follow_run(run_id: int):
last_run = None
while True:
run = await state.simbricks_client().get_run(run_id)
if not last_run or last_run['state'] != run['state']:
print(f'State:', run['state'])
if not last_run or (
len(last_run['output']) != len(run['output']) and
len(run['output']) != 0):
prev_len = len(last_run['output']) if last_run else 0
print(run['output'][prev_len:])
if run['state'] != 'pending' and run['state'] != 'running':
break
last_run = run
await asyncio.sleep(1)
@app.command()
@async_cli()
async def follow(run_id: int):
"""Follow individual run as it executes."""
await follow_run(run_id)
@app.command()
@async_cli()
async def submit_script(
path: Annotated[Path, Argument(help="Python simulation script to submit.")],
follow: Annotated[bool, Option("--follow", "-f",
help="Wait for run to terminate and show output live.")] = False,
):
"""Submit a SimBricks python simulation script to run."""
system_client = state.simbricks_client()
experiment_mod = simbricks.utils.load_module(module_path=path)
instantiations = experiment_mod.instantiations
sb_inst = instantiations[0]
sb_simulation = sb_inst.simulation
sb_system = sb_simulation.system
system = await system_client.create_system(sb_system)
system_id = int(system["id"])
system = await system_client.get_system(system_id)
print(system)
systems = await system_client.get_systems()
print(systems)
simulation = await system_client.create_simulation(system_id, sb_simulation)
sim_id = int(simulation["id"])
simulation = await system_client.get_simulation(sim_id)
print(simulation)
simulations = await system_client.get_simulations()
print(simulations)
instantiation = await system_client.create_instantiation(sim_id, None)
inst_id = int(instantiation["id"])
run = await system_client.create_run(inst_id)
print(run)
if follow:
await follow_run(run['id'])
\ No newline at end of file
symphony/cli/simbricks/cli/state.py 0 → 100644
View file @ 23867991
# Copyright 2024 Max Planck Institute for Software Systems, and
# National University of Singapore
#
# Permission is hereby granted, free of charge, to any person obtaining
# a copy of this software and associated documentation files (the
# "Software"), to deal in the Software without restriction, including
# without limitation the rights to use, copy, modify, merge, publish,
# distribute, sublicense, and/or sell copies of the Software, and to
# permit persons to whom the Software is furnished to do so, subject to
# the following conditions:
#
# The above copyright notice and this permission notice shall be
# included in all copies or substantial portions of the Software.
#
# THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND,
# EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF
# MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT.
# IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY
# CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN AN ACTION OF CONTRACT,
# TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION WITH THE
# SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.
import os
from simbricks.client import BaseClient, NSClient, SimBricksClient
class State():
def __init__(self):
self.namespace = ''
self.api_base = os.environ.get("SIMBRICKS_BASE_URL",
"http://0.0.0.0:8000")
self._base_client: BaseClient | None = None
self._ns_client: NSClient | None = None
self._simbricks_client: SimBricksClient | None = None
def base_client(self):
if self._base_client is None:
self._base_client = BaseClient(base_url=self.api_base)
return self._base_client
def ns_client(self):
if self._ns_client is None:
self._ns_client = NSClient(base_client=self.base_client(),
namespace=self.namespace)
return self._ns_client
def simbricks_client(self):
if self._simbricks_client is None:
self._simbricks_client = SimBricksClient(self.ns_client())
return self._simbricks_client
state = State()
\ No newline at end of file
symphony/cli/simbricks/cli/utils.py 0 → 100644
View file @ 23867991
# Copyright 2024 Max Planck Institute for Software Systems, and
# National University of Singapore
#
# Permission is hereby granted, free of charge, to any person obtaining
# a copy of this software and associated documentation files (the
# "Software"), to deal in the Software without restriction, including
# without limitation the rights to use, copy, modify, merge, publish,
# distribute, sublicense, and/or sell copies of the Software, and to
# permit persons to whom the Software is furnished to do so, subject to
# the following conditions:
#
# The above copyright notice and this permission notice shall be
# included in all copies or substantial portions of the Software.
#
# THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND,
# EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF
# MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT.
# IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY
# CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN AN ACTION OF CONTRACT,
# TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION WITH THE
# SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.
import asyncio
import functools
def async_cli():
"""
Decorator function turning async cli routines into regular ones for
typer.
"""
def decorator_async_cli(f):
@functools.wraps(f)
def wrapper(*args, **kwargs):
return asyncio.run(f(*args, **kwargs))
return wrapper
return decorator_async_cli
\ No newline at end of file
symphony/client/pyproject.toml 0 → 100644
View file @ 23867991
# Copyright 2024 Max Planck Institute for Software Systems, and
# National University of Singapore
#
# Permission is hereby granted, free of charge, to any person obtaining
# a copy of this software and associated documentation files (the
# "Software"), to deal in the Software without restriction, including
# without limitation the rights to use, copy, modify, merge, publish,
# distribute, sublicense, and/or sell copies of the Software, and to
# permit persons to whom the Software is furnished to do so, subject to
# the following conditions:
#
# The above copyright notice and this permission notice shall be
# included in all copies or substantial portions of the Software.
#
# THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND,
# EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF
# MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT.
# IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY
# CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN AN ACTION OF CONTRACT,
# TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION WITH THE
# SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.
[build-system]
requires = ["poetry-core>=1.0.0"]
build-backend = "poetry.core.masonry.api"
[tool.poetry]
name = "simbricks-client"
version = "0.0.1"
description = "simbricks client lib to send and retrieve experiments"
authors = [ "Jakob Görgen <jakob@simbricks.io>", ]
packages = [
{ include = "simbricks" }
]
[tool.poetry.dependencies]
aiohttp = "3.9.5"
\ No newline at end of file
symphony/client/requirements.txt 0 → 100644
View file @ 23867991
# Copyright 2024 Max Planck Institute for Software Systems, and
# National University of Singapore
#
# Permission is hereby granted, free of charge, to any person obtaining
# a copy of this software and associated documentation files (the
# "Software"), to deal in the Software without restriction, including
# without limitation the rights to use, copy, modify, merge, publish,
# distribute, sublicense, and/or sell copies of the Software, and to
# permit persons to whom the Software is furnished to do so, subject to
# the following conditions:
#
# The above copyright notice and this permission notice shall be
# included in all copies or substantial portions of the Software.
#
# THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND,
# EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF
# MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT.
# IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY
# CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN AN ACTION OF CONTRACT,
# TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION WITH THE
# SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.
# external dependencies
aiohttp==3.9.5
# local dependencies
-e ../orchestration
\ No newline at end of file
symphony/client/simbricks/client/__init__.py 0 → 100644
View file @ 23867991
# Copyright 2024 Max Planck Institute for Software Systems, and
# National University of Singapore
#
# Permission is hereby granted, free of charge, to any person obtaining
# a copy of this software and associated documentation files (the
# "Software"), to deal in the Software without restriction, including
# without limitation the rights to use, copy, modify, merge, publish,
# distribute, sublicense, and/or sell copies of the Software, and to
# permit persons to whom the Software is furnished to do so, subject to
# the following conditions:
#
# The above copyright notice and this permission notice shall be
# included in all copies or substantial portions of the Software.
#
# THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND,
# EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF
# MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT.
# IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY
# CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN AN ACTION OF CONTRACT,
# TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION WITH THE
# SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.
from .client import BaseClient, NSClient, SimBricksClient, RunnerClient
symphony/client/simbricks/client/auth.py 0 → 100644
View file @ 23867991
# Copyright 2024 Max Planck Institute for Software Systems, and
# National University of Singapore
#
# Permission is hereby granted, free of charge, to any person obtaining
# a copy of this software and associated documentation files (the
# "Software"), to deal in the Software without restriction, including
# without limitation the rights to use, copy, modify, merge, publish,
# distribute, sublicense, and/or sell copies of the Software, and to
# permit persons to whom the Software is furnished to do so, subject to
# the following conditions:
#
# The above copyright notice and this permission notice shall be
# included in all copies or substantial portions of the Software.
#
# THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND,
# EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF
# MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT.
# IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY
# CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN AN ACTION OF CONTRACT,
# TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION WITH THE
# SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.
import json
import aiohttp
import time
import os
class Token:
def __init__(
self,
access_token: str,
refresh_token: str,
session_state: str,
access_valid_until: int,
refresh_valid_until: int,
):
self.access_token: str = access_token
self.refresh_token: str = refresh_token
self.session_state: str = session_state
self.access_valid_until: int = access_valid_until
self.refresh_valid_until: int = refresh_valid_until
def toJSON(self) -> dict:
return self.__dict__
def is_access_valid(self) -> bool:
return self.access_valid_until > int(time.time())
def is_refresh_valid(self) -> bool:
return self.refresh_valid_until > int(time.time())
class TokenClient:
def __init__(
self,
device_auth_url: str = "https://auth.simbricks.io/realms/SimBricks/protocol/openid-connect/auth/device",
token_url: str = "https://auth.simbricks.io/realms/SimBricks/protocol/openid-connect/token",
client_id: str = "api.auth.simbricks.io",
):
self._device_auth_url: str = device_auth_url
self._token_url: str = token_url
self._client_id: str = client_id
def _create_token_from_resp(self, json_obj) -> Token:
access_valid_until = int(time.time()) - 10 + int(json_obj["expires_in"])
refresh_valid_until = (
int(time.time()) - 10 + int(json_obj["refresh_expires_in"])
)
return Token(
access_token=json_obj["access_token"],
refresh_token=json_obj["refresh_token"],
session_state=json_obj["session_state"],
access_valid_until=access_valid_until,
refresh_valid_until=refresh_valid_until,
)
async def retrieve_token(self) -> Token:
token = None
async with aiohttp.ClientSession() as session:
# get device_code, interval, verification_uri, user_code
device_code = None
interval = None
verification_uri = None
user_code = None
async with session.post(
url=self._device_auth_url, data={"client_id": self._client_id}
) as resp:
resp.raise_for_status() # TODO: handel gracefully
json_resp = await resp.json()
device_code = json_resp["device_code"]
interval = json_resp["interval"]
verification_uri = json_resp["verification_uri"]
user_code = json_resp["user_code"]
assert device_code and interval and verification_uri and user_code
# retrieve valid token upon successfull user authentication
print(f"Please visit {verification_uri} in the browser")
print(f"There, enter the code: {user_code}")
print("Waiting...")
while True:
time.sleep(interval) # TODO: check timeout...
async with session.post(
url=self._token_url,
data={
"client_id": self._client_id,
"grant_type": "urn:ietf:params:oauth:grant-type:device_code",
"device_code": device_code,
},
) as resp:
if resp.status >= 500:
raise Exception(
f"{resp.status} error while trying toi recieve token"
) # TODO: handel gracefully
json_resp = await resp.json()
if (
"error" in json_resp
and json_resp["error"] != "authorization_pending"
):
raise Exception(
f"error retrievening retrieving token: {json_resp}"
)
elif "error" not in json_resp:
token = self._create_token_from_resp(json_obj=json_resp)
break
assert token
return token
async def refresh_token(self, old_token: Token) -> Token:
assert old_token.is_refresh_valid()
token = None
async with aiohttp.ClientSession() as session:
# get device_code, interval, verification_uri, user_code
async with session.post(
url=self._token_url,
data={
"client_id": self._client_id,
"grant_type": "refresh_token",
"refresh_token": old_token.refresh_token,
},
) as resp:
resp.raise_for_status() # TODO: handel gracefully
json_resp = await resp.json()
if "error" in json_resp:
raise Exception(f"error refreshing token: {json_resp}")
token = self._create_token_from_resp(json_obj=json_resp)
assert token
return token
class TokenProvider:
def __init__(self) -> None:
self._toke_filepath: str = "auth.json"
self._token: Token | None = self._load_token()
self._toke_client = TokenClient()
def _load_token(self) -> Token | None:
if os.path.isfile(self._toke_filepath):
with open(self._toke_filepath) as fh:
json_token = json.load(fh)
return Token(
access_token=json_token["access_token"],
refresh_token=json_token["refresh_token"],
session_state=json_token["session_state"],
access_valid_until=int(json_token["access_valid_until"]),
refresh_valid_until=int(json_token["refresh_valid_until"]),
)
return None
def _store_token(self) -> None:
if self._token is None:
return
with open(self._toke_filepath, "w") as fh:
json.dump(self._token.toJSON(), fh)
def _access_valid(self) -> bool:
return self._token and self._token.is_access_valid()
def _refresh_valid(self) -> bool:
return self._token and self._token.is_refresh_valid()
async def _refresh_token(self) -> None:
if self._access_valid():
return
if self._refresh_valid():
assert self._token
self._token = await self._toke_client.refresh_token(old_token=self._token)
else:
self._token = await self._toke_client.retrieve_token()
assert self._token
self._store_token()
async def access_token(self) -> str:
await self._refresh_token()
assert self._token
return self._token.access_token
Markdown is supported
0% or .
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment