nmigen/sim/_core.py

   1 __all__ = ["Process", "Timeline"]
   2
   3
   4 class Process:
   5     def __init__(self, *, is_comb):
   6         self.is_comb  = is_comb
   7
   8         self.reset()
   9
  10     def reset(self):
  11         self.runnable = self.is_comb
  12         self.passive  = True
  13
  14     def run(self):
  15         raise NotImplementedError
  16
  17
  18 class Timeline:
  19     def __init__(self):
  20         self.now = 0.0
  21         self.deadlines = dict()
  22
  23     def reset(self):
  24         self.now = 0.0
  25         self.deadlines.clear()
  26
  27     def at(self, run_at, process):
  28         assert process not in self.deadlines
  29         self.deadlines[process] = run_at
  30
  31     def delay(self, delay_by, process):
  32         if delay_by is None:
  33             run_at = self.now
  34         else:
  35             run_at = self.now + delay_by
  36         self.at(run_at, process)
  37
  38     def advance(self):
  39         nearest_processes = set()
  40         nearest_deadline = None
  41         for process, deadline in self.deadlines.items():
  42             if deadline is None:
  43                 if nearest_deadline is not None:
  44                     nearest_processes.clear()
  45                 nearest_processes.add(process)
  46                 nearest_deadline = self.now
  47                 break
  48             elif nearest_deadline is None or deadline <= nearest_deadline:
  49                 assert deadline >= self.now
  50                 if nearest_deadline is not None and deadline < nearest_deadline:
  51                     nearest_processes.clear()
  52                 nearest_processes.add(process)
  53                 nearest_deadline = deadline
  54
  55         if not nearest_processes:
  56             return False
  57
  58         for process in nearest_processes:
  59             process.runnable = True
  60             del self.deadlines[process]
  61         self.now = nearest_deadline
  62
  63         return True