Added the HPFQ scheduler and reference scheduling

This commit adds the HierarchicalPacket  Fair  Queueing  (HPFQ)
scheduler from the paper "Programmable packet scheduling at line rate"
by Sivaraman, Anirudh, et al.

This commit also changes the framework to make it easy to schedule
references to flows, queues, or schedulers. It does so by adding a
reference with every enqueue operation. By adding the reference with the
packet or flow, we can create a rank from packet or flow and queue the
reference.

This commit adds the HierarchicalPacket  Fair  Queueing  (HPFQ) scheduler from the paper "Programmable packet scheduling at line rate" by Sivaraman, Anirudh, et al. This commit also changes the framework to make it easy to schedule references to flows, queues, or schedulres.

Signed-off-by: Frey Alfredsson <[email protected]>

Author: freysteinn

queue-exp/pifo_fifo.py

@@ -39,12 +39,13 @@
 class Fifo(SchedulingAlgorithm):
     """First in, first out (FIFO)"""
 
-    def __init__(self):
+    def __init__(self, name=None):
+        super().__init__(name)
         self._pifo = Pifo()
 
-    def enqueue(self, item):
+    def enqueue(self, ref, item):
         rank = self.get_rank(item)
-        self._pifo.enqueue(item, rank)
+        self._pifo.enqueue(ref, rank)
 
     def get_rank(self, item):
         return self._pifo.qlen

queue-exp/pifo_hpfq.py

@@ -0,0 +1,88 @@
+#!/usr/bin/env python3
+# coding: utf-8 -*-
+#
+# SPDX-License-Identifier: GPL-3.0-or-later
+#
+# pifo-hpfq.py
+
+"""HierarchicalPacket  Fair  Queueing  (HPFQ)
+
+This scheduling algorithm is mentioned in the paper "Programmable packet
+scheduling at line rate" by Sivaraman, Anirudh, et al. It creates a hierarchy of
+WFQ schedulers. The central scheduler is called root and contains references to
+other WFQ schedulers. Those two WFQ schedulers are called left and right. We
+chose that packets with flow ids lower than ten go into the left scheduler in
+our implementation. In contrast, the others go into the right scheduler."""
+
+__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_wfq import Wfq
+
+
+class Hpfq(SchedulingAlgorithm):
+    """HierarchicalPacket  Fair  Queueing  (HPFQ)"""
+
+    def __init__(self, name=None):
+        super().__init__(name)
+        self._root = Wfq("root")
+        self._left = Wfq("Left")
+        self._right = Wfq("Right")
+
+    def enqueue(self, ref, pkt):
+        queue = None
+        if pkt.flow < 10:
+            self._left.enqueue(ref, pkt)
+            queue = self._left
+        else:
+            self._right.enqueue(ref, pkt)
+            queue = self._right
+
+        self._root.enqueue(queue, pkt)
+
+    def dequeue(self):
+        queue = self._root.dequeue()
+        return queue.dequeue() if queue is not None else None
+
+    def dump(self):
+        print("   Root:")
+        self._root.dump()
+        print("   Left:")
+        self._left.dump()
+        print("   Right:")
+        self._right.dump()
+
+
+if __name__ == "__main__":
+    pkts = [
+        Packet(flow=1, idn=1, length=200),
+        Packet(flow=1, idn=2, length=200),
+        Packet(flow=10, idn=1, length=200),
+        Packet(flow=10, idn=2, length=200),
+        Packet(flow=2, idn=1, length=100),
+        Packet(flow=2, idn=2, length=100),
+        Packet(flow=2, idn=3, length=100),
+        Packet(flow=20, idn=1, length=100),
+        Packet(flow=20, idn=2, length=100),
+        Packet(flow=20, idn=3, length=100),
+    ]
+    Runner(pkts, Hpfq()).run()

queue-exp/pifo_lib.py

@@ -48,7 +48,7 @@ def run(self):
         print("  Inserting packets into scheduler:")
         pprint(self.input_pkts, indent=4)
         for p in self.input_pkts:
-            self.scheduler.enqueue(p)
+            self.scheduler.enqueue(p, p)
         print("  Scheduler state:")
         self.scheduler.dump()
         output = []
@@ -70,10 +70,10 @@ class SchedulingAlgorithm():
     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 __init__(self, name=None):
+        self._name = name
 
-    def enqueue(self, item):
+    def enqueue(self, pkt):
         raise NotImplementedError(self.__class__.__name__ + ' missing implementation')
 
     def dequeue(self):
@@ -83,25 +83,28 @@ def dump(self):
         raise NotImplementedError(self.__class__.__name__ + ' missing implementation')
 
     def __next__(self):
-        item = self.dequeue()
-        if item is None:
+        pkt = self.dequeue()
+        if pkt is None:
             raise StopIteration
-        return item
+        return pkt
 
     def __iter__(self):
         return self
 
     def __repr__(self):
-        return f"{self.__class__.__name__} - {self.__class__.__doc__}"
+        result = f"{self.__class__.__name__} - {self.__class__.__doc__}"
+        if self._name is not None:
+            result = f"{self._name}: {result}"
+        return result
 
 
 class Queue:
     def __init__(self, idx=None):
         self._list = []
         self.idx = idx
 
-    def enqueue(self, item):
-        self._list.append(item)
+    def enqueue(self, ref, rank=None):
+        self._list.append(ref)
 
     def peek(self):
         try:
@@ -139,11 +142,11 @@ def dump(self):
 
 
 class Pifo(Queue):
-    def enqueue(self, item, rank):
+    def enqueue(self, ref, rank):
         if rank is None:
             raise ValueError("Rank can't be of value 'None'.")
 
-        super().enqueue((rank, item))
+        super().enqueue((rank, ref))
         self.sort()
 
     def sort(self):
@@ -160,8 +163,7 @@ def peek(self):
 
 class Flow(Queue):
     def __init__(self, idx):
-        super().__init__()
-        self.idx = idx
+        super().__init__(idx)
 
     def __repr__(self):
         return f"F(I:{self.idx}, Q:{self.qlen}, L:{self.length})"

queue-exp/pifo_srpt.py

@@ -44,7 +44,8 @@
 class Srpt(SchedulingAlgorithm):
     """Shortest Remaining Processing Time"""
 
-    def __init__(self):
+    def __init__(self, name=None):
+        super().__init__(name)
         self._pifo = Pifo()
         self._flow_tracker = FlowTracker()
 
@@ -57,22 +58,22 @@ def __init__(self):
             else:
                 self._remains[pkt.flow] = pkt.length
 
-    def get_rank(self, pkt):
-        rank = self._remains[pkt.flow]
-        self._remains[pkt.flow] -= pkt.length
+    def get_rank(self, item):
+        rank = self._remains[item.flow]
+        self._remains[item.flow] -= item.length
         return rank
 
-    def enqueue(self, item):
+    def enqueue(self, ref, 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
+        item = None
         if flow is not None:
-            pkt = flow.dequeue()
-        return pkt
+            item = flow.dequeue()
+        return item
 
     def dump(self):
         self._pifo.dump()

queue-exp/pifo_stfq.py

@@ -41,7 +41,8 @@
 class Stfq(SchedulingAlgorithm):
     """Start-Time Fair Queuing (STFQ)"""
 
-    def __init__(self):
+    def __init__(self, name=None):
+        super().__init__(name)
         self._pifo = Pifo()
 
         self._last_finish = {}
@@ -56,9 +57,9 @@ def get_rank(self, pkt):
         self._last_finish[flow_id] = rank + pkt.length
         return rank
 
-    def enqueue(self, item):
+    def enqueue(self, ref, item):
         rank = self.get_rank(item)
-        self._pifo.enqueue(item, rank)
+        self._pifo.enqueue(ref, rank)
 
     def dequeue(self):
         return self._pifo.dequeue()

queue-exp/pifo_wfq.py

@@ -41,7 +41,8 @@
 class Wfq(SchedulingAlgorithm):
     """Weighted Fair Queueing (WFQ)"""
 
-    def __init__(self):
+    def __init__(self, name=None):
+        super().__init__(name)
         self._pifo = Pifo()
         self._last_finish = {}
         self._virt_time = 0
@@ -56,9 +57,9 @@ def get_rank(self, pkt):
         self._last_finish[flow] = rank + pkt.length / weight
         return rank
 
-    def enqueue(self, item):
+    def enqueue(self, ref, item):
         rank = self.get_rank(item)
-        self._pifo.enqueue(item, rank)
+        self._pifo.enqueue(ref, rank)
 
     def dequeue(self):
         return self._pifo.dequeue()