Add method to terminate SbDut

switchboard/network.py

@@ -191,6 +191,9 @@ def __init__(self, cmdline=False, tool: str = 'verilator', trace: bool = False,
 
  self.name = name
 
+ # keep track of processes started
+ self.process_collection = ProcessCollection()
+
  if cleanup:
  import atexit
 
@@ -520,10 +523,6 @@ def simulate(self, start_delay=None, run=None, intf_objs=True, plusargs=None):
  if plusargs is None:
  plusargs = []
 
- # keep track of processes started
-
- process_collection = ProcessCollection()
-
  # create interface objects
 
  if self.single_netlist:
@@ -538,7 +537,7 @@ def simulate(self, start_delay=None, run=None, intf_objs=True, plusargs=None):
  plusargs = plusargs_processed
  process = self.dut.simulate(start_delay=start_delay, run=run,
  intf_objs=intf_objs, plusargs=plusargs)
- process_collection.add(process)
+ self.process_collection.add(process)
 
  if intf_objs:
  self.intfs = self.dut.intfs
@@ -590,14 +589,21 @@ def simulate(self, start_delay=None, run=None, intf_objs=True, plusargs=None):
  process = block.simulate(start_delay=start_delay, run=inst.name,
  intf_objs=False, plusargs=inst_plusargs)
 
- process_collection.add(process)
+ self.process_collection.add(process)
 
  # start TCP bridges as needed
  for tcp_kwargs in self.tcp_intfs.values():
  process = start_tcp_bridge(**tcp_kwargs)
- process_collection.add(process)
+ self.process_collection.add(process)
+
+ return self.process_collection
 
- return process_collection
+ def terminate(
+ self,
+ stop_timeout=10,
+ use_sigint=False
+ ):
+ self.process_collection.terminate(stop_timeout=stop_timeout, use_sigint=use_sigint)
 
  def generate_inst_name(self, prefix):
  if prefix not in self.inst_name_counters:

switchboard/sbdut.py

@@ -21,7 +21,7 @@
 from .switchboard import path as sb_path
 from .verilator import verilator_run
 from .icarus import icarus_build_vpi, icarus_find_vpi, icarus_run
-from .util import plusargs_to_args, binary_run
+from .util import plusargs_to_args, binary_run, ProcessCollection
 from .xyce import xyce_flags
 from .ams import make_ams_spice_wrapper, make_ams_verilog_wrapper, parse_spice_subckts
 from .autowrap import (normalize_clocks, normalize_interfaces, normalize_resets, normalize_tieoffs,
@@ -217,6 +217,9 @@ def __init__(
 
  self.intfs = {}
 
+ # keep track of processes started
+ self.process_collection = ProcessCollection()
+
  # simulator-agnostic settings
 
  if builddir is None:
@@ -614,10 +617,20 @@ def simulate(
  args=plusargs_to_args(plusargs) + args
  )
 
+ # Add newly created Popen object to subprocess list
+ self.process_collection.add(p)
+
  # return a Popen object that one can wait() on
 
  return p
 
+ def terminate(
+ self,
+ stop_timeout=10,
+ use_sigint=False
+ ):
+ self.process_collection.terminate(stop_timeout=stop_timeout, use_sigint=use_sigint)
+
  def input_analog(
  self,
  filename: str,

switchboard/util.py

@@ -97,3 +97,35 @@ def wait(self):
  process.join()
  else:
  raise Exception(f'Unknown process type: {type(process)}')
+
+ def terminate(
+ self,
+ stop_timeout=10,
+ use_sigint=False
+ ):
+ if not self.processes:
+ return
+
+ for p in self.processes:
+ if isinstance(p, ProcessCollection):
+ p.terminate(stop_timeout=stop_timeout, use_sigint=use_sigint)
+ else:
+ poll = p.poll()
+ if poll is not None:
+ # process has stopped
+ return
+
+ if use_sigint:
+ try:
+ p.send_signal(signal.SIGINT)
+ p.wait(stop_timeout)
+ return
+ except: # noqa: E722
+ # if there is an exception for any reason, including
+ # Ctrl-C during the wait() call, want to make sure
+ # that the process is actually terminated
+ pass
+
+ # if we get to this point, the process is still running
+ # and sigint didn't work (or we didn't try it)
+ p.terminate()