rumba: Fix backpressure on tc links in veth testbeds

Now tracks netem/htb qdisc state per IPCP instead of in a global
set. It places place netem on the bridge-side interface (host_ifname)
so HTB rate limiting keeps back-pressure (mixing htb and tc on the
same interface doesn't work).

4 files changed, 126 insertions, 53 deletions

diff --git a/rumba/elements/experimentation.py b/rumba/elements/experimentation.py
index 2872f7f..8aa4fe0 100644
--- a/rumba/elements/experimentation.py
+++ b/rumba/elements/experimentation.py
@@ -42,6 +42,12 @@ class Testbed(object):
     """
     Base class for every testbed plugin.
     """
+
+    # Whether this testbed provides interfaces that can carry tc link
+    # quality (rate/delay/loss).  Process-only testbeds (e.g. the local
+    # testbed) have no interfaces, so applying a LinkQuality is a no-op.
+    supports_link_quality = True
+
     def __init__(self,
                  exp_name,
                  username,
@@ -80,9 +86,12 @@ class Testbed(object):
 
         self._swap_in(experiment)
 
-        for layer in experiment.layer_ordering:
-            if layer.is_eth:
-                layer.link_quality.apply(layer)
+        # Process-only testbeds have no interfaces to shape; applying a
+        # LinkQuality there is a no-op (see Testbed.supports_link_quality).
+        if self.supports_link_quality:
+            for layer in experiment.layer_ordering:
+                if layer.is_eth:
+                    layer.link_quality.apply(layer)
 
     @abc.abstractmethod
     def _swap_in(self, experiment):
diff --git a/rumba/elements/topology.py b/rumba/elements/topology.py
index 3d11761..f6ad32a 100644
--- a/rumba/elements/topology.py
+++ b/rumba/elements/topology.py
@@ -21,6 +21,7 @@
 # Foundation, Inc., http://www.fsf.org/about/contact/.
 #
 
+import threading
 from enum import Enum
 
 import rumba.log as log
@@ -159,6 +160,7 @@ class EthDixLayer(Layer):
         Layer.__init__(self, name, layer_type, members)
         self._link_quality = link_quality if link_quality is not None \
             else LinkQuality()
+        self._link_quality_lock = threading.RLock()
 
     def get_ipcp_class(self):
         return EthIPCP
@@ -177,12 +179,15 @@ class EthDixLayer(Layer):
         if not _link_quality:
             raise ValueError("Cannot set link_quality to None, use del "
                              "link_quality to reset")
-        self._link_quality = _link_quality
-        _link_quality.apply(self)
+        with self._link_quality_lock:
+            self._link_quality = _link_quality
+            _link_quality.apply(self)
 
     @link_quality.deleter
     def link_quality(self):
-        self._link_quality.deactivate(self)
+        with self._link_quality_lock:
+            self._link_quality.deactivate(self)
+            self._link_quality = LinkQuality()
 
     def set_delay(self, delay=0,
                   jitter=None,
@@ -202,8 +207,10 @@ class EthDixLayer(Layer):
         :type distribution: :py:class:`.Distribution`
         """
         new_delay = Delay(delay, jitter, correlation, distribution)
-        new_quality = LinkQuality.clone(self.link_quality, delay=new_delay)
-        self.link_quality = new_quality
+        with self._link_quality_lock:
+            new_quality = LinkQuality.clone(
+                self._link_quality, delay=new_delay)
+            self.link_quality = new_quality
 
     def set_loss(self,
                  loss=0,
@@ -218,8 +225,10 @@ class EthDixLayer(Layer):
         :type correlation: :py:class:`int` or :py:class:`float`
         """
         new_loss = Loss(loss, correlation)
-        new_quality = LinkQuality.clone(self.link_quality, loss=new_loss)
-        self.link_quality = new_quality
+        with self._link_quality_lock:
+            new_quality = LinkQuality.clone(
+                self._link_quality, loss=new_loss)
+            self.link_quality = new_quality
 
     def set_rate(self, rate=None):
         """
@@ -228,8 +237,9 @@ class EthDixLayer(Layer):
         :param rate: The desired rate in mbps
         :type rate: :py:class:`int`
         """
-        new_quality = LinkQuality.clone(self.link_quality, rate=rate)
-        self.link_quality = new_quality
+        with self._link_quality_lock:
+            new_quality = LinkQuality.clone(self._link_quality, rate=rate)
+            self.link_quality = new_quality
 
     def set_duplicate(self, duplicate=0, correlation=None):
         """
@@ -242,8 +252,10 @@ class EthDixLayer(Layer):
         :type correlation: :py:class:`int` or :py:class:`float`
         """
         new_dup = Duplicate(duplicate, correlation)
-        new_quality = LinkQuality.clone(self.link_quality, duplicate=new_dup)
-        self.link_quality = new_quality
+        with self._link_quality_lock:
+            new_quality = LinkQuality.clone(
+                self._link_quality, duplicate=new_dup)
+            self.link_quality = new_quality
 
     def set_reorder(self, reorder=0, correlation=None):
         """
@@ -256,9 +268,10 @@ class EthDixLayer(Layer):
         :type correlation: :py:class:`int` or :py:class:`float`
         """
         new_reorder = Reorder(reorder, correlation)
-        new_quality = LinkQuality.clone(
-            self.link_quality, reorder=new_reorder)
-        self.link_quality = new_quality
+        with self._link_quality_lock:
+            new_quality = LinkQuality.clone(
+                self._link_quality, reorder=new_reorder)
+            self.link_quality = new_quality
 
     def set_corrupt(self, corrupt=0, correlation=None):
         """
@@ -271,9 +284,10 @@ class EthDixLayer(Layer):
         :type correlation: :py:class:`int` or :py:class:`float`
         """
         new_corrupt = Corrupt(corrupt, correlation)
-        new_quality = LinkQuality.clone(
-            self.link_quality, corrupt=new_corrupt)
-        self.link_quality = new_quality
+        with self._link_quality_lock:
+            new_quality = LinkQuality.clone(
+                self._link_quality, corrupt=new_corrupt)
+            self.link_quality = new_quality
 
     def set_quality(self, delay, loss, rate):
         """
@@ -287,7 +301,8 @@ class EthDixLayer(Layer):
         :type rate: :py:class:`int`
         """
         new_quality = LinkQuality(delay, loss, rate)
-        self.link_quality = new_quality
+        with self._link_quality_lock:
+            self.link_quality = new_quality
 
 
 class EthLlcLayer(EthDixLayer):
@@ -645,7 +660,6 @@ class LinkQuality(object):
     """
     A class representing the link quality.
     """
-    _active = set()
 
     @classmethod
     def clone(cls, old_quality, delay=None, loss=None, rate=None,
@@ -734,26 +748,35 @@ class LinkQuality(object):
     def build_commands(self, ipcp):
         netem_cmd = []
         cmds = []
-        qref = "root"
 
-        if ipcp in LinkQuality._active:
-            cmds.append("tc qdisc del dev %s root" % ipcp.ifname)
+        # Rate limiting (HTB) goes on the IPCP interface so a blocking
+        # send back-pressures the source.  All netem impairments go on
+        # the bridge-side interface (host_ifname), where frames are
+        # already orphaned -- netem's skb_orphan() there no longer
+        # defeats that back-pressure.  Testbeds without a bridge (e.g.
+        # UDP veths) have no host_ifname and put both on the IPCP
+        # interface.  apply_to_ipcp() clears any prior qdiscs first, so
+        # these are plain adds.
+        netem_dev = getattr(ipcp, 'host_ifname', None) or ipcp.ifname
 
         if self.rate:
             cmds.append(
                 "tc qdisc add dev %s root handle 1: htb default 1"
                 % ipcp.ifname)
             cmds.append(
-                "tc class add dev %s parent 1: classid 1:1 htb rate %imbit"
-                % (ipcp.ifname, self.rate))
-            qref = "parent 1:1"
+                "tc class add dev %s parent 1: classid 1:1 "
+                "htb rate %imbit" % (ipcp.ifname, self.rate))
 
         netem_parts = [self.delay, self.loss,
                        self.duplicate, self.reorder, self.corrupt]
         if any(netem_parts):
+            if self.rate and netem_dev == ipcp.ifname:
+                qref = "parent 1:1"      # no bridge: chain under HTB
+            else:
+                qref = "root"
             netem_cmd.append(
-                "tc qdisc add dev %s %s netem"
-                % (ipcp.ifname, qref))
+                "tc qdisc add dev %s %s netem limit 65536"
+                % (netem_dev, qref))
             for part in netem_parts:
                 if part:
                     netem_cmd.append(part.build_command())
@@ -761,34 +784,67 @@ class LinkQuality(object):
 
         return cmds
 
-    def apply(self, layer):
+    def apply_to_ipcp(self, ipcp):
+        """Apply this link quality configuration to a single *ipcp*."""
+        if getattr(ipcp, 'node', None) is None:
+            logger.error(
+                "Could not apply LinkQuality to IPCP because "
+                "ipcp.node is None")
+            return
+        if not ipcp.ifname:
+            logger.error(
+                "Could not apply LinkQuality to IPCP because "
+                "the interface name is None")
+            return
         if not (self.delay or self.loss or self.rate
                 or self.duplicate or self.reorder or self.corrupt):
-            self.deactivate(layer)
+            self.deactivate_ipcp(ipcp)
         else:
-            for ipcp in layer.ipcps:
-                if not ipcp.ifname:
-                    logger.error(
-                        "Could not apply LinkQuality to IPCP because "
-                        "the interface name is None")
-                    continue
-                ipcp.node.execute_commands(
-                    self.build_commands(ipcp), as_root=True)
-                LinkQuality._active.add(ipcp)
+            # Clear any prior qdiscs first (best-effort, both devices) so
+            # the adds in build_commands never hit an existing root qdisc.
+            self.deactivate_ipcp(ipcp)
+            ipcp.node.execute_commands(
+                self.build_commands(ipcp), as_root=True)
+            ipcp._netem_active = True
+
+    def deactivate_ipcp(self, ipcp):
+        """Remove any active netem/htb qdisc from a single *ipcp*."""
+        if not getattr(ipcp, '_netem_active', False):
+            return
+        # Clear the flag up-front: teardown is best-effort, and we do not
+        # want a later failure in the tc command to leave the IPCP stuck
+        # in the "active" state.
+        ipcp._netem_active = False
+        if getattr(ipcp, 'node', None) is None:
+            logger.error(
+                "Could not remove LinkQuality from IPCP because "
+                "ipcp.node is None")
+            return
+        if not ipcp.ifname:
+            logger.error(
+                "Could not remove LinkQuality from IPCP because "
+                "the interface name is None")
+            return
+        # Best-effort teardown: HTB lives on the IPCP interface, netem on
+        # the bridge-side interface; either may be absent, so do not check.
+        ipcp.node.execute_command(
+            "tc qdisc del dev %s root" % ipcp.ifname,
+            as_root=True, check=False)
+        host_ifname = getattr(ipcp, 'host_ifname', None)
+        if host_ifname and host_ifname != ipcp.ifname:
+            ipcp.node.execute_command(
+                "tc qdisc del dev %s root" % host_ifname,
+                as_root=True, check=False)
+
+    def apply(self, layer):
+        """Apply this link quality to all IPCPs in *layer*."""
+        for ipcp in layer.ipcps:
+            self.apply_to_ipcp(ipcp)
 
     def deactivate(self, layer):
+        """Remove any active netem/htb qdisc from all IPCPs in *layer*."""
         for ipcp in layer.ipcps:
-            if ipcp not in LinkQuality._active:
-                continue
-            if not ipcp.ifname:
-                logger.error(
-                    "Could not remove LinkQuality from IPCP because "
-                    "the interface name is None")
-                continue
-            ipcp.node.execute_command(
-                "tc qdisc del dev %s root" % ipcp.ifname,
-                as_root=True)
-            LinkQuality._active.remove(ipcp)
+            self.deactivate_ipcp(ipcp)
 
 
 # ---------------------------------------------------------------------------
diff --git a/rumba/testbeds/local.py b/rumba/testbeds/local.py
index 9326d94..ad0db9f 100644
--- a/rumba/testbeds/local.py
+++ b/rumba/testbeds/local.py
@@ -39,6 +39,10 @@ class Testbed(mod.Testbed):
     processes locally. Also useful for debugging in the other plugins.
     """
 
+    # Process-only testbed: no network interfaces, so tc link quality
+    # (rate/delay/loss) cannot be applied -- treat apply as a no-op.
+    supports_link_quality = False
+
     def __init__(self, exp_name='foo', username='bar',
                  proj_name="rumba", password=""):
         """
diff --git a/rumba/testbeds/localnet.py b/rumba/testbeds/localnet.py
index 95a6bdc..4278367 100644
--- a/rumba/testbeds/localnet.py
+++ b/rumba/testbeds/localnet.py
@@ -95,8 +95,12 @@ class LocalNetworkTestbed(mod.Testbed):
                 host_end = 'vb%d' % veth_idx
                 veth_idx += 1
 
-                # Assign the IPCP-facing interface name
+                # Assign the IPCP-facing interface name, and expose the
+                # bridge-side peer so LinkQuality can place netem there
+                # (its skb_orphan() would otherwise defeat HTB
+                # back-pressure on the IPCP interface).
                 ipcp.ifname = ipcp_end
+                ipcp.host_ifname = host_end
 
                 # Create veth pair
                 _run('sudo ip link add %s type veth peer name %s'