diff options
| -rw-r--r-- | rumba/elements/experimentation.py | 15 | ||||
| -rw-r--r-- | rumba/elements/topology.py | 154 | ||||
| -rw-r--r-- | rumba/testbeds/local.py | 4 | ||||
| -rw-r--r-- | rumba/testbeds/localnet.py | 6 |
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' |