# Copyright (c) Max-Planck-Gesellschaft zur Förderung der Wissenschaften e.V.
# Developed at the Max Planck Institute for Informatics, Network and Cloud Systems Group
#
# Author: Yiming Lei (ylei@mpi-inf.mpg.de)
#
# This software is licensed for non-commercial scientific research purposes only.
#
# License text: Creative Commons NC BY SA 4.0
# https://creativecommons.org/licenses/by-nc-sa/4.0/deed.en
import os
import time
import warnings
import networkx as nx
import openoptics.utils as utils
from openoptics.backends import create_backend
from openoptics.dashboard import NullDashboard
from openoptics.DeviceManager import DeviceManager
from openoptics.OpticalCLI import OpticalCLI
from openoptics.TimeFlowTable import Path, TimeFlowEntry
from typing import List, Union
[docs]
class BaseNetwork:
"""
The base class of optical networks.
This class provides the foundation for creating and managing optical networks
in OpenOptics. It includes topology and routing configurations,
network monitoring, and interaction with the underlying backend.
The class handles both Traffic-Oblivious (TO) and Traffic-Aware (TA) architectures,
allowing for pre-defined topology or runtime reconfiguration.
"""
[docs]
def __init__(
self,
name,
nb_node,
nb_link=1,
nb_host_per_tor=1,
backend="Mininet",
time_slice_duration_us=None,
time_slice_duration_ms=None,
guardband_us=None,
guardband_ms=None,
arch_mode="TO", # TO for traffic-oblivious, TA for traffic-aware
use_webserver=True,
ocs_tor_link_bw_gbps: float = 1.0,
tor_host_link_bw_gbps: float = 1.0,
**backend_kwargs,
):
"""
Initialize the BaseNetwork.
Args:
name (str): Name of the network
backend (str): Backend of OpenOptics. "Mininet", "Testbed", or "ns3"
nb_node (int): Number of nodes in the network
nb_link (int, optional): Number of links per node (defaults to 1)
nb_host_per_tor (int, optional): Number of hosts per ToR (defaults to 1)
time_slice_duration_us (int, optional): Duration of each time slice in microseconds
time_slice_duration_ms (int, optional): Duration of each time slice in milliseconds
Specify at most one of time_slice_duration_us or time_slice_duration_ms (default 128 ms).
guardband_us (int, optional): Guard band in microseconds
guardband_ms (int, optional): Guard band in milliseconds
Specify at most one of guardband_us or guardband_ms (default 25 ms).
arch_mode (str, optional): architecture mode: "TO" or "TA" (defaults to "TO")
use_webserver (bool, optional): Whether to use web server for dashboard (defaults to True)
ocs_tor_link_bw_gbps (float, optional): Bandwidth in Gbps for OCS↔ToR links (defaults to 1.0)
tor_host_link_bw_gbps (float, optional): Bandwidth in Gbps for ToR↔host links (defaults to 1.0)
**backend_kwargs: Backend-specific parameters (e.g. ``link_delay_ms`` for
Mininet/ns-3). Validated immediately against the selected backend's
``accepted_kwargs()``; unknown names raise ``ValueError``.
"""
self.host_tor_port = 0
assert nb_link > 0
self.tor_host_port = nb_link
self.tor_ocs_ports = list(range(nb_link))
self.name = name
self.nb_time_slices = 1
# Time slice and guardband: accept either µs or ms, store canonical µs
# only. Backends receive the µs form via setup() and convert if their
# underlying runtime is ms-only (e.g. Mininet's BMv2 CLI).
if time_slice_duration_us is not None and time_slice_duration_ms is not None:
raise ValueError("Specify only one of time_slice_duration_us or time_slice_duration_ms.")
if time_slice_duration_us is not None:
self.time_slice_duration_us = int(time_slice_duration_us)
elif time_slice_duration_ms is not None:
self.time_slice_duration_us = int(time_slice_duration_ms * 1000)
else:
self.time_slice_duration_us = 128_000 # default: 128 ms
if guardband_us is not None and guardband_ms is not None:
raise ValueError("Specify only one of guardband_us or guardband_ms.")
if guardband_us is not None:
self.guardband_us = int(guardband_us)
elif guardband_ms is not None:
self.guardband_us = int(guardband_ms * 1000)
else:
self.guardband_us = 25_000 # default: 25 ms
self.arch_mode = arch_mode
self.calendar_queue_mode = 0 if arch_mode == "TO" else 1
self.slice_to_topo = {}
self.nb_node = nb_node
self.nb_link = nb_link
self.nb_host_per_tor = nb_host_per_tor
self.nodes_created = False
self.use_webserver = use_webserver
self.ocs_tor_link_bw_gbps = float(ocs_tor_link_bw_gbps)
self.tor_host_link_bw_gbps = float(tor_host_link_bw_gbps)
# Always present so callers can invoke dashboard methods unconditionally;
# start_monitor() replaces this with a real DashboardService when
# use_webserver=True.
self.dashboard = NullDashboard()
self._backend = create_backend(backend)
accepted = type(self._backend).accepted_kwargs()
unknown = set(backend_kwargs) - accepted
if unknown:
raise ValueError(
f"Backend '{backend}' does not accept: {unknown}. "
f"Accepted backend-specific kwargs: {accepted or 'none'}"
)
self._backend_kwargs = backend_kwargs
print("Setting up OpenOptics...")
def __str__(self) -> str:
"""
Return string representation of the network.
Returns:
str: Name of the network
"""
return self.name
def start_monitor(self):
"""
Start OpenOptics DeviceManager and Dashboard.
Initialises the monitoring system and starts the web dashboard when
``use_webserver`` is enabled and the selected backend can feed it
metrics. The dashboard runs in-process on the host and port configured
by :class:`DashboardConfig` (default ``localhost:8001``).
"""
dashboard_enabled = self.use_webserver and (
self._backend.supports_device_manager
or getattr(self._backend, "supports_dashboard_without_device_manager", False)
)
if dashboard_enabled:
from openoptics.dashboard.collectors import ReconfigEventPublisher
reconfig_publisher = ReconfigEventPublisher()
else:
reconfig_publisher = None
if self._backend.supports_device_manager:
self.device_manager = DeviceManager(
self._backend,
self.tor_ocs_ports,
nb_queue=self.nb_time_slices if self.arch_mode == "TO" else self.nb_node,
event_publisher=reconfig_publisher,
)
else:
self.device_manager = None
if dashboard_enabled:
from openoptics.dashboard import DashboardConfig, DashboardService
self.dashboard = DashboardService(DashboardConfig.from_env())
self.dashboard.begin_epoch()
self.dashboard.update_topology(self.slice_to_topo)
# Mininet-style polling collector (needs a live DeviceManager).
if self.device_manager is not None:
from openoptics.dashboard.collectors import DeviceMetricCollector
self.dashboard.register_collector(DeviceMetricCollector(
self.device_manager,
nb_port=self.nb_link,
nb_queue=(
self.nb_time_slices
if self.calendar_queue_mode == 0
else self.nb_node
),
interval_s=self.dashboard.config.poll_interval_s,
))
# Backend-specific wiring: simulator backends register their own
# event-source sinks and connect ns-3 TraceSources. Default is a
# no-op; Ns3Backend overrides.
self._backend.setup_dashboard(self.dashboard)
self.dashboard.register_event_source(reconfig_publisher)
self.dashboard.start()
def start_cli(self):
"""
Start OpenOptics CLI.
Launches the command-line interface for interacting with the network.
"""
OpticalCLI(self)
def stop_network(self):
"""
Stop the network.
Stops the dashboard (if running) and the backend network.
"""
self.dashboard.stop()
self._backend.stop()
def create_nodes(self):
"""Create nodes via the configured backend."""
self._backend.setup(
nb_node=self.nb_node,
nb_host_per_tor=self.nb_host_per_tor,
nb_link=self.nb_link,
nb_time_slices=self.nb_time_slices,
time_slice_duration_us=self.time_slice_duration_us,
guardband_us=self.guardband_us,
calendar_queue_mode=self.calendar_queue_mode,
ocs_tor_link_bw_gbps=self.ocs_tor_link_bw_gbps,
tor_host_link_bw_gbps=self.tor_host_link_bw_gbps,
**self._backend_kwargs,
)
def udp_traffic(self, **defaults):
"""Return a UDP traffic builder when the selected backend supports it."""
traffic_builder = getattr(self._backend, "udp_traffic", None)
if not callable(traffic_builder):
raise NotImplementedError(
"udp_traffic() is currently supported only by the ns-3 backend"
)
return traffic_builder(**defaults)
def tcp_traffic(self, **defaults):
"""Return a TCP traffic builder when the selected backend supports it."""
traffic_builder = getattr(self._backend, "tcp_traffic", None)
if not callable(traffic_builder):
raise NotImplementedError(
"tcp_traffic() is currently supported only by the ns-3 backend"
)
return traffic_builder(**defaults)
def cal_node_port_to_ocs_port(self, node_id, port_id):
"""
Find the OCS's port that connects to a node's port.
Args:
node_id (int): ID of the ToR
port_id (int): ID of the ToR's port
Returns:
int: the OCS's port
"""
return port_id * self.nb_node + node_id
def setup_ocs(self):
"""
Generate commands for OCS forwarding.
Creates and loads the OCS forwarding table entries based on
the current topology configuration.
"""
ocs_slice_port1_port2 = []
for ts, graph in self.slice_to_topo.items():
for node1, node2, attr in nx.to_edgelist(graph):
port1, port2 = attr["port1"], attr["port2"]
ocs_port1 = self.cal_node_port_to_ocs_port(node1, port1)
ocs_port2 = self.cal_node_port_to_ocs_port(node2, port2)
ocs_slice_port1_port2.append((ts, ocs_port1, ocs_port2))
ocs_entries = utils.gen_ocs_commands(ocs_slice_port1_port2)
self._backend.load_table(
switch_name="ocs",
entries=ocs_entries,
print_flag=False,
)
def setup_nodes(self):
"""
Load utility tables into nodes.
Configures the ToR switches with necessary routing and forwarding tables
including IP to destination mappings, arrival verification, and port calculation.
"""
print("Setting up switch tables...")
ip_to_tor = self._backend.get_ip_to_tor()
ip_to_dst_entries = utils.tor_table_ip_to_dst(ip_to_tor)
for tor_id in range(self.nb_node):
arrive_at_dst = utils.tor_table_arrive_at_dst(tor_id, self.tor_host_port)
verify_desired_node = utils.tor_table_verify_desired_node(tor_id)
cal_port_enqueue = utils.tor_table_cal_port_slice_to_node(
tor_id, self.slice_to_topo
)
self._backend.load_table(
switch_name=f"tor{tor_id}",
entries=ip_to_dst_entries
+ arrive_at_dst
+ verify_desired_node
+ cal_port_enqueue,
print_flag=False,
save_flag=False,
)
[docs]
def start(self):
"""
Start OpenOptics user interface (CLI, dashboard, ...).
Traffic Oblivious. Initializes DeviceManager, starts CLI, and handles
network shutdown.
Simulator-style backends (``supports_cli == False``) skip the
interactive CLI and drive the backend's own ``run()`` instead; this
path is taken by the ns-3 backend.
"""
self.start_monitor()
if getattr(self._backend, "supports_cli", True):
self.start_cli()
else:
self._backend.run()
self.stop_network()
[docs]
def start_traffic_aware(
self, topo_func=None, routing_func=None, routing_mode=None, update_interval=1
) -> bool:
"""Deploy traffic aware architecture.
Args:
topo_func: traffic aware topology function with traffic matrix as input
routing_func: routing function used by traffic aware architecture
routing_mode: Source or Per-hop
update_interval: interval in seconds to update topology and routing
Return:
Whether the traffic aware architecture is successfully deployed.
"""
if self.nb_link != 1:
raise ValueError("Traffic aware architecture only supports one link.")
import threading
stop_event = threading.Event()
self.start_monitor()
self.activate_calendar_queue()
def evolve():
prev_circuits = []
while not stop_event.is_set():
metric = self.device_manager.get_device_metric()
traffic_matrix = utils.metric_to_matrix(metric)
if topo_func:
circuits = topo_func(
nb_node=self.nb_node,
nb_link=self.nb_link,
traffic_matrix=traffic_matrix,
prev_circuits=prev_circuits,
)
if prev_circuits != circuits:
prev_circuits = circuits
self.pause_calendar_queue()
assert self.deploy_topo(circuits, start_fresh=True)
self.activate_calendar_queue()
stop_event.wait(timeout=update_interval)
evolve_thread = threading.Thread(target=evolve)
evolve_thread.start()
self.start_cli()
stop_event.set()
evolve_thread.join()
self.stop_network()
##########################
# Optical Topology #
##########################
[docs]
def connect(
self, time_slice, node1, node2, port1=0, port2=0, unidirectional=False
) -> bool:
"""
Connect two node ports at the given time slice by configuring OCS,
if no optical connections haven't been built for both ports.
Args:
time_slice (int): The time slice to connect
node1 (int): The first of two nodes to connect
node2 (int): The second of two nodes to connect
port1 (int, optional): The port the first node uses to connect. Defaults to 0.
port2 (int, optional): The port the second node uses to connect. Defaults to 0.
unidirectional (bool, optional): Whether connection is unidirectional. Defaults to False.
Returns:
bool: Whether the connect() was successful.
"""
if not isinstance(time_slice, int) or time_slice < 0:
raise ValueError(f"Invalid time slice {time_slice}.")
if not isinstance(node1, int) or node1 < 0 or node1 >= self.nb_node:
raise ValueError(
f"Invalid node {node1}. Only nodes 0 to {self.nb_node - 1} are valid. Are you setting the correct nb_node when generating topology?"
)
if not isinstance(node2, int) or node2 < 0 or node2 >= self.nb_node:
raise ValueError(
f"Invalid node {node2}. Only nodes 0 to {self.nb_node - 1} are valid. Are you setting the correct nb_node when generating topology?"
)
if time_slice not in self.slice_to_topo.keys():
for added_time_slice in range(time_slice + 1):
if added_time_slice not in self.slice_to_topo.keys():
self.slice_to_topo[added_time_slice] = nx.DiGraph()
self.slice_to_topo[added_time_slice].add_nodes_from(
range(self.nb_node)
)
nodes = self.slice_to_topo[time_slice].nodes
if ((port1 not in nodes[node1]) or (nodes[node1][port1] == False)) and (
(port2 not in nodes[node2]) or (nodes[node2][port2] == False)
):
nodes[node1][port1] = True
nodes[node2][port2] = True
self.slice_to_topo[time_slice].add_edge(node1, node2, port1=port1, port2=port2)
if unidirectional == False:
self.slice_to_topo[time_slice].add_edge(
node2, node1, port1=port2, port2=port1
)
return True
else:
print(
f"Port(s) occupied: Time slice {time_slice} can NOT connect node {node1} port {port1} to node {node2} port {port2} "
)
return False
[docs]
def disconnect(
self, time_slice, node1, node2, port1=0, port2=0, unidirectional=False
) -> bool:
"""
Disconnect two node ports at the given time slice.
Args:
node1 (int): First node
port1 (int): Port of first node
node2 (int): Second node
port2 (int): Port of second node
time_slice (int): Time slice to disconnect
unidirectional (bool, optional): Whether disconnection is unidirectional. Defaults to False.
Returns:
bool: Whether the disconnection was successful
"""
if time_slice not in self.slice_to_topo:
print(f"Time slice {time_slice} not found.")
return False
nodes = self.slice_to_topo[time_slice].nodes
if self.slice_to_topo[time_slice].has_edge(node1, node2) == False:
print(
f"Time slice {time_slice} does not have edge between node {node1} and node {node2}."
)
return False
if nodes[node1][port1] == False:
print(
f"Node {node1} Port {port1} is not the correct port connected to node {node2}."
)
return False
if nodes[node2][port2] == False:
print(
f"Node {node2} Port {port2} is not the correct port connected to node {node1}."
)
return False
nodes[node1][port1] = False
nodes[node2][port2] = False
self.slice_to_topo[time_slice].remove_edge(node1, node2)
if unidirectional == False:
self.slice_to_topo[time_slice].remove_edge(node2, node1)
return True
[docs]
def deploy_topo(self, circuits=[], start_fresh=False) -> bool:
"""
Create ocs schedules based on given circuits or existing slice_to_topo variable (updated by connect() before),
and load them to OCS.
Create nodes in the backend if it is the first time deploy_topo is called.
Args:
circuits (list, optional): A list of tuples (time_slice, node1, node2, port1, port2). Defaults to [].
start_fresh (bool, optional): Whether to start with a fresh topology. Defaults to False.
Returns:
bool: Whether the given circuits are successfully deployed.
"""
if start_fresh:
self.slice_to_topo = {}
nb_time_slices_hint = 0
for time_slice, node1, node2, port1, port2 in circuits:
nb_time_slices_hint = max(nb_time_slices_hint, time_slice + 1)
if node1 == -1 or node2 == -1:
# Placeholder for an empty time slice (e.g. guardband with no circuits)
continue
if not self.connect(time_slice, node1, node2, port1, port2):
print("Topology deployment failed.")
return False
# Create empty DiGraphs for any time slices not yet populated
for ts in range(nb_time_slices_hint):
if ts not in self.slice_to_topo:
self.slice_to_topo[ts] = nx.DiGraph()
self.slice_to_topo[ts].add_nodes_from(range(self.nb_node))
self.nb_time_slices = len(self.slice_to_topo.keys())
if self.nb_time_slices == 0:
raise Exception("No time slices deployed.")
if self.nodes_created:
self.dashboard.update_topology(self.slice_to_topo)
if not self.nodes_created:
self.create_nodes()
self.nodes_created = True
print("Deploying optical topologies...")
self._backend.clear_table(
switch_name="ocs",
table="ocs_schedule",
)
self.setup_ocs()
if hasattr(self._backend, 'gen_schedule'):
self._backend.gen_schedule(self.slice_to_topo)
return True
[docs]
def get_topo(self, time_slice=None) -> nx.Graph:
"""
Get the topology (nx.Graph) at the given time slice.
Args:
time_slice (int): Time slice to retrieve topology for
Returns:
nx.Graph: The network topology at the given time slice, or None if not found
"""
if time_slice is None:
return self.slice_to_topo
if time_slice not in self.slice_to_topo.keys():
print("Time slice not found.")
return None
return self.slice_to_topo[time_slice]
[docs]
def pause_calendar_queue(self):
"""Pause traffic before reconfigure topology. Used in TA architecture.
Sets the active queue for each node to itself, as no pkts should be there,
this effectively pauses the calendar queues during topology reconfiguration.
"""
for node1, node2, attr in nx.to_edgelist(self.slice_to_topo[0]):
port1, port2 = attr["port1"], attr["port2"]
assert port1 == 0 and port2 == 0, (
"Now control calendar queue only supports one link per node"
)
self.device_manager.set_active_queue(f"tor{node1}", node1)
[docs]
def activate_calendar_queue(self):
"""Update active calendar queues based on the current topology, for traffic-aware.
Each calendar queue buffers packets to a destination node.
Pause calendar queues whose packets' dst is not directly connected.
"""
assert len(self.slice_to_topo.keys()) == 1, (
f"Control calendar queue is only supported for TA with one time slice. \
{len(self.slice_to_topo.keys())} time slices found."
)
for node1, node2, attr in nx.to_edgelist(self.slice_to_topo[0]):
port1, port2 = attr["port1"], attr["port2"]
assert port1 == 0 and port2 == 0, (
"Now control calendar queue only supports one link per node"
)
self.device_manager.set_active_queue(f"tor{node1}", node2)
##########################
# Routing #
##########################
[docs]
def add_time_flow_entry(
self, node_id, entries: Union[List[TimeFlowEntry], TimeFlowEntry], routing_mode="Per-hop"
) -> bool:
"""
Add the time flow entry(s) to the node.
Args:
node_id (int): The node to add entries to
entries (TimeFlowEntry or List[TimeFlowEntry]): A TimeFlowEntry or a list of TimeFlowEntry
routing_mode (str): Source or Per-hop
Returns:
bool: Whether the entries were successfully added.
"""
if isinstance(entries, TimeFlowEntry):
entries = [entries]
elif not isinstance(entries, list):
raise ValueError("entries must be a TimeFlowEntry or a list of TimeFlowEntry")
table_entries = []
if routing_mode == "Source":
for entry in entries:
table_entries += utils.tor_table_routing_source(entry, nb_time_slices=self.nb_time_slices)
elif routing_mode == "Per-hop":
for entry in entries:
table_entries += utils.tor_table_routing_per_hop(entry, nb_time_slices=self.nb_time_slices)
else:
assert False, "Unsupported routing mode"
if not self._backend.switch_exists(f"tor{node_id}"):
print(f"Error: Try deploying paths to non-existent node: node{node_id}.")
return False
return self._backend.load_table(f"tor{node_id}", table_entries)
[docs]
def deploy_routing(
self,
paths: List[Path],
routing_mode="Per-hop",
arch_mode="TO",
start_fresh=False,
) -> bool:
"""
Deploy routing to nodes.
Args:
paths (List[Path]): A list of paths to be deployed to the network nodes
routing_mode (str): The routing mode, either "Per-hop" or "Source"
arch_mode (str, optional): The architecture mode, either "TO" (Traffic-Oblivious) or "TA" (Traffic-Aware). Defaults to "TO".
start_fresh (bool, optional): If True, clears existing routing table entries before deploying new ones. Defaults to False.
Returns:
bool: True if routing deployment is successful
"""
# Load utility tables into ToR switches (ip_to_dst, arrive_at_dst, etc.)
self.setup_nodes()
if routing_mode == "Source":
cap = getattr(self._backend, "max_source_route_hops", None)
if cap is not None:
offending = [p for p in paths if len(p.steps) > cap]
if offending:
longest = max(len(p.steps) for p in offending)
sample = offending[0]
warnings.warn(
f"[deploy_routing] {len(offending)}/{len(paths)} "
f"source-routed paths exceed the "
f"{type(self._backend).__name__} SR cap of {cap} "
f"hops (longest={longest}, e.g. src={sample.src} "
f"dst={sample.dst} arrival_ts={sample.arrival_ts}). "
f"Bound the routing function "
f"(`routing_hoho(..., max_hop={cap})`) or switch to "
f"`routing_mode=\"Per-hop\"`.",
RuntimeWarning,
stacklevel=2,
)
if start_fresh:
print("Loading routings...")
table_name = (
"per_hop_routing" if routing_mode == "Per-hop" else "add_source_routing_entries"
)
for node_id in range(self.nb_node):
self._backend.clear_table(
switch_name=f"tor{node_id}",
table=table_name,
)
entry_dict = utils.path2entries(paths, routing_mode, arch_mode=arch_mode)
for src, entries in entry_dict.items():
self.add_time_flow_entry(src, entries, routing_mode=routing_mode)
return True
