Changed and refactored the code

queue-exp/pifo_fifo.py

@@ -5,6 +5,13 @@
 #
 # pifo-fifo.py
 
+"""First in, first out (FIFO)
+
+The FIFO scheduling algorithm preserves the order of the scheduled packets. This
+implementation is here for completeness and uses a PIFO. It is here to help
+people understand how to add new scheduling algorithms to this framework.
+"""
+
 __copyright__ = """
 Copyright (c) 2021, Toke Høiland-Jørgensen <[email protected]>
 Copyright (c) 2021, Frey Alfredsson <[email protected]>
@@ -26,11 +33,28 @@
 """
 
 from pifo_lib import Packet, Runner, Pifo
+from pifo_lib import SchedulingAlgorithm
+
+
+class Fifo(SchedulingAlgorithm):
+    """First in, first out (FIFO)"""
 
+    def __init__(self):
+        self._pifo = Pifo()
+
+    def enqueue(self, item):
+        rank = self.get_rank(item)
+        self._pifo.enqueue(item, rank)
 
-class Fifo(Pifo):
     def get_rank(self, item):
-        return self.qlen
+        return self._pifo.qlen
+
+    def dequeue(self):
+        return self._pifo.dequeue()
+
+    def dump(self):
+        self._pifo.dump()
+
 
 if __name__ == "__main__":
     pkts = [

queue-exp/pifo_lib.py

@@ -27,32 +27,74 @@ def __init__(self, flow, idn, length=1):
         self.flow = flow
         self.idn = idn
         self.length = length
-        self.rank = 0
 
     def __repr__(self):
         return f"P(F:{self.flow}, I:{self.idn}, L:{self.length})"
 
 
 class Runner:
-    def __init__(self, pkts, queue):
+    """This class is responsible for running a test on a packet scheduling
+    algorithm. It is accountable for enquing and dequeing packets. For now, it
+    does so by dequing as many packets as it enqued. In the next iteration, when
+    we add pacing, it will need to handle virtual time cycling.
+    """
+
+    def __init__(self, pkts, scheduler):
         self.input_pkts = pkts
-        self.queue = queue
+        self.scheduler = scheduler
 
     def run(self):
-        print(f"Running with queue: {self.queue}")
-        print("  Inserting packets into queue:")
+        print(f"Running with scheduler: {self.scheduler}")
+        print("  Inserting packets into scheduler:")
         pprint(self.input_pkts, indent=4)
         for p in self.input_pkts:
-            self.queue.enqueue(p)
-        print("  Queue state:")
-        self.queue.dump()
+            self.scheduler.enqueue(p)
+        print("  Scheduler state:")
+        self.scheduler.dump()
         output = []
-        for p in self.queue:
+
+        for p in self.scheduler:
             output.append(p)
         print("  Got packets from queue:")
         pprint(output, indent=4)
 
 
+class SchedulingAlgorithm():
+
+    """A queuing packet scheduling algorithm requires an abstraction that keeps
+    the queuing data structure and the algorithm separate. To create a new
+    Scheduling algorithm, inherit this class, add the scheduling data structures
+    to the constructor, and implement the constructor, enqueue, dequeue, and the
+    dump functions.
+
+    Please look at the pifo_fifo.py to see how you implement a FIFO.
+    """
+
+    def __init__(self):
+        raise NotImplementedError(self.__class__.__name__ + ' missing implementation')
+
+    def enqueue(self, item):
+        raise NotImplementedError(self.__class__.__name__ + ' missing implementation')
+
+    def dequeue(self):
+        raise NotImplementedError(self.__class__.__name__ + ' missing implementation')
+
+    def dump(self):
+        raise NotImplementedError(self.__class__.__name__ + ' missing implementation')
+
+    def __next__(self):
+        item = self.dequeue()
+        if item is None:
+            raise StopIteration
+        return item
+
+    def __iter__(self):
+        return self
+
+    def __repr__(self):
+        return f"{self.__class__.__name__} - {self.__class__.__doc__}"
+
+
 class Queue:
     def __init__(self, idx=None):
         self._list = []
@@ -97,11 +139,10 @@ def dump(self):
 
 
 class Pifo(Queue):
-
-    def enqueue(self, item, rank=None):
+    def enqueue(self, item, rank):
         if rank is None:
-            rank = self.get_rank(item)
-        item.rank = rank
+            raise ValueError("Rank can't be of value 'None'.")
+
         super().enqueue((rank, item))
         self.sort()
 
@@ -116,24 +157,42 @@ def peek(self):
         itm = super().peek()
         return itm[1] if itm else None
 
-    def get_rank(self, item):
-        raise NotImplementedError
-
 
 class Flow(Queue):
     def __init__(self, idx):
         super().__init__()
         self.idx = idx
-        self.rank = 0
 
     def __repr__(self):
-        return f"F({self.idx})"
+        return f"F(I:{self.idx}, Q:{self.qlen}, L:{self.length})"
 
-    # Return the length of the first packet in the queue as the "length" of the
-    # flow. This is not the correct thing to do, but it works as a stopgap
-    # solution for testing the hierarchical mode, and we're only using
-    # unit-length for that anyway
     @property
     def length(self):
-        itm = self.peek()
-        return itm.length if itm else 0
+        result = 0
+        for itm in self._list:
+            result += itm.length if itm else 0
+        return result
+
+
+class FlowTracker():
+    """This class provides us with the typical operation of keeping track of
+    flows. Use this class in your scheduling algorithms when your algorithm only
+    has one type of flows.
+    """
+
+    def __init__(self):
+        self._flows = {}
+
+    def enqueue(self, pkt, flow_id=None):
+        if not isinstance(pkt, Packet):
+            raise ValueError(f"Expected a packet, but got '{pkt}' instead.")
+        if flow_id is None:
+            flow_id = pkt.flow
+        if not flow_id in self._flows:
+            self._flows[flow_id] = Flow(flow_id)
+        flow = self._flows[flow_id]
+        flow.enqueue(pkt)
+        return flow
+
+    def get_flow(self, flow_id):
+        return self._flows[flow_id]

queue-exp/pifo_srpt.py

@@ -0,0 +1,88 @@
+#!/usr/bin/env python3
+# coding: utf-8 -*-
+#
+# SPDX-License-Identifier: GPL-3.0-or-later
+#
+# pifo-srpt.py
+
+"""Shortest Remaining Processing Time (SRPT).
+
+This scheduling algorithm is referenced in companion C++ implementation for the
+paper "Programmable packet scheduling at line rate" by Sivaraman, Anirudh, et
+al.
+
+It schedules packets in the order of how much data the flow has left. It assumes
+complete knowledge of the flow length. In the real world, this would either need
+to be estimated or limited to predictable flows.
+"""
+
+__copyright__ = """
+Copyright (c) 2021 Toke Høiland-Jørgensen <[email protected]>
+Copyright (c) 2021 Frey Alfredsson <[email protected]>
+"""
+
+__license__ = """
+This program is free software: you can redistribute it and/or modify
+it under the terms of the GNU General Public License as published by
+the Free Software Foundation, either version 3 of the License, or
+(at your option) any later version.
+
+This program is distributed in the hope that it will be useful,
+but WITHOUT ANY WARRANTY; without even the implied warranty of
+MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+GNU General Public License for more details.
+
+You should have received a copy of the GNU General Public License
+along with this program.  If not, see <http://www.gnu.org/licenses/>.
+"""
+
+from pifo_lib import Packet, Runner, Pifo
+from pifo_lib import SchedulingAlgorithm
+from pifo_lib import FlowTracker
+
+
+class Srpt(SchedulingAlgorithm):
+    """Shortest Remaining Processing Time"""
+
+    def __init__(self):
+        self._pifo = Pifo()
+        self._flow_tracker = FlowTracker()
+
+        self._remains = {}
+
+        # We cheat by accessing the global packet list directly
+        for pkt in pkts:
+            if pkt.flow in self._remains.keys():
+                self._remains[pkt.flow] += pkt.length
+            else:
+                self._remains[pkt.flow] = pkt.length
+
+    def get_rank(self, pkt):
+        rank = self._remains[pkt.flow]
+        self._remains[pkt.flow] -= pkt.length
+        return rank
+
+    def enqueue(self, item):
+        flow = self._flow_tracker.enqueue(item)
+        rank = self.get_rank(item)
+        self._pifo.enqueue(flow, rank)
+
+    def dequeue(self):
+        flow = self._pifo.dequeue()
+        pkt = None
+        if flow is not None:
+            pkt = flow.dequeue()
+        return pkt
+
+    def dump(self):
+        self._pifo.dump()
+
+if __name__ == "__main__":
+    pkts = [
+        Packet(flow=1, idn=1, length=2),
+        Packet(flow=1, idn=2, length=2),
+        Packet(flow=2, idn=1, length=1),
+        Packet(flow=2, idn=2, length=1),
+        Packet(flow=2, idn=3, length=1),
+    ]
+    Runner(pkts, Srpt()).run()