From e3081d9bccbdc7df13ad8a2e8b3fb185c1e9b466 Mon Sep 17 00:00:00 2001
From: Saraswath BV <saraswathgowda@gmail.com>
Date: Mon, 2 Nov 2026 09:16:42 -0600
Subject: [PATCH 1/4] Add adaptive probe scheduling and multipath support

---
 controller/failover.py       | 370 ++++++++++++++++++++++++++++++-----
 controller/multipath.py      |  75 +++++++
 controller/probe_schedule.py |  73 +++++++
 src/client.c                 |  36 +++-
 tests/test_multipath.py      |  38 ++++
 tests/test_probe_schedule.py |  87 ++++++++
 6 files changed, 618 insertions(+), 61 deletions(-)
 create mode 100644 controller/multipath.py
 create mode 100644 controller/probe_schedule.py
 create mode 100644 tests/test_multipath.py
 create mode 100644 tests/test_probe_schedule.py

diff --git a/controller/failover.py b/controller/failover.py
index 656bfbf..fe0e8ef 100644
--- a/controller/failover.py
+++ b/controller/failover.py
@@ -34,9 +34,11 @@
 sys.path.insert(0, str(Path(__file__).resolve().parents[1]))
 from controller.routing import PathPolicy, SWITCHES, forwarding_rules
 from controller.probes import ProbeTracker, ENDPOINTS, ETHERTYPE
+from controller.probe_schedule import AdaptiveProbeSchedule
 
 
 COOKIE = 0x5255445000000001
+MULTIPATH_COOKIE = 0x5255445000000010
 
 
 class FailoverController(AppBase):
@@ -44,7 +46,10 @@ class FailoverController(AppBase):
 
     def __init__(self, *args, **kwargs):
         super().__init__(*args, **kwargs)
-        self.policy = PathPolicy(os.environ.get("RUDP_MODE", "failover"), os.environ.get("RUDP_INITIAL_PATH", "upper"))
+        self.policy = PathPolicy(
+            os.environ.get("RUDP_MODE", "failover"),
+            os.environ.get("RUDP_INITIAL_PATH", "upper"),
+        )
         self.datapaths = {}
         self.discovered = set()
         self.started = False
@@ -53,135 +58,292 @@ def __init__(self, *args, **kwargs):
         self.pending = {}
         self.failed_generation = False
         self.cause = "startup"
+
         event_path = os.environ.get("RUDP_EVENT_LOG")
         self.event_file = open(event_path, "a", buffering=1) if event_path else None
+
         self._event("controller_started", mode=self.policy.mode)
+
         self.probes = ProbeTracker()
+        self.probe_schedules = {
+            key: AdaptiveProbeSchedule() for key in ENDPOINTS
+        }
+
+        # Each path/direction has its own next-probe deadline.
+        self.next_probe_ns = {key: 0 for key in ENDPOINTS}
+
         self.last_route_change_ns = 0
         self.monitor = hub.spawn(self._monitor)
 
     def _refresh_health(self, now):
         self.policy.health = self.probes.scores(now)
 
+        # Adapt each direction's schedule according to its own path's health.
+        # A degraded upper path should not automatically accelerate the lower
+        # path's probes.
+        for (path, direction), schedule in self.probe_schedules.items():
+            path_score = self.policy.health.get(path)
+            degraded = path_score is None or path_score > 0.10
+            schedule.set_degraded(degraded, now)
+
     def _monitor(self):
         while True:
             now = time.clock_gettime_ns(time.CLOCK_MONOTONIC)
+
+            # Record timed-out probes and make their next probe eligible now.
             for observation in self.probes.expire(now):
                 self._event("probe_result", **observation)
+
+                key = (observation["path"], observation["direction"])
+                self.probe_schedules[key].record_result(False, now)
+                self.next_probe_ns[key] = now
+
             self._refresh_health(now)
+
             if self.started:
-                self._event("path_health", scores=self.policy.health,
-                            source="bidirectional_ethernet_probes", samples=self.probes.samples_as_json(),
-                            hysteresis=.02, hold_down_seconds=2)
+                self._event(
+                    "path_health",
+                    scores=self.policy.health,
+                    source="bidirectional_ethernet_probes",
+                    samples=self.probes.samples_as_json(),
+                    hysteresis=0.02,
+                    hold_down_seconds=2,
+                )
                 self._reconcile("health_update")
+
+                # Send only the probes whose individual deadlines have arrived.
                 for (path, direction), (source, out_port, _, _) in ENDPOINTS.items():
                     if not self.policy.available(path):
                         continue
+
                     dp = self.datapaths.get(source)
                     if dp is None:
                         continue
+
+                    key = (path, direction)
+                    if now < self.next_probe_ns[key]:
+                        continue
+
                     frame = self.probes.send(path, direction, now)
+
                     dp.send_msg(dp.ofproto_parser.OFPPacketOut(
-                        datapath=dp, buffer_id=dp.ofproto.OFP_NO_BUFFER,
+                        datapath=dp,
+                        buffer_id=dp.ofproto.OFP_NO_BUFFER,
                         in_port=dp.ofproto.OFPP_CONTROLLER,
-                        actions=[dp.ofproto_parser.OFPActionOutput(out_port)], data=frame))
-            hub.sleep(.25)
+                        actions=[dp.ofproto_parser.OFPActionOutput(out_port)],
+                        data=frame,
+                    ))
+
+                    interval_ns = int(
+                        self.probe_schedules[key].interval_seconds(now)
+                        * 1_000_000_000
+                    )
+                    self.next_probe_ns[key] = now + interval_ns
+
+            # Wake at the next probe deadline or pending-probe timeout.
+            # Cap sleep at 250 ms so health and controller state are refreshed.
+            deadlines = []
+
+            if self.started:
+                for (path, direction), (source, _, _, _) in ENDPOINTS.items():
+                    if not self.policy.available(path):
+                        continue
+                    if source not in self.datapaths:
+                        continue
+
+                    deadline = self.next_probe_ns[(path, direction)]
+                    if deadline > now:
+                        deadlines.append(deadline)
+
+                for _, _, sent_ns in self.probes.pending.values():
+                    timeout_deadline = sent_ns + self.probes.timeout_ns
+                    if timeout_deadline > now:
+                        deadlines.append(timeout_deadline)
+
+            if deadlines:
+                sleep_ns = max(1_000_000, min(deadlines) - now)
+                interval = min(sleep_ns / 1_000_000_000, 0.25)
+            else:
+                interval = 0.25
+
+            hub.sleep(interval)
 
     @set_ev_cls(ofp_event.EventOFPPacketIn, MAIN_DISPATCHER)
     def probe_received(self, event):
         msg = event.msg
         now = time.clock_gettime_ns(time.CLOCK_MONOTONIC)
-        observation = self.probes.receive(bytes(msg.data), msg.datapath.id, msg.match.get("in_port"), now)
+
+        observation = self.probes.receive(
+            bytes(msg.data),
+            msg.datapath.id,
+            msg.match.get("in_port"),
+            now,
+        )
+
         if observation:
             self._event("probe_result", **observation)
+
+            key = (observation["path"], observation["direction"])
+            schedule = self.probe_schedules[key]
+            schedule.record_result(True, now)
+
+            # If the new result calls for faster probing, bring the deadline
+            # forward. Never delay a probe that is already due.
+            interval_ns = int(
+                schedule.interval_seconds(now) * 1_000_000_000
+            )
+            proposed_deadline = now + interval_ns
+            self.next_probe_ns[key] = min(
+                self.next_probe_ns[key],
+                proposed_deadline,
+            )
+
             self._refresh_health(now)
 
     def _event(self, event, **fields):
-        record = dict(event=event, monotonic_ns=time.clock_gettime_ns(time.CLOCK_MONOTONIC),
-                      wall_time=datetime.datetime.now(datetime.timezone.utc).isoformat(),
-                      **fields)
+        record = dict(
+            event=event,
+            monotonic_ns=time.clock_gettime_ns(time.CLOCK_MONOTONIC),
+            wall_time=datetime.datetime.now(datetime.timezone.utc).isoformat(),
+            **fields,
+        )
         line = json.dumps(record, sort_keys=True)
         self.logger.info(line)
+
         if self.event_file:
             self.event_file.write(line + "\n")
 
     @set_ev_cls(ofp_event.EventOFPSwitchFeatures, CONFIG_DISPATCHER)
     def features(self, event):
         dp = event.msg.datapath
+
         if dp.id not in SWITCHES:
             self._event("unexpected_switch", dpid=dp.id)
             return
+
         self.probes.invalidate_switch(dp.id)
         self.datapaths[dp.id] = dp
         self.policy.connected.add(dp.id)
         self.discovered.discard(dp.id)
+
         parser, ofp = dp.ofproto_parser, dp.ofproto
-        # This is a dedicated lab controller. Clear stale flows on reconnect,
-        # then keep table-miss as drop rather than learning/flooding the loop.
+
+        # Clear stale flows on reconnect, then keep table-miss as drop.
         dp.send_msg(parser.OFPFlowMod(
-            datapath=dp, command=ofp.OFPFC_DELETE, table_id=ofp.OFPTT_ALL,
-            out_port=ofp.OFPP_ANY, out_group=ofp.OFPG_ANY,
-            match=parser.OFPMatch()))
+            datapath=dp,
+            command=ofp.OFPFC_DELETE,
+            table_id=ofp.OFPTT_ALL,
+            out_port=ofp.OFPP_ANY,
+            out_group=ofp.OFPG_ANY,
+            match=parser.OFPMatch(),
+        ))
+
         dp.send_msg(parser.OFPFlowMod(
-            datapath=dp, cookie=COOKIE, priority=0,
-            match=parser.OFPMatch(), instructions=[]))
-        # Probe rules are independent of the chosen application path. Transit
-        # crosses the branch; edge ingress punts to the controller, never hosts.
+            datapath=dp,
+            cookie=COOKIE,
+            priority=0,
+            match=parser.OFPMatch(),
+            instructions=[],
+        ))
+
+        # Probe rules are independent of the selected application path.
         probe_ports = (1, 2) if dp.id in (2, 3, 4) else (2, 3)
+
         for in_port in probe_ports:
-            out_port = 3 - in_port if dp.id in (2, 3) else ofp.OFPP_CONTROLLER
-            action = parser.OFPActionOutput(out_port, ofp.OFPCML_NO_BUFFER)
+            out_port = (
+                3 - in_port if dp.id in (2, 3)
+                else ofp.OFPP_CONTROLLER
+            )
+            action = parser.OFPActionOutput(
+                out_port, ofp.OFPCML_NO_BUFFER
+            )
+
             dp.send_msg(parser.OFPFlowMod(
-                datapath=dp, cookie=COOKIE, priority=200,
-                match=parser.OFPMatch(in_port=in_port, eth_type=ETHERTYPE),
-                instructions=[parser.OFPInstructionActions(ofp.OFPIT_APPLY_ACTIONS, [action])]))
+                datapath=dp,
+                cookie=COOKIE,
+                priority=200,
+                match=parser.OFPMatch(
+                    in_port=in_port,
+                    eth_type=ETHERTYPE,
+                ),
+                instructions=[
+                    parser.OFPInstructionActions(
+                        ofp.OFPIT_APPLY_ACTIONS,
+                        [action],
+                    )
+                ],
+            ))
+
         dp.send_msg(parser.OFPPortDescStatsRequest(dp, 0))
         self._event("switch_connected", dpid=dp.id)
 
     @staticmethod
     def _port_up(dp, port):
         ofp = dp.ofproto
-        return not (port.state & ofp.OFPPS_LINK_DOWN
-                    or port.config & ofp.OFPPC_PORT_DOWN)
+        return not (
+            port.state & ofp.OFPPS_LINK_DOWN
+            or port.config & ofp.OFPPC_PORT_DOWN
+        )
 
     @set_ev_cls(ofp_event.EventOFPPortDescStatsReply, MAIN_DISPATCHER)
     def port_description(self, event):
         msg, dp = event.msg, event.msg.datapath
+
         if dp.id not in self.datapaths:
             return
+
         for port in msg.body:
             self.policy.ports[(dp.id, port.port_no)] = self._port_up(dp, port)
+
         if msg.flags & dp.ofproto.OFPMPF_REPLY_MORE:
             return
+
         self.discovered.add(dp.id)
-        self._reconcile("startup" if not self.started else "switch_up", force=True)
+        self._reconcile(
+            "startup" if not self.started else "switch_up",
+            force=True,
+        )
 
     @set_ev_cls(ofp_event.EventOFPPortStatus, MAIN_DISPATCHER)
     def port_status(self, event):
         msg, dp = event.msg, event.msg.datapath
+
         if dp.id not in self.datapaths:
             return
+
         key = (dp.id, msg.desc.port_no)
-        up = msg.reason != dp.ofproto.OFPPR_DELETE and self._port_up(dp, msg.desc)
+        up = (
+            msg.reason != dp.ofproto.OFPPR_DELETE
+            and self._port_up(dp, msg.desc)
+        )
         previous = self.policy.ports.get(key)
         self.policy.ports[key] = up
+
         if previous == up:
             return
+
         cause = "link_up" if up else "link_down"
+
         self.probes.invalidate_port(dp.id, msg.desc.port_no)
-        self._refresh_health(time.clock_gettime_ns(time.CLOCK_MONOTONIC))
+        self._refresh_health(
+            time.clock_gettime_ns(time.CLOCK_MONOTONIC)
+        )
+
         self._event(cause, dpid=dp.id, port=msg.desc.port_no)
         self._reconcile(cause)
 
     @set_ev_cls(ofp_event.EventOFPStateChange, DEAD_DISPATCHER)
     def disconnected(self, event):
         dp = event.datapath
+
         if self.datapaths.get(dp.id) is not dp:
             return
+
         self.probes.invalidate_switch(dp.id)
         del self.datapaths[dp.id]
         self.policy.connected.discard(dp.id)
         self.discovered.discard(dp.id)
+
         self._event("switch_down", dpid=dp.id)
         self._reconcile("switch_down", force=True)
 
@@ -190,41 +352,122 @@ def _reconcile(self, cause, force=False):
             if self.discovered != SWITCHES:
                 return
             self.started = True
+
         path = self.policy.choose(self.path)
+
         if not force and path == self.path:
             return
+
         now = time.clock_gettime_ns(time.CLOCK_MONOTONIC)
+
         if cause == "health_update" and path != self.path:
             if self.pending or now - self.last_route_change_ns < 2_000_000_000:
-                self._event("route_decision", selected=self.path, candidate=path, reason="hold_down_or_barrier",
-                            health=self.policy.health)
+                self._event(
+                    "route_decision",
+                    selected=self.path,
+                    candidate=path,
+                    reason="hold_down_or_barrier",
+                    health=self.policy.health,
+                )
                 return
+
         self.last_route_change_ns = now
-        self._event("route_decision", previous=self.path, selected=path, reason=cause,
-                    health=self.policy.health)
+        self._event(
+            "route_decision",
+            previous=self.path,
+            selected=path,
+            reason=cause,
+            health=self.policy.health,
+        )
+
         self.path = path
         self.generation += 1
         self.cause = cause
         self.pending.clear()
         self.failed_generation = False
-        self._event("route_installing", path=path, cause=cause,
-                    generation=self.generation, health=self.policy.health)
+
+        self._event(
+            "route_installing",
+            path=path,
+            cause=cause,
+            generation=self.generation,
+            health=self.policy.health,
+        )
+
         rules = forwarding_rules(path)
-        # Each ADD replaces the previous identical match/priority atomically
-        # within a switch. During an update, asymmetrical forward/reverse paths
-        # still reach their host: branch ingress never sends into another branch.
+
+        # Install the selected route and remove stale multipath rules.
         for dpid, dp in sorted(self.datapaths.items()):
             parser, ofp = dp.ofproto_parser, dp.ofproto
+
             for rule in rules[dpid]:
-                actions = ([] if rule.out_port is None else
-                           [parser.OFPActionOutput(rule.out_port)])
-                instructions = [parser.OFPInstructionActions(
-                    ofp.OFPIT_APPLY_ACTIONS, actions)]
+                actions = (
+                    [] if rule.out_port is None
+                    else [parser.OFPActionOutput(rule.out_port)]
+                )
+                instructions = [
+                    parser.OFPInstructionActions(
+                        ofp.OFPIT_APPLY_ACTIONS,
+                        actions,
+                    )
+                ]
+
                 dp.send_msg(parser.OFPFlowMod(
-                    datapath=dp, cookie=COOKIE, priority=100,
-                    match=parser.OFPMatch(in_port=rule.in_port,
-                                          eth_type=rule.eth_type),
-                    instructions=instructions))
+                    datapath=dp,
+                    cookie=COOKIE,
+                    priority=100,
+                    match=parser.OFPMatch(
+                        in_port=rule.in_port,
+                        eth_type=rule.eth_type,
+                    ),
+                    instructions=instructions,
+                ))
+
+            # Remove this controller's previous DSCP-specific rules.
+            dp.send_msg(parser.OFPFlowMod(
+                datapath=dp,
+                cookie=MULTIPATH_COOKIE,
+                cookie_mask=0xffffffffffffffff,
+                table_id=0,
+                command=ofp.OFPFC_DELETE,
+                out_port=ofp.OFPP_ANY,
+                out_group=ofp.OFPG_ANY,
+                match=parser.OFPMatch(),
+            ))
+
+            # DATA uses DSCP 8 for upper and DSCP 16 for lower.
+            multipath_paths = {
+                "upper": (8, ((1, 1, 2), (2, 1, 2), (4, 1, 3))),
+                "lower": (16, ((1, 1, 3), (3, 1, 2), (4, 2, 3))),
+            }
+
+            for path_name, (dscp, entries) in multipath_paths.items():
+                if not self.policy.available(path_name):
+                    continue
+
+                for switch_id, ingress, output in entries:
+                    if switch_id != dpid:
+                        continue
+
+                    instructions = [
+                        parser.OFPInstructionActions(
+                            ofp.OFPIT_APPLY_ACTIONS,
+                            [parser.OFPActionOutput(output)],
+                        )
+                    ]
+
+                    dp.send_msg(parser.OFPFlowMod(
+                        datapath=dp,
+                        cookie=MULTIPATH_COOKIE,
+                        priority=150,
+                        match=parser.OFPMatch(
+                            in_port=ingress,
+                            eth_type=0x0800,
+                            ip_dscp=dscp,
+                        ),
+                        instructions=instructions,
+                    ))
+
             barrier = parser.OFPBarrierRequest(dp)
             dp.set_xid(barrier)
             self.pending[(dpid, barrier.xid)] = self.generation
@@ -234,20 +477,41 @@ def _reconcile(self, cause, force=False):
     def barrier_reply(self, event):
         key = (event.msg.datapath.id, event.msg.xid)
         generation = self.pending.pop(key, None)
-        if generation != self.generation or self.pending or self.failed_generation:
+
+        if (
+            generation != self.generation
+            or self.pending
+            or self.failed_generation
+        ):
             return
-        self._event("route_ready" if self.path else "route_unavailable",
-                    path=self.path, cause=self.cause, generation=self.generation)
 
-    @set_ev_cls(ofp_event.EventOFPErrorMsg, [CONFIG_DISPATCHER, MAIN_DISPATCHER])
+        self._event(
+            "route_ready" if self.path else "route_unavailable",
+            path=self.path,
+            cause=self.cause,
+            generation=self.generation,
+        )
+
+    @set_ev_cls(
+        ofp_event.EventOFPErrorMsg,
+        [CONFIG_DISPATCHER, MAIN_DISPATCHER],
+    )
     def openflow_error(self, event):
         msg = event.msg
         self.failed_generation = True
-        self._event("openflow_error", dpid=msg.datapath.id,
-                    error_type=msg.type, code=msg.code, xid=msg.xid)
+
+        self._event(
+            "openflow_error",
+            dpid=msg.datapath.id,
+            error_type=msg.type,
+            code=msg.code,
+            xid=msg.xid,
+        )
 
     def close(self):
         hub.kill(self.monitor)
+
         if self.event_file:
             self.event_file.close()
+
         super().close()
diff --git a/controller/multipath.py b/controller/multipath.py
new file mode 100644
index 0000000..f54233a
--- /dev/null
+++ b/controller/multipath.py
@@ -0,0 +1,75 @@
+"""Multipath scheduling logic for Reliable UDP File Transfer.
+
+Selects a path for each outgoing data packet using weighted scheduling.
+This module is independent of OS-Ken so its scheduling logic is testable.
+"""
+
+from dataclasses import dataclass
+from itertools import cycle
+from threading import Lock
+
+
+@dataclass(frozen=True)
+class PathState:
+    """Current scheduling state for a network path."""
+
+    name: str
+    healthy: bool = True
+    weight: int = 1
+
+
+class MultipathScheduler:
+    """Thread-safe weighted scheduler for healthy network paths."""
+
+    def __init__(self, paths=("upper", "lower")):
+        if not paths:
+            raise ValueError("At least one path is required")
+
+        self._lock = Lock()
+        self._paths = {
+            name: PathState(name=name)
+            for name in paths
+        }
+        self._schedule = cycle(paths)
+
+    def update_path(self, name, *, healthy=None, weight=None):
+        """Update a path's health or scheduling weight."""
+        with self._lock:
+            if name not in self._paths:
+                raise ValueError(f"Unknown path: {name}")
+
+            current = self._paths[name]
+            new_weight = current.weight if weight is None else weight
+
+            if new_weight < 1:
+                raise ValueError("Path weight must be at least 1")
+
+            self._paths[name] = PathState(
+                name=name,
+                healthy=current.healthy if healthy is None else healthy,
+                weight=new_weight,
+            )
+            self._rebuild_schedule()
+
+    def _rebuild_schedule(self):
+        """Build a weighted cycle from currently healthy paths."""
+        entries = []
+
+        for path in self._paths.values():
+            if path.healthy:
+                entries.extend([path.name] * path.weight)
+
+        self._schedule = cycle(entries)
+
+    def select_path(self):
+        """Return the next healthy path, or None if all paths are down."""
+        with self._lock:
+            if not any(path.healthy for path in self._paths.values()):
+                return None
+
+            return next(self._schedule)
+
+    def get_paths(self):
+        """Return a snapshot of the configured path states."""
+        with self._lock:
+            return dict(self._paths)
diff --git a/controller/probe_schedule.py b/controller/probe_schedule.py
new file mode 100644
index 0000000..ac97b15
--- /dev/null
+++ b/controller/probe_schedule.py
@@ -0,0 +1,73 @@
+"""Adaptive scheduling intervals for SDN health probes."""
+
+
+class AdaptiveProbeSchedule:
+    """Choose probe frequency based on recent results and path health.
+
+    Normal operation:       1000 ms
+    Uncertain/degraded:      250 ms
+    After a probe failure:   100 ms until three consecutive successes
+    Recovery period:         250 ms for two seconds
+    """
+
+    STABLE_NS = 1_000_000_000
+    UNCERTAIN_NS = 250_000_000
+    FAILURE_NS = 100_000_000
+    RECOVERY_NS = 250_000_000
+    RECOVERY_HOLD_NS = 2_000_000_000
+    SUCCESS_THRESHOLD = 3
+
+    def __init__(self):
+        self.degraded = False
+        self.consecutive_successes = 0
+        self.recovery_started_ns = None
+        self.failure_seen = False
+
+    def record_result(self, success, now_ns):
+        """Record a completed probe result using monotonic nanoseconds."""
+        if not success:
+            self.failure_seen = True
+            self.consecutive_successes = 0
+            self.recovery_started_ns = None
+            return
+
+        self.consecutive_successes += 1
+
+        if (
+            self.failure_seen
+            and self.consecutive_successes >= self.SUCCESS_THRESHOLD
+            and self.recovery_started_ns is None
+        ):
+            self.recovery_started_ns = now_ns
+
+    def set_degraded(self, degraded, now_ns):
+        """Mark whether current path health is uncertain or degraded."""
+        degraded = bool(degraded)
+
+        if degraded:
+            self.degraded = True
+            self.recovery_started_ns = None
+        elif self.degraded:
+            self.degraded = False
+            self.recovery_started_ns = now_ns
+
+    def interval_ns(self, now_ns):
+        """Return the appropriate interval in nanoseconds."""
+        if self.failure_seen and self.consecutive_successes < self.SUCCESS_THRESHOLD:
+            return self.FAILURE_NS
+
+        if self.degraded:
+            return self.UNCERTAIN_NS
+
+        if self.recovery_started_ns is not None:
+            if now_ns - self.recovery_started_ns < self.RECOVERY_HOLD_NS:
+                return self.RECOVERY_NS
+
+            self.failure_seen = False
+            self.recovery_started_ns = None
+
+        return self.STABLE_NS
+
+    def interval_seconds(self, now_ns):
+        """Return the interval in seconds for scheduler integrations."""
+        return self.interval_ns(now_ns) / 1_000_000_000
diff --git a/src/client.c b/src/client.c
index 8596184..8116fcb 100644
--- a/src/client.c
+++ b/src/client.c
@@ -1,5 +1,6 @@
 #include "utils.h"
 #include <arpa/inet.h>
+#include <netinet/in.h>
 #include <getopt.h>
 #include <math.h>
 #include <time.h>
@@ -68,17 +69,34 @@ static int exchange(int fd, const struct sockaddr_in *server,
 }
 
 struct transport_options {
-    int adaptive, sack, congestion;
+    int adaptive, sack, congestion, multipath;
     double drop_rate;
 };
 
 /* Simulate loss after the send decision so normal ACK/retry logic observes
  * exactly the same silence it would see after network DATA loss. */
 static int send_data(int fd, const struct packet *packet,
-                     const struct sockaddr_in *peer, double drop_rate)
+                     const struct sockaddr_in *peer, double drop_rate,
+                     int multipath)
 {
     if (drop_rate > 0 && rand() / (RAND_MAX + 1.0) < drop_rate) return 0;
-    return send_packet(fd, packet, peer);
+
+    if (!multipath)
+        return send_packet(fd, packet, peer);
+
+    /* DSCP 8 selects upper; DSCP 16 selects lower.
+     * Sequence parity keeps retransmissions on their original branch. */
+    int tos = (packet->sequence % 2 == 0 ? 8 : 16) << 2;
+    if (setsockopt(fd, IPPROTO_IP, IP_TOS, &tos, sizeof(tos)) != 0)
+        return -1;
+
+    int result = send_packet(fd, packet, peer);
+    int reset_tos = 0;
+    if (setsockopt(fd, IPPROTO_IP, IP_TOS,
+                   &reset_tos, sizeof(reset_tos)) != 0 && result == 0)
+        return -1;
+
+    return result;
 }
 
 static int send_file(int fd, FILE *file, const struct sockaddr_in *server,
@@ -109,7 +127,7 @@ static int send_file(int fd, FILE *file, const struct sockaddr_in *server,
                 fprintf(stderr, "Input file changed or could not be read.\n");
                 return -1;
             }
-            if (send_data(fd, &slot->packet, server, options.drop_rate) != 0) {
+            if (send_data(fd, &slot->packet, server, options.drop_rate, options.multipath) != 0) {
                 perror("send DATA");
                 return -1;
             }
@@ -133,7 +151,7 @@ static int send_file(int fd, FILE *file, const struct sockaddr_in *server,
                     fprintf(stderr, "Retry limit reached for DATA %" PRIu32 ".\n", sequence);
                     return -1;
                 }
-                if (send_data(fd, &slot->packet, server, options.drop_rate) != 0) {
+                if (send_data(fd, &slot->packet, server, options.drop_rate, options.multipath) != 0) {
                     perror("retransmit DATA");
                     return -1;
                 }
@@ -235,7 +253,7 @@ static int send_file(int fd, FILE *file, const struct sockaddr_in *server,
                     struct send_slot *slot = &slots[seq % WINDOW_SIZE];
                     if (slot->acknowledged || slot->fast_sent || slot->gap_reports < 3) continue;
                     if (slot->retries >= MAX_RETRIES) return -1;
-                    if (send_data(fd, &slot->packet, server, options.drop_rate) != 0) return -1;
+                    if (send_data(fd, &slot->packet, server, options.drop_rate, options.multipath) != 0) return -1;
                     slot->fast_sent = 1;
                     slot->sent_at = monotonic_ms();
                     ++slot->retries; ++stats->data_sends; ++stats->data_retries;
@@ -280,12 +298,12 @@ int main(int argc, char *argv[])
     uint16_t port;
     const char *host = NULL, *port_text = NULL, *filename = NULL;
     const char *rto_mode = getenv("RUDP_RTO");
-    struct transport_options options = {0, option_enabled("RUDP_SACK"), option_enabled("RUDP_CC"), 0};
+    struct transport_options options = {0, option_enabled("RUDP_SACK"), option_enabled("RUDP_CC"), option_enabled("RUDP_MULTIPATH"), 0};
     static const struct option cli[] = {
         {"host", required_argument, NULL, 'H'}, {"port", required_argument, NULL, 'p'},
         {"file", required_argument, NULL, 'f'}, {"sack", no_argument, NULL, 's'},
         {"no-sack", no_argument, NULL, 'S'}, {"cc", no_argument, NULL, 'c'},
-        {"no-cc", no_argument, NULL, 'C'}, {"adaptive-rto", no_argument, NULL, 'a'},
+        {"no-cc", no_argument, NULL, 'C'}, {"multipath", no_argument, NULL, 'm'}, {"adaptive-rto", no_argument, NULL, 'a'},
         {"fixed-rto", no_argument, NULL, 'F'}, {"drop-rate", required_argument, NULL, 'd'},
         {"help", no_argument, NULL, 'h'}, {NULL, 0, NULL, 0}
     };
@@ -299,6 +317,7 @@ int main(int argc, char *argv[])
         case 'F': rto_mode = "fixed"; break;
         case 's': case 'S': options.sack = opt == 's'; break;
         case 'c': case 'C': options.congestion = opt == 'c'; break;
+        case 'm': options.multipath = 1; break;
         case 'd': {
             char *end;
             options.drop_rate = strtod(optarg, &end);
@@ -313,6 +332,7 @@ int main(int argc, char *argv[])
         case 'h':
             puts("Usage: client HOST PORT FILE | client --host IPv4 --port PORT --file FILE\n"
                  "Options: --sack | --no-sack, --adaptive-rto | --fixed-rto, --cc | --no-cc\n"
+                 "         --multipath    distribute DATA across upper/lower SDN paths\n"
                  "         --drop-rate P   simulate loss on each outgoing DATA send (0..1)");
             return EXIT_SUCCESS;
         default: return EXIT_FAILURE;
diff --git a/tests/test_multipath.py b/tests/test_multipath.py
new file mode 100644
index 0000000..f4f4788
--- /dev/null
+++ b/tests/test_multipath.py
@@ -0,0 +1,38 @@
+import unittest
+
+from controller.multipath import MultipathScheduler
+
+
+class MultipathTests(unittest.TestCase):
+    def test_alternates_between_healthy_paths(self):
+        scheduler = MultipathScheduler()
+        selected = [scheduler.select_path() for _ in range(6)]
+        self.assertEqual(
+            selected,
+            ["upper", "lower", "upper", "lower", "upper", "lower"],
+        )
+
+    def test_unhealthy_path_is_not_selected(self):
+        scheduler = MultipathScheduler()
+        scheduler.update_path("upper", healthy=False)
+        self.assertEqual(
+            [scheduler.select_path() for _ in range(4)],
+            ["lower", "lower", "lower", "lower"],
+        )
+
+    def test_weighted_schedule(self):
+        scheduler = MultipathScheduler()
+        scheduler.update_path("upper", weight=2)
+        selected = [scheduler.select_path() for _ in range(6)]
+        self.assertEqual(selected.count("upper"), 4)
+        self.assertEqual(selected.count("lower"), 2)
+
+    def test_all_paths_unhealthy(self):
+        scheduler = MultipathScheduler()
+        scheduler.update_path("upper", healthy=False)
+        scheduler.update_path("lower", healthy=False)
+        self.assertIsNone(scheduler.select_path())
+
+
+if __name__ == "__main__":
+    unittest.main()
diff --git a/tests/test_probe_schedule.py b/tests/test_probe_schedule.py
new file mode 100644
index 0000000..cf924e5
--- /dev/null
+++ b/tests/test_probe_schedule.py
@@ -0,0 +1,87 @@
+"""Tests for adaptive probe scheduling."""
+
+import unittest
+
+from controller.probe_schedule import AdaptiveProbeSchedule
+
+
+class AdaptiveProbeScheduleTests(unittest.TestCase):
+    def setUp(self):
+        self.schedule = AdaptiveProbeSchedule()
+        self.now = 10_000_000_000
+
+    def test_starts_at_stable_interval(self):
+        self.assertEqual(
+            self.schedule.interval_ns(self.now),
+            AdaptiveProbeSchedule.STABLE_NS,
+        )
+
+    def test_failure_increases_probe_frequency(self):
+        self.schedule.record_result(False, self.now)
+
+        self.assertEqual(
+            self.schedule.interval_ns(self.now),
+            AdaptiveProbeSchedule.FAILURE_NS,
+        )
+
+    def test_three_successes_enter_recovery_period(self):
+        self.schedule.record_result(False, self.now)
+
+        for i in range(3):
+            self.schedule.record_result(True, self.now + (i + 1) * 100_000_000)
+
+        recovery_time = self.now + 300_000_000
+
+        self.assertEqual(
+            self.schedule.interval_ns(recovery_time),
+            AdaptiveProbeSchedule.RECOVERY_NS,
+        )
+
+        self.assertEqual(
+            self.schedule.interval_ns(
+                recovery_time + AdaptiveProbeSchedule.RECOVERY_HOLD_NS
+            ),
+            AdaptiveProbeSchedule.STABLE_NS,
+        )
+
+    def test_another_failure_resets_recovery(self):
+        self.schedule.record_result(False, self.now)
+
+        for i in range(3):
+            self.schedule.record_result(True, self.now + (i + 1) * 100_000_000)
+
+        self.schedule.record_result(False, self.now + 400_000_000)
+
+        self.assertEqual(
+            self.schedule.interval_ns(self.now + 400_000_000),
+            AdaptiveProbeSchedule.FAILURE_NS,
+        )
+
+    def test_degraded_health_uses_uncertain_interval(self):
+        self.schedule.set_degraded(True, self.now)
+
+        self.assertEqual(
+            self.schedule.interval_ns(self.now),
+            AdaptiveProbeSchedule.UNCERTAIN_NS,
+        )
+
+    def test_recovered_health_has_temporary_faster_interval(self):
+        self.schedule.set_degraded(True, self.now)
+        self.schedule.set_degraded(False, self.now + 100_000_000)
+
+        self.assertEqual(
+            self.schedule.interval_ns(self.now + 100_000_000),
+            AdaptiveProbeSchedule.RECOVERY_NS,
+        )
+
+        self.assertEqual(
+            self.schedule.interval_ns(
+                self.now + 100_000_000
+                + AdaptiveProbeSchedule.RECOVERY_HOLD_NS
+            ),
+            AdaptiveProbeSchedule.STABLE_NS,
+        )
+
+
+if __name__ == "__main__":
+    unittest.main(verbosity=2)

From 7ace1b2cc7089861b95a5e7b6aeb4ec816bcd300 Mon Sep 17 00:00:00 2001
From: Saraswath BV <saraswathgowda@gmail.com>
Date: Mon, 2 Nov 2026 14:15:05 -0600
Subject: [PATCH 2/4] Implement adaptive probe scheduling

---
 controller/failover.py       |  99 ++++++++++++++++++++++++----
 controller/probe_schedule.py | 111 +++++++++++++------------------
 tests/test_probe_schedule.py | 122 +++++++++++++----------------------
 3 files changed, 175 insertions(+), 157 deletions(-)

diff --git a/controller/failover.py b/controller/failover.py
index fe0e8ef..42c22dc 100644
--- a/controller/failover.py
+++ b/controller/failover.py
@@ -1,3 +1,4 @@
+
 """Proactive OpenFlow 1.3 routes with port-status-driven path recovery.
 
 Run with osken-manager --ofp-tcp-listen-port 6653 controller/failover.py.
@@ -32,6 +33,7 @@
 
 # The manager may load this file by path without adding the repository root.
 sys.path.insert(0, str(Path(__file__).resolve().parents[1]))
+
 from controller.routing import PathPolicy, SWITCHES, forwarding_rules
 from controller.probes import ProbeTracker, ENDPOINTS, ETHERTYPE
 from controller.probe_schedule import AdaptiveProbeSchedule
@@ -46,10 +48,12 @@ class FailoverController(AppBase):
 
     def __init__(self, *args, **kwargs):
         super().__init__(*args, **kwargs)
+
         self.policy = PathPolicy(
             os.environ.get("RUDP_MODE", "failover"),
             os.environ.get("RUDP_INITIAL_PATH", "upper"),
         )
+
         self.datapaths = {}
         self.discovered = set()
         self.started = False
@@ -60,11 +64,14 @@ def __init__(self, *args, **kwargs):
         self.cause = "startup"
 
         event_path = os.environ.get("RUDP_EVENT_LOG")
-        self.event_file = open(event_path, "a", buffering=1) if event_path else None
+        self.event_file = (
+            open(event_path, "a", buffering=1) if event_path else None
+        )
 
         self._event("controller_started", mode=self.policy.mode)
 
         self.probes = ProbeTracker()
+
         self.probe_schedules = {
             key: AdaptiveProbeSchedule() for key in ENDPOINTS
         }
@@ -94,7 +101,11 @@ def _monitor(self):
             for observation in self.probes.expire(now):
                 self._event("probe_result", **observation)
 
-                key = (observation["path"], observation["direction"])
+                key = (
+                    observation["path"],
+                    observation["direction"],
+                )
+
                 self.probe_schedules[key].record_result(False, now)
                 self.next_probe_ns[key] = now
 
@@ -109,18 +120,25 @@ def _monitor(self):
                     hysteresis=0.02,
                     hold_down_seconds=2,
                 )
+
                 self._reconcile("health_update")
 
                 # Send only the probes whose individual deadlines have arrived.
-                for (path, direction), (source, out_port, _, _) in ENDPOINTS.items():
+                for (
+                    (path, direction),
+                    (source, out_port, _, _),
+                ) in ENDPOINTS.items():
+
                     if not self.policy.available(path):
                         continue
 
                     dp = self.datapaths.get(source)
+
                     if dp is None:
                         continue
 
                     key = (path, direction)
+
                     if now < self.next_probe_ns[key]:
                         continue
 
@@ -130,7 +148,9 @@ def _monitor(self):
                         datapath=dp,
                         buffer_id=dp.ofproto.OFP_NO_BUFFER,
                         in_port=dp.ofproto.OFPP_CONTROLLER,
-                        actions=[dp.ofproto_parser.OFPActionOutput(out_port)],
+                        actions=[
+                            dp.ofproto_parser.OFPActionOutput(out_port)
+                        ],
                         data=frame,
                     ))
 
@@ -138,6 +158,7 @@ def _monitor(self):
                         self.probe_schedules[key].interval_seconds(now)
                         * 1_000_000_000
                     )
+
                     self.next_probe_ns[key] = now + interval_ns
 
             # Wake at the next probe deadline or pending-probe timeout.
@@ -145,18 +166,25 @@ def _monitor(self):
             deadlines = []
 
             if self.started:
-                for (path, direction), (source, _, _, _) in ENDPOINTS.items():
+                for (
+                    (path, direction),
+                    (source, _, _, _),
+                ) in ENDPOINTS.items():
+
                     if not self.policy.available(path):
                         continue
+
                     if source not in self.datapaths:
                         continue
 
                     deadline = self.next_probe_ns[(path, direction)]
+
                     if deadline > now:
                         deadlines.append(deadline)
 
                 for _, _, sent_ns in self.probes.pending.values():
                     timeout_deadline = sent_ns + self.probes.timeout_ns
+
                     if timeout_deadline > now:
                         deadlines.append(timeout_deadline)
 
@@ -183,7 +211,11 @@ def probe_received(self, event):
         if observation:
             self._event("probe_result", **observation)
 
-            key = (observation["path"], observation["direction"])
+            key = (
+                observation["path"],
+                observation["direction"],
+            )
+
             schedule = self.probe_schedules[key]
             schedule.record_result(True, now)
 
@@ -192,7 +224,9 @@ def probe_received(self, event):
             interval_ns = int(
                 schedule.interval_seconds(now) * 1_000_000_000
             )
+
             proposed_deadline = now + interval_ns
+
             self.next_probe_ns[key] = min(
                 self.next_probe_ns[key],
                 proposed_deadline,
@@ -204,9 +238,12 @@ def _event(self, event, **fields):
         record = dict(
             event=event,
             monotonic_ns=time.clock_gettime_ns(time.CLOCK_MONOTONIC),
-            wall_time=datetime.datetime.now(datetime.timezone.utc).isoformat(),
+            wall_time=datetime.datetime.now(
+                datetime.timezone.utc
+            ).isoformat(),
             **fields,
         )
+
         line = json.dumps(record, sort_keys=True)
         self.logger.info(line)
 
@@ -222,6 +259,7 @@ def features(self, event):
             return
 
         self.probes.invalidate_switch(dp.id)
+
         self.datapaths[dp.id] = dp
         self.policy.connected.add(dp.id)
         self.discovered.discard(dp.id)
@@ -254,8 +292,10 @@ def features(self, event):
                 3 - in_port if dp.id in (2, 3)
                 else ofp.OFPP_CONTROLLER
             )
+
             action = parser.OFPActionOutput(
-                out_port, ofp.OFPCML_NO_BUFFER
+                out_port,
+                ofp.OFPCML_NO_BUFFER,
             )
 
             dp.send_msg(parser.OFPFlowMod(
@@ -280,6 +320,7 @@ def features(self, event):
     @staticmethod
     def _port_up(dp, port):
         ofp = dp.ofproto
+
         return not (
             port.state & ofp.OFPPS_LINK_DOWN
             or port.config & ofp.OFPPC_PORT_DOWN
@@ -293,12 +334,15 @@ def port_description(self, event):
             return
 
         for port in msg.body:
-            self.policy.ports[(dp.id, port.port_no)] = self._port_up(dp, port)
+            self.policy.ports[(dp.id, port.port_no)] = (
+                self._port_up(dp, port)
+            )
 
         if msg.flags & dp.ofproto.OFPMPF_REPLY_MORE:
             return
 
         self.discovered.add(dp.id)
+
         self._reconcile(
             "startup" if not self.started else "switch_up",
             force=True,
@@ -312,10 +356,12 @@ def port_status(self, event):
             return
 
         key = (dp.id, msg.desc.port_no)
+
         up = (
             msg.reason != dp.ofproto.OFPPR_DELETE
             and self._port_up(dp, msg.desc)
         )
+
         previous = self.policy.ports.get(key)
         self.policy.ports[key] = up
 
@@ -325,6 +371,7 @@ def port_status(self, event):
         cause = "link_up" if up else "link_down"
 
         self.probes.invalidate_port(dp.id, msg.desc.port_no)
+
         self._refresh_health(
             time.clock_gettime_ns(time.CLOCK_MONOTONIC)
         )
@@ -340,6 +387,7 @@ def disconnected(self, event):
             return
 
         self.probes.invalidate_switch(dp.id)
+
         del self.datapaths[dp.id]
         self.policy.connected.discard(dp.id)
         self.discovered.discard(dp.id)
@@ -351,6 +399,7 @@ def _reconcile(self, cause, force=False):
         if not self.started:
             if self.discovered != SWITCHES:
                 return
+
             self.started = True
 
         path = self.policy.choose(self.path)
@@ -361,7 +410,10 @@ def _reconcile(self, cause, force=False):
         now = time.clock_gettime_ns(time.CLOCK_MONOTONIC)
 
         if cause == "health_update" and path != self.path:
-            if self.pending or now - self.last_route_change_ns < 2_000_000_000:
+            if (
+                self.pending
+                or now - self.last_route_change_ns < 2_000_000_000
+            ):
                 self._event(
                     "route_decision",
                     selected=self.path,
@@ -372,6 +424,7 @@ def _reconcile(self, cause, force=False):
                 return
 
         self.last_route_change_ns = now
+
         self._event(
             "route_decision",
             previous=self.path,
@@ -405,6 +458,7 @@ def _reconcile(self, cause, force=False):
                     [] if rule.out_port is None
                     else [parser.OFPActionOutput(rule.out_port)]
                 )
+
                 instructions = [
                     parser.OFPInstructionActions(
                         ofp.OFPIT_APPLY_ACTIONS,
@@ -437,12 +491,30 @@ def _reconcile(self, cause, force=False):
 
             # DATA uses DSCP 8 for upper and DSCP 16 for lower.
             multipath_paths = {
-                "upper": (8, ((1, 1, 2), (2, 1, 2), (4, 1, 3))),
-                "lower": (16, ((1, 1, 3), (3, 1, 2), (4, 2, 3))),
+                "upper": (
+                    8,
+                    ((1, 1, 2), (2, 1, 2), (4, 1, 3)),
+                ),
+                "lower": (
+                    16,
+                    ((1, 1, 3), (3, 1, 2), (4, 2, 3)),
+                ),
             }
 
             for path_name, (dscp, entries) in multipath_paths.items():
-                if not self.policy.available(path_name):
+                details = self.policy.availability_details(path_name)
+
+                self._event(
+                    "multipath_path_check",
+                    dpid=dpid,
+                    path=path_name,
+                    dscp=dscp,
+                    available=details["available"],
+                    missing_switches=details["missing_switches"],
+                    unavailable_ports=details["unavailable_ports"],
+                )
+
+                if not details["available"]:
                     continue
 
                 for switch_id, ingress, output in entries:
@@ -470,6 +542,7 @@ def _reconcile(self, cause, force=False):
 
             barrier = parser.OFPBarrierRequest(dp)
             dp.set_xid(barrier)
+
             self.pending[(dpid, barrier.xid)] = self.generation
             dp.send_msg(barrier)
 
diff --git a/controller/probe_schedule.py b/controller/probe_schedule.py
index ac97b15..882684d 100644
--- a/controller/probe_schedule.py
+++ b/controller/probe_schedule.py
@@ -1,73 +1,52 @@
-"""Adaptive scheduling intervals for SDN health probes."""
+"""Adaptive per-path, per-direction probe scheduling."""
+import time
 
 
 class AdaptiveProbeSchedule:
-    """Choose probe frequency based on recent results and path health.
-
-    Normal operation:       1000 ms
-    Uncertain/degraded:      250 ms
-    After a probe failure:   100 ms until three consecutive successes
-    Recovery period:         250 ms for two seconds
-    """
-
-    STABLE_NS = 1_000_000_000
-    UNCERTAIN_NS = 250_000_000
-    FAILURE_NS = 100_000_000
-    RECOVERY_NS = 250_000_000
-    RECOVERY_HOLD_NS = 2_000_000_000
-    SUCCESS_THRESHOLD = 3
-
-    def __init__(self):
-        self.degraded = False
-        self.consecutive_successes = 0
-        self.recovery_started_ns = None
-        self.failure_seen = False
-
-    def record_result(self, success, now_ns):
-        """Record a completed probe result using monotonic nanoseconds."""
-        if not success:
-            self.failure_seen = True
-            self.consecutive_successes = 0
-            self.recovery_started_ns = None
-            return
-
-        self.consecutive_successes += 1
-
-        if (
-            self.failure_seen
-            and self.consecutive_successes >= self.SUCCESS_THRESHOLD
-            and self.recovery_started_ns is None
-        ):
-            self.recovery_started_ns = now_ns
-
-    def set_degraded(self, degraded, now_ns):
-        """Mark whether current path health is uncertain or degraded."""
-        degraded = bool(degraded)
-
-        if degraded:
+    def __init__(
+        self,
+        min_interval=0.1,
+        max_interval=2.0,
+        initial_interval=0.25,
+        healthy_threshold=0.05,
+        degraded_threshold=0.10,
+    ):
+        if not 0 < min_interval <= initial_interval <= max_interval:
+            raise ValueError("Invalid probe interval bounds")
+        if not 0 <= healthy_threshold < degraded_threshold:
+            raise ValueError("Invalid health thresholds")
+
+        self.min_interval = min_interval
+        self.max_interval = max_interval
+        self.healthy_threshold = healthy_threshold
+        self.degraded_threshold = degraded_threshold
+        self.interval = initial_interval
+        self.degraded = True
+        self.last_result_ns = None
+
+    def set_degraded(self, degraded, now_ns=None):
+        """Update health state and optionally record its update time."""
+        self.degraded = bool(degraded)
+        if now_ns is not None:
+            self.last_result_ns = now_ns
+
+    def record_result(self, success, now_ns=None):
+        """Record a probe result and adapt the next interval."""
+        if now_ns is None:
+            now_ns = time.monotonic_ns()
+        self.last_result_ns = now_ns
+
+        if success:
+            if not self.degraded:
+                self.interval = min(self.max_interval, self.interval * 1.25)
+        else:
             self.degraded = True
-            self.recovery_started_ns = None
-        elif self.degraded:
-            self.degraded = False
-            self.recovery_started_ns = now_ns
+            self.interval = max(self.min_interval, self.interval / 2)
 
-    def interval_ns(self, now_ns):
-        """Return the appropriate interval in nanoseconds."""
-        if self.failure_seen and self.consecutive_successes < self.SUCCESS_THRESHOLD:
-            return self.FAILURE_NS
+        return self.interval
 
+    def interval_seconds(self, now_ns=None):
+        """Return the interval used by the controller's next deadline."""
         if self.degraded:
-            return self.UNCERTAIN_NS
-
-        if self.recovery_started_ns is not None:
-            if now_ns - self.recovery_started_ns < self.RECOVERY_HOLD_NS:
-                return self.RECOVERY_NS
-
-            self.failure_seen = False
-            self.recovery_started_ns = None
-
-        return self.STABLE_NS
-
-    def interval_seconds(self, now_ns):
-        """Return the interval in seconds for scheduler integrations."""
-        return self.interval_ns(now_ns) / 1_000_000_000
+            return max(self.min_interval, min(self.interval, 0.25))
+        return min(self.max_interval, self.interval)
diff --git a/tests/test_probe_schedule.py b/tests/test_probe_schedule.py
index cf924e5..14ba95b 100644
--- a/tests/test_probe_schedule.py
+++ b/tests/test_probe_schedule.py
@@ -1,87 +1,53 @@
-"""Tests for adaptive probe scheduling."""
-
 import unittest
 
 from controller.probe_schedule import AdaptiveProbeSchedule
 
 
 class AdaptiveProbeScheduleTests(unittest.TestCase):
-    def setUp(self):
-        self.schedule = AdaptiveProbeSchedule()
-        self.now = 10_000_000_000
-
-    def test_starts_at_stable_interval(self):
-        self.assertEqual(
-            self.schedule.interval_ns(self.now),
-            AdaptiveProbeSchedule.STABLE_NS,
-        )
-
-    def test_failure_increases_probe_frequency(self):
-        self.schedule.record_result(False, self.now)
-
-        self.assertEqual(
-            self.schedule.interval_ns(self.now),
-            AdaptiveProbeSchedule.FAILURE_NS,
-        )
-
-    def test_three_successes_enter_recovery_period(self):
-        self.schedule.record_result(False, self.now)
-
-        for i in range(3):
-            self.schedule.record_result(True, self.now + (i + 1) * 100_000_000)
-
-        recovery_time = self.now + 300_000_000
-
-        self.assertEqual(
-            self.schedule.interval_ns(recovery_time),
-            AdaptiveProbeSchedule.RECOVERY_NS,
-        )
-
-        self.assertEqual(
-            self.schedule.interval_ns(
-                recovery_time + AdaptiveProbeSchedule.RECOVERY_HOLD_NS
-            ),
-            AdaptiveProbeSchedule.STABLE_NS,
-        )
-
-    def test_another_failure_resets_recovery(self):
-        self.schedule.record_result(False, self.now)
-
-        for i in range(3):
-            self.schedule.record_result(True, self.now + (i + 1) * 100_000_000)
-
-        self.schedule.record_result(False, self.now + 400_000_000)
-
-        self.assertEqual(
-            self.schedule.interval_ns(self.now + 400_000_000),
-            AdaptiveProbeSchedule.FAILURE_NS,
-        )
-
-    def test_degraded_health_uses_uncertain_interval(self):
-        self.schedule.set_degraded(True, self.now)
-
-        self.assertEqual(
-            self.schedule.interval_ns(self.now),
-            AdaptiveProbeSchedule.UNCERTAIN_NS,
-        )
-
-    def test_recovered_health_has_temporary_faster_interval(self):
-        self.schedule.set_degraded(True, self.now)
-        self.schedule.set_degraded(False, self.now + 100_000_000)
-
-        self.assertEqual(
-            self.schedule.interval_ns(self.now + 100_000_000),
-            AdaptiveProbeSchedule.RECOVERY_NS,
-        )
-
-        self.assertEqual(
-            self.schedule.interval_ns(
-                self.now + 100_000_000
-                + AdaptiveProbeSchedule.RECOVERY_HOLD_NS
-            ),
-            AdaptiveProbeSchedule.STABLE_NS,
-        )
+    def test_starts_at_250ms(self):
+        scheduler = AdaptiveProbeSchedule()
+        self.assertEqual(scheduler.interval, 0.25)
+
+    def test_degraded_schedule_uses_fast_interval(self):
+        scheduler = AdaptiveProbeSchedule()
+        scheduler.set_degraded(True)
+        self.assertEqual(scheduler.interval_seconds(0), 0.25)
+
+    def test_failed_probe_halves_interval(self):
+        scheduler = AdaptiveProbeSchedule()
+        scheduler.record_result(False, now_ns=1_000_000_000)
+        self.assertEqual(scheduler.interval, 0.125)
+
+    def test_healthy_schedule_can_increase_interval(self):
+        scheduler = AdaptiveProbeSchedule()
+        scheduler.set_degraded(False)
+        scheduler.record_result(True, now_ns=1_000_000_000)
+        self.assertAlmostEqual(scheduler.interval, 0.3125)
+
+    def test_interval_never_exceeds_maximum(self):
+        scheduler = AdaptiveProbeSchedule()
+        scheduler.set_degraded(False)
+        for i in range(30):
+            scheduler.record_result(True, now_ns=i * 1_000_000_000)
+        self.assertLessEqual(scheduler.interval, scheduler.max_interval)
+
+    def test_interval_never_falls_below_minimum(self):
+        scheduler = AdaptiveProbeSchedule()
+        for i in range(30):
+            scheduler.record_result(False, now_ns=i * 1_000_000_000)
+        self.assertGreaterEqual(scheduler.interval, scheduler.min_interval)
+
+    def test_degraded_interval_is_capped_at_250ms(self):
+        scheduler = AdaptiveProbeSchedule()
+        scheduler.set_degraded(True)
+        scheduler.interval = 1.0
+        self.assertEqual(scheduler.interval_seconds(0), 0.25)
+
+    def test_initial_interval_is_within_bounds(self):
+        scheduler = AdaptiveProbeSchedule()
+        self.assertGreaterEqual(scheduler.interval, scheduler.min_interval)
+        self.assertLessEqual(scheduler.interval, scheduler.max_interval)
 
 
 if __name__ == "__main__":
-    unittest.main(verbosity=2)
+    unittest.main()

From 25eda08d910887ddefe390599bc0a798b3722b65 Mon Sep 17 00:00:00 2001
From: Saraswath BV <saraswathgowda@gmail.com>
Date: Mon, 2 Nov 2026 14:34:38 -0600
Subject: [PATCH 3/4] Implement multipath data distribution

---
 controller/routing.py   | 68 +++++++++++++++++++++++++++++++++++------
 src/client.c            | 63 ++++++++++++++++++++++++++++++++------
 tests/test_multipath.py | 36 +++++++++++++++++++++-
 3 files changed, 147 insertions(+), 20 deletions(-)

diff --git a/controller/routing.py b/controller/routing.py
index 7fff25b..c1af4dc 100644
--- a/controller/routing.py
+++ b/controller/routing.py
@@ -1,3 +1,4 @@
+
 """Small routing policy shared by the controller and its unit tests.
 
 The topology deliberately uses explicit ports. Both branches carry transit
@@ -28,18 +29,23 @@ def forwarding_rules(path):
     """Return exact-port IPv4/ARP forwarding rules; None blocks host ingress."""
     if path not in (None, "upper", "lower"):
         raise ValueError("unknown path")
+
     first = {"upper": 2, "lower": 3, None: None}[path]
     last = {"upper": 1, "lower": 2, None: None}[path]
+
     port_pairs = {
         1: ((1, first), (2, 1), (3, 1)),
         2: ((1, 2), (2, 1)),
         3: ((1, 2), (2, 1)),
         4: ((1, 3), (2, 3), (3, last)),
     }
+
     return {
-        dpid: [Flow(source, eth_type, destination)
-               for source, destination in pairs
-               for eth_type in (0x0800, 0x0806)]
+        dpid: [
+            Flow(source, eth_type, destination)
+            for source, destination in pairs
+            for eth_type in (0x0800, 0x0806)
+        ]
         for dpid, pairs in port_pairs.items()
     }
 
@@ -48,36 +54,78 @@ class PathPolicy:
     def __init__(self, mode="failover", initial="upper"):
         if mode not in ("static", "failover"):
             raise ValueError("mode must be static or failover")
+
         if initial not in PATH_PORTS:
             raise ValueError("initial path must be upper or lower")
+
         self.initial = initial
         self.mode = mode
         self.connected = set()
         self.ports = {}
         self.health = {}
 
+    def availability_details(self, path):
+        missing_switches = [
+            dpid for dpid in PATH_SWITCHES[path]
+            if dpid not in self.connected
+        ]
+
+        unavailable_ports = [
+            port for port in HOST_PORTS + PATH_PORTS[path]
+            if not self.ports.get(port, False)
+        ]
+
+        return {
+            "available": not missing_switches and not unavailable_ports,
+            "missing_switches": missing_switches,
+            "unavailable_ports": unavailable_ports,
+        }
+
     def available(self, path):
-        return (all(dpid in self.connected for dpid in PATH_SWITCHES[path])
-                and all(self.ports.get(port, False)
-                        for port in HOST_PORTS + PATH_PORTS[path]))
+        return self.availability_details(path)["available"]
 
     def choose(self, current):
         # A static route intentionally does not recover from a branch failure.
         # Host loss, however, leaves no usable endpoint in either mode.
         if not all(dpid in self.connected for dpid in (1, 4)):
             return None
+
         if not all(self.ports.get(port, False) for port in HOST_PORTS):
             return None
+
         if self.mode == "static":
             return self.initial
-        candidates = [path for path in (self.initial, "lower" if self.initial == "upper" else "upper") if self.available(path)]
+
+        candidates = [
+            path
+            for path in (
+                self.initial,
+                "lower" if self.initial == "upper" else "upper",
+            )
+            if self.available(path)
+        ]
+
         if not candidates:
             return None
-        if current in candidates and not all(path in self.health for path in candidates):
+
+        if current in candidates and not all(
+            path in self.health for path in candidates
+        ):
             return current
-        best = min(candidates, key=lambda path: self.health.get(path, 0))
+
+        best = min(
+            candidates,
+            key=lambda path: self.health.get(path, 0),
+        )
+
         # Require a score improvement greater than .02; this score combines
         # smoothed loss and delay, so the threshold is not a loss percentage.
-        if current in candidates and self.health.get(current, 0) <= self.health.get(best, 0) + .02:
+        if (
+            current in candidates
+            and self.health.get(current, 0)
+            <= self.health.get(best, 0) + 0.02
+        ):
             return current
+
         return best
+
diff --git a/src/client.c b/src/client.c
index 8116fcb..13143e4 100644
--- a/src/client.c
+++ b/src/client.c
@@ -1,5 +1,6 @@
 #include "utils.h"
 #include <arpa/inet.h>
+#include <errno.h>
 #include <netinet/in.h>
 #include <getopt.h>
 #include <math.h>
@@ -70,6 +71,7 @@ static int exchange(int fd, const struct sockaddr_in *server,
 
 struct transport_options {
     int adaptive, sack, congestion, multipath;
+    unsigned int upper_weight, lower_weight;
     double drop_rate;
 };
 
@@ -77,16 +79,27 @@ struct transport_options {
  * exactly the same silence it would see after network DATA loss. */
 static int send_data(int fd, const struct packet *packet,
                      const struct sockaddr_in *peer, double drop_rate,
-                     int multipath)
+                     int multipath, unsigned int upper_weight,
+                     unsigned int lower_weight)
 {
     if (drop_rate > 0 && rand() / (RAND_MAX + 1.0) < drop_rate) return 0;
 
     if (!multipath)
         return send_packet(fd, packet, peer);
 
-    /* DSCP 8 selects upper; DSCP 16 selects lower.
-     * Sequence parity keeps retransmissions on their original branch. */
-    int tos = (packet->sequence % 2 == 0 ? 8 : 16) << 2;
+    /* Weighted sequence mapping is deterministic, so retransmissions of
+     * a sequence number stay on the same path. */
+    if (upper_weight == 0 || lower_weight == 0 ||
+        upper_weight > 1000000U || lower_weight > 1000000U ||
+        upper_weight > 1000000U - lower_weight) {
+        errno = EINVAL;
+        return -1;
+    }
+
+    unsigned int total_weight = upper_weight + lower_weight;
+    unsigned int position = packet->sequence % total_weight;
+    int dscp = position < upper_weight ? 8 : 16;
+    int tos = dscp << 2;
     if (setsockopt(fd, IPPROTO_IP, IP_TOS, &tos, sizeof(tos)) != 0)
         return -1;
 
@@ -127,7 +140,9 @@ static int send_file(int fd, FILE *file, const struct sockaddr_in *server,
                 fprintf(stderr, "Input file changed or could not be read.\n");
                 return -1;
             }
-            if (send_data(fd, &slot->packet, server, options.drop_rate, options.multipath) != 0) {
+            if (send_data(fd, &slot->packet, server, options.drop_rate,
+                           options.multipath, options.upper_weight,
+                           options.lower_weight) != 0) {
                 perror("send DATA");
                 return -1;
             }
@@ -151,7 +166,9 @@ static int send_file(int fd, FILE *file, const struct sockaddr_in *server,
                     fprintf(stderr, "Retry limit reached for DATA %" PRIu32 ".\n", sequence);
                     return -1;
                 }
-                if (send_data(fd, &slot->packet, server, options.drop_rate, options.multipath) != 0) {
+                if (send_data(fd, &slot->packet, server, options.drop_rate,
+                           options.multipath, options.upper_weight,
+                           options.lower_weight) != 0) {
                     perror("retransmit DATA");
                     return -1;
                 }
@@ -253,7 +270,9 @@ static int send_file(int fd, FILE *file, const struct sockaddr_in *server,
                     struct send_slot *slot = &slots[seq % WINDOW_SIZE];
                     if (slot->acknowledged || slot->fast_sent || slot->gap_reports < 3) continue;
                     if (slot->retries >= MAX_RETRIES) return -1;
-                    if (send_data(fd, &slot->packet, server, options.drop_rate, options.multipath) != 0) return -1;
+                    if (send_data(fd, &slot->packet, server, options.drop_rate,
+                           options.multipath, options.upper_weight,
+                           options.lower_weight) != 0) return -1;
                     slot->fast_sent = 1;
                     slot->sent_at = monotonic_ms();
                     ++slot->retries; ++stats->data_sends; ++stats->data_retries;
@@ -298,13 +317,16 @@ int main(int argc, char *argv[])
     uint16_t port;
     const char *host = NULL, *port_text = NULL, *filename = NULL;
     const char *rto_mode = getenv("RUDP_RTO");
-    struct transport_options options = {0, option_enabled("RUDP_SACK"), option_enabled("RUDP_CC"), option_enabled("RUDP_MULTIPATH"), 0};
+    struct transport_options options = {0, option_enabled("RUDP_SACK"), option_enabled("RUDP_CC"), option_enabled("RUDP_MULTIPATH"), 1, 1, 0};
     static const struct option cli[] = {
         {"host", required_argument, NULL, 'H'}, {"port", required_argument, NULL, 'p'},
         {"file", required_argument, NULL, 'f'}, {"sack", no_argument, NULL, 's'},
         {"no-sack", no_argument, NULL, 'S'}, {"cc", no_argument, NULL, 'c'},
         {"no-cc", no_argument, NULL, 'C'}, {"multipath", no_argument, NULL, 'm'}, {"adaptive-rto", no_argument, NULL, 'a'},
-        {"fixed-rto", no_argument, NULL, 'F'}, {"drop-rate", required_argument, NULL, 'd'},
+        {"fixed-rto", no_argument, NULL, 'F'},
+        {"upper-weight", required_argument, NULL, 'u'},
+        {"lower-weight", required_argument, NULL, 'l'},
+        {"drop-rate", required_argument, NULL, 'd'},
         {"help", no_argument, NULL, 'h'}, {NULL, 0, NULL, 0}
     };
     int opt;
@@ -318,6 +340,21 @@ int main(int argc, char *argv[])
         case 's': case 'S': options.sack = opt == 's'; break;
         case 'c': case 'C': options.congestion = opt == 'c'; break;
         case 'm': options.multipath = 1; break;
+        case 'u': case 'l': {
+            char *end;
+            errno = 0;
+            unsigned long weight = strtoul(optarg, &end, 10);
+            if (errno || end == optarg || *end ||
+                weight < 1 || weight > 1000000UL) {
+                fprintf(stderr, "Path weights must be integers from 1 to 1000000\n");
+                return EXIT_FAILURE;
+            }
+            if (opt == 'u')
+                options.upper_weight = (unsigned int)weight;
+            else
+                options.lower_weight = (unsigned int)weight;
+            break;
+        }
         case 'd': {
             char *end;
             options.drop_rate = strtod(optarg, &end);
@@ -333,6 +370,8 @@ int main(int argc, char *argv[])
             puts("Usage: client HOST PORT FILE | client --host IPv4 --port PORT --file FILE\n"
                  "Options: --sack | --no-sack, --adaptive-rto | --fixed-rto, --cc | --no-cc\n"
                  "         --multipath    distribute DATA across upper/lower SDN paths\n"
+                 "         --upper-weight N  upper path weight (default 1)\n"
+                 "         --lower-weight N  lower path weight (default 1)\n"
                  "         --drop-rate P   simulate loss on each outgoing DATA send (0..1)");
             return EXIT_SUCCESS;
         default: return EXIT_FAILURE;
@@ -344,6 +383,12 @@ int main(int argc, char *argv[])
         fprintf(stderr, "Usage: client HOST PORT FILE (see --help)\n");
         return EXIT_FAILURE;
     }
+    if (options.upper_weight < 1 || options.lower_weight < 1 ||
+        options.lower_weight > 1000000U ||
+        options.upper_weight > 1000000U - options.lower_weight) {
+        fprintf(stderr, "Combined path weights must not exceed 1000000\n");
+        return EXIT_FAILURE;
+    }
     if (rto_mode && strcmp(rto_mode, "fixed") && strcmp(rto_mode, "adaptive")) {
         fprintf(stderr, "RUDP_RTO must be fixed or adaptive.\n");
         return EXIT_FAILURE;
diff --git a/tests/test_multipath.py b/tests/test_multipath.py
index f4f4788..a67dc5f 100644
--- a/tests/test_multipath.py
+++ b/tests/test_multipath.py
@@ -4,9 +4,12 @@
 
 
 class MultipathTests(unittest.TestCase):
+
     def test_alternates_between_healthy_paths(self):
         scheduler = MultipathScheduler()
+
         selected = [scheduler.select_path() for _ in range(6)]
+
         self.assertEqual(
             selected,
             ["upper", "lower", "upper", "lower", "upper", "lower"],
@@ -15,15 +18,20 @@ def test_alternates_between_healthy_paths(self):
     def test_unhealthy_path_is_not_selected(self):
         scheduler = MultipathScheduler()
         scheduler.update_path("upper", healthy=False)
+
+        selected = [scheduler.select_path() for _ in range(4)]
+
         self.assertEqual(
-            [scheduler.select_path() for _ in range(4)],
+            selected,
             ["lower", "lower", "lower", "lower"],
         )
 
     def test_weighted_schedule(self):
         scheduler = MultipathScheduler()
         scheduler.update_path("upper", weight=2)
+
         selected = [scheduler.select_path() for _ in range(6)]
+
         self.assertEqual(selected.count("upper"), 4)
         self.assertEqual(selected.count("lower"), 2)
 
@@ -31,8 +39,34 @@ def test_all_paths_unhealthy(self):
         scheduler = MultipathScheduler()
         scheduler.update_path("upper", healthy=False)
         scheduler.update_path("lower", healthy=False)
+
         self.assertIsNone(scheduler.select_path())
 
+    def test_weighted_schedule_respects_configured_ratio(self):
+        scheduler = MultipathScheduler()
+        scheduler.update_path("upper", weight=3)
+        scheduler.update_path("lower", weight=1)
+
+        selected = [scheduler.select_path() for _ in range(8)]
+
+        self.assertEqual(selected.count("upper"), 6)
+        self.assertEqual(selected.count("lower"), 2)
+
+    def test_recovered_path_returns_to_schedule(self):
+        scheduler = MultipathScheduler()
+        scheduler.update_path("upper", healthy=False)
+
+        self.assertEqual(scheduler.select_path(), "lower")
+
+        scheduler.update_path("upper", healthy=True)
+
+        selected = [scheduler.select_path() for _ in range(4)]
+
+        self.assertCountEqual(
+            selected,
+            ["upper", "upper", "lower", "lower"],
+        )
+
 
 if __name__ == "__main__":
     unittest.main()

From 1b0e9daf501159d6faf130e106af4810b672a4e5 Mon Sep 17 00:00:00 2001
From: sameermanvi <sameermanvi07@gmail.com>
Date: Tue, 3 Nov 2026 14:05:55 +0530
Subject: [PATCH 4/4] Fix multipath health routing, portable build and
 validation coverage

---
 .github/workflows/checks.yml |  30 +++++++---
 Makefile                     |   2 +
 controller/failover.py       |  17 ++++--
 controller/multipath.py      |  91 +++++------------------------
 controller/probe_schedule.py |   3 +-
 docs/ARCHITECTURE.md         |  25 +++++++-
 docs/DEMO.md                 |  13 +++++
 docs/VALIDATION.md           |  26 +++++++++
 src/client.c                 |   4 ++
 tests/multipath_sender.c     |  44 ++++++++++++++
 tests/test_local.py          |  18 ++++++
 tests/test_multipath.py      | 109 +++++++++++++++--------------------
 tests/test_openflow.py       |  43 ++++++++++++++
 tests/test_probe_schedule.py |   9 +++
 14 files changed, 278 insertions(+), 156 deletions(-)
 create mode 100644 tests/multipath_sender.c

diff --git a/.github/workflows/checks.yml b/.github/workflows/checks.yml
index 0dbf844..43f33a3 100644
--- a/.github/workflows/checks.yml
+++ b/.github/workflows/checks.yml
@@ -20,9 +20,8 @@ jobs:
       - uses: actions/checkout@v4
       - name: Build and test
         run: |
-          make -C src CFLAGS='-std=c11 -O2 -Wall -Wextra -Wpedantic -Werror' test
-          python3 tests/test_controller.py
-          python3 tests/test_metrics.py
+          make -C src CFLAGS='-std=c11 -O2 -Wall -Wextra -Wpedantic -Werror'
+          make test
       - name: Sanitizer checks
         if: runner.os == 'Linux'
         run: |
@@ -30,22 +29,35 @@ jobs:
           make -C src CC=clang CFLAGS='-std=c11 -O1 -g -Wall -Wextra -Wpedantic -Werror -fsanitize=address,undefined -fno-omit-frame-pointer' LDFLAGS='-fsanitize=address,undefined' test
 
   network:
+    strategy:
+      fail-fast: false
+      matrix:
+        suite: [matrix, concurrent]
     runs-on: ubuntu-22.04
-    timeout-minutes: 20
+    timeout-minutes: 45
     steps:
       - uses: actions/checkout@v4
       - name: Install network lab
         run: sudo bash scripts/setup_env.sh
       - name: Exercise normal, loss and link failure
+        if: matrix.suite == 'matrix'
         run: |
           sudo .venv/bin/python scripts/run.py experiment --repeats 1 --output results/ci
-          sudo .venv/bin/python scripts/run.py experiment --scenarios link-failure --modes failover --flows 2 --repeats 1 --output results/ci-concurrent
-          .venv/bin/python scripts/run.py plot results/ci/metrics.csv --output results/ci/plots
+          sudo .venv/bin/python scripts/run.py plot results/ci/metrics.csv --output results/ci/plots
+      - name: Exercise concurrent failover
+        if: matrix.suite == 'concurrent'
+        run: sudo .venv/bin/python scripts/run.py experiment --scenarios link-failure --modes failover --flows 2 --repeats 1 --output results/ci-concurrent
       - name: Save logs and measurements
         if: always()
         uses: actions/upload-artifact@v4
         with:
-          name: network-results
+          name: network-results-${{ matrix.suite }}
+          # Retain measurements and diagnostics, excluding temporary payloads.
           path: |
-            results/ci/
-            results/ci-concurrent/
+            results/ci*/**/*.log
+            results/ci*/**/*.json
+            results/ci*/**/*.jsonl
+            results/ci*/**/*.csv
+            results/ci*/**/*.svg
+            results/ci*/**/*.md
+            results/ci*/**/*.txt
diff --git a/Makefile b/Makefile
index cb5000c..b052bd0 100644
--- a/Makefile
+++ b/Makefile
@@ -9,6 +9,8 @@ clean:
 test: all
 	python3 tests/test_protocol.py
 	python3 tests/test_transport.py
+	python3 tests/test_multipath.py
+	python3 tests/test_probe_schedule.py
 	python3 tests/test_probes.py
 	python3 tests/test_openflow.py
 	python3 tests/test_controller.py
diff --git a/controller/failover.py b/controller/failover.py
index 42c22dc..16b53d9 100644
--- a/controller/failover.py
+++ b/controller/failover.py
@@ -37,6 +37,7 @@
 from controller.routing import PathPolicy, SWITCHES, forwarding_rules
 from controller.probes import ProbeTracker, ENDPOINTS, ETHERTYPE
 from controller.probe_schedule import AdaptiveProbeSchedule
+from controller.multipath import eligible_paths
 
 
 COOKIE = 0x5255445000000001
@@ -79,6 +80,7 @@ def __init__(self, *args, **kwargs):
         # Each path/direction has its own next-probe deadline.
         self.next_probe_ns = {key: 0 for key in ENDPOINTS}
 
+        self.multipath_paths = frozenset()
         self.last_route_change_ns = 0
         self.monitor = hub.spawn(self._monitor)
 
@@ -404,7 +406,9 @@ def _reconcile(self, cause, force=False):
 
         path = self.policy.choose(self.path)
 
-        if not force and path == self.path:
+        multipath_paths = eligible_paths(self.policy, path)
+
+        if not force and path == self.path and multipath_paths == self.multipath_paths:
             return
 
         now = time.clock_gettime_ns(time.CLOCK_MONOTONIC)
@@ -423,7 +427,9 @@ def _reconcile(self, cause, force=False):
                 )
                 return
 
-        self.last_route_change_ns = now
+        if path != self.path:
+            self.last_route_change_ns = now
+        self.multipath_paths = multipath_paths
 
         self._event(
             "route_decision",
@@ -490,7 +496,7 @@ def _reconcile(self, cause, force=False):
             ))
 
             # DATA uses DSCP 8 for upper and DSCP 16 for lower.
-            multipath_paths = {
+            multipath_rules = {
                 "upper": (
                     8,
                     ((1, 1, 2), (2, 1, 2), (4, 1, 3)),
@@ -501,7 +507,7 @@ def _reconcile(self, cause, force=False):
                 ),
             }
 
-            for path_name, (dscp, entries) in multipath_paths.items():
+            for path_name, (dscp, entries) in multipath_rules.items():
                 details = self.policy.availability_details(path_name)
 
                 self._event(
@@ -514,7 +520,8 @@ def _reconcile(self, cause, force=False):
                     unavailable_ports=details["unavailable_ports"],
                 )
 
-                if not details["available"]:
+                # Health changes can remove a branch even while its ports stay up.
+                if path_name not in multipath_paths:
                     continue
 
                 for switch_id, ingress, output in entries:
diff --git a/controller/multipath.py b/controller/multipath.py
index f54233a..d955702 100644
--- a/controller/multipath.py
+++ b/controller/multipath.py
@@ -1,75 +1,16 @@
-"""Multipath scheduling logic for Reliable UDP File Transfer.
-
-Selects a path for each outgoing data packet using weighted scheduling.
-This module is independent of OS-Ken so its scheduling logic is testable.
-"""
-
-from dataclasses import dataclass
-from itertools import cycle
-from threading import Lock
-
-
-@dataclass(frozen=True)
-class PathState:
-    """Current scheduling state for a network path."""
-
-    name: str
-    healthy: bool = True
-    weight: int = 1
-
-
-class MultipathScheduler:
-    """Thread-safe weighted scheduler for healthy network paths."""
-
-    def __init__(self, paths=("upper", "lower")):
-        if not paths:
-            raise ValueError("At least one path is required")
-
-        self._lock = Lock()
-        self._paths = {
-            name: PathState(name=name)
-            for name in paths
-        }
-        self._schedule = cycle(paths)
-
-    def update_path(self, name, *, healthy=None, weight=None):
-        """Update a path's health or scheduling weight."""
-        with self._lock:
-            if name not in self._paths:
-                raise ValueError(f"Unknown path: {name}")
-
-            current = self._paths[name]
-            new_weight = current.weight if weight is None else weight
-
-            if new_weight < 1:
-                raise ValueError("Path weight must be at least 1")
-
-            self._paths[name] = PathState(
-                name=name,
-                healthy=current.healthy if healthy is None else healthy,
-                weight=new_weight,
-            )
-            self._rebuild_schedule()
-
-    def _rebuild_schedule(self):
-        """Build a weighted cycle from currently healthy paths."""
-        entries = []
-
-        for path in self._paths.values():
-            if path.healthy:
-                entries.extend([path.name] * path.weight)
-
-        self._schedule = cycle(entries)
-
-    def select_path(self):
-        """Return the next healthy path, or None if all paths are down."""
-        with self._lock:
-            if not any(path.healthy for path in self._paths.values()):
-                return None
-
-            return next(self._schedule)
-
-    def get_paths(self):
-        """Return a snapshot of the configured path states."""
-        with self._lock:
-            return dict(self._paths)
+"""Choose branches eligible for DSCP DATA rules using the routing policy."""
+
+
+def eligible_paths(policy, selected):
+    # Static experiments must retain their intentionally fixed route.
+    if policy.mode != "failover" or selected is None:
+        return frozenset()
+    paths = {selected} if policy.available(selected) else set()
+    selected_score = policy.health.get(selected)
+    if selected_score is not None:
+        for path in ("upper", "lower"):
+            score = policy.health.get(path)
+            if (policy.available(path) and score is not None
+                    and score <= selected_score + .02):
+                paths.add(path)
+    return frozenset(paths)
diff --git a/controller/probe_schedule.py b/controller/probe_schedule.py
index 882684d..23d6ca7 100644
--- a/controller/probe_schedule.py
+++ b/controller/probe_schedule.py
@@ -6,7 +6,8 @@ class AdaptiveProbeSchedule:
     def __init__(
         self,
         min_interval=0.1,
-        max_interval=2.0,
+        # Leave the 500 ms response deadline inside the tracker's 2 s freshness.
+        max_interval=1.5,
         initial_interval=0.25,
         healthy_threshold=0.05,
         degraded_threshold=0.10,
diff --git a/docs/ARCHITECTURE.md b/docs/ARCHITECTURE.md
index 792d5a5..87c8420 100644
--- a/docs/ARCHITECTURE.md
+++ b/docs/ARCHITECTURE.md
@@ -116,7 +116,7 @@ failback. Initial branch defaults to upper; RUDP_INITIAL_PATH=lower permits the
 reverse-direction recovery check.
 
 `controller/probes.py` tracks independent Ethernet probes in both directions on
-both branches, including the idle branch. Every 250 ms the controller sends a
+both branches, including the idle branch. At each adaptive deadline the controller sends a
 60-byte experimental-EtherType 0x88b5 frame with a random session token and sequence
 through the selected branch port. High-priority transit rules cross s2/s3; edge
 rules punt the arriving frame to the controller and never to hosts. Validation
@@ -129,7 +129,8 @@ Latency includes controller/OpenFlow overhead; it is not a pure link RTT or loss
 probability. Both directions need four samples, each at most 2 s old. Health-only
 moves require an improvement exceeding .02 and a 2 s route hold-down, and wait for
 outstanding barriers. Hard port failures bypass that hold-down. Link changes clear
-old branch health. Approximately 16 small frames/s probe both live branches.
+old branch health. At the initial 250 ms spacing, both branches use approximately
+16 small frames/s; healthy spacing grows and loss accelerates probes.
 
 ## Socket–SDN observability and recovery definition
 
@@ -150,3 +151,23 @@ No route or no later useful ACK is censored/N/A, never zero. Repeated-fault
 attribution ends at the next intervention; a failure on the inactive branch is
 labeled separately. The joined SVG and recovery JSON preserve all three timestamps.
 See PROGRESS_LOG.md, DEMO.md and scripts/evaluate.py.
+
+## Optional multipath DATA and adaptive probes
+
+`client --multipath --upper-weight 3 --lower-weight 1` marks DATA with DSCP 8
+or 16 using sequence modulo total weight. Retries retain the sequence's mark;
+control packets reset to DSCP 0. The controller maps those marks onto eligible
+branches. This is forward DATA distribution; ACKs use the selected reverse route.
+It is not independent congestion control per path.
+
+A branch must have working ports and measured health within .02 of the selected
+path's score to join it. Unmeasured alternates are excluded. Removing an unhealthy
+mark-specific rule makes that traffic use the selected base route. Static mode
+installs no multipath overrides, preserving the permanent-failure comparison.
+
+Probe deadlines are independent per branch/direction. Successful healthy probes
+increase spacing up to 1.5 seconds; loss halves it to a 100 ms minimum. Degraded
+paths are capped at 250 ms. The tracker still expires outstanding probes at
+500 ms; the 1.5 s spacing leaves that deadline inside the 2 s health freshness.
+These changes need fresh Linux measurements; historical fixed-interval
+results do not measure their overhead or recovery behavior.
diff --git a/docs/DEMO.md b/docs/DEMO.md
index 34f20d9..913596b 100644
--- a/docs/DEMO.md
+++ b/docs/DEMO.md
@@ -81,3 +81,16 @@ docker run --rm --privileged -v "$PWD/results:/project/results" rudp-lab python
 The console's `recovery=` is injection-to-progress, not link-down-to-progress.
 The CSV and joined events distinguish both. Never label a startup failure or
 uninjected transfer as successful failover.
+
+## Two simultaneous uploads
+
+Each flow uses a separate receiver process and UDP port on h2; the receiver is
+one upload per process, rather than a persistent multi-session service. Both
+senders run on h1. This demonstrates concurrent uploads, not separate sender hosts.
+
+```sh
+sudo .venv/bin/python scripts/run.py experiment --scenarios link-failure --modes failover --rtos adaptive --sacks bitmap --ccs off --flows 2 --repeats 1 --output results/live-concurrent
+```
+
+Check both flow rows for success, matching hashes and recovery. Rehearse this
+before assessment; use retained output during the ten-minute presentation.
diff --git a/docs/VALIDATION.md b/docs/VALIDATION.md
index f00372f..4931047 100644
--- a/docs/VALIDATION.md
+++ b/docs/VALIDATION.md
@@ -121,3 +121,29 @@ Historical material is preserved in [the archive](../results/archive/README.md).
 These measurements are retained from commit
 `79f097dea2ae8f61e70dd4558816531dfbbb97a5`. This newer client/codec interface has separate localhost checks. The recorded Linux
 measurements describe the earlier commit and its original source fingerprints.
+
+
+## Validate the final submission revision
+
+After merging, freeze the code and rebuild the Linux image from that checkout.
+Run the preflight and evaluation into fresh directories. Record the code commit
+alongside the output so later documentation commits cannot obscure its identity.
+
+```sh
+git rev-parse HEAD > results/submission-source.txt
+docker build -t rudp-lab:submission .
+sudo docker run --rm --privileged -v "$PWD/results:/project/results" rudp-lab:submission python scripts/run.py demo --mode failover --datapath user --output results/submission-preflight
+sudo docker run --rm --privileged -v "$PWD/results:/project/results" rudp-lab:submission python scripts/run.py evaluate --output results/submission-evaluation
+```
+
+The standard evaluation does not enable the optional multipath sender. Its
+successful completion alone cannot establish simultaneous use of both branches.
+Multipath needs a Linux trial with marked DATA, per-branch packet counters,
+matching hashes and hard/soft failure interventions. Localhost and recording
+OpenFlow tests cover implementation behavior, not that Mininet evidence gap.
+
+For an optional performance comparison, existing experiment switches support
+`--file-size`, `--delay`, `--flows` and `--ccs off on`. Change one factor at a time,
+use three repeats and fresh directories, and compare per-flow goodput and total
+goodput. A fairness claim requires competition measurements; it cannot be inferred
+from a clean single-flow transfer. No new performance measurements are claimed here.
diff --git a/src/client.c b/src/client.c
index 13143e4..497087b 100644
--- a/src/client.c
+++ b/src/client.c
@@ -1,3 +1,7 @@
+#ifdef __APPLE__
+/* IP_TOS is a Darwin extension hidden by the POSIX feature profile. */
+#define _DARWIN_C_SOURCE
+#endif
 #include "utils.h"
 #include <arpa/inet.h>
 #include <errno.h>
diff --git a/tests/multipath_sender.c b/tests/multipath_sender.c
new file mode 100644
index 0000000..d480c79
--- /dev/null
+++ b/tests/multipath_sender.c
@@ -0,0 +1,44 @@
+/* Exercise the real sender's DSCP selection without requiring an SDN switch. */
+#define main client_main
+#define setsockopt record_setsockopt
+#define send_packet record_send_packet
+#include "../src/client.c"
+#undef main
+#undef setsockopt
+#undef send_packet
+#include <assert.h>
+
+static int current_tos, observed_tos;
+
+int record_setsockopt(int fd, int level, int option, const void *value, socklen_t length)
+{
+    (void)fd;
+    assert(level == IPPROTO_IP && option == IP_TOS && length == sizeof(int));
+    current_tos = *(const int *)value;
+    return 0;
+}
+
+int record_send_packet(int fd, const struct packet *packet, const struct sockaddr_in *peer)
+{
+    (void)fd; (void)packet; (void)peer;
+    observed_tos = current_tos;
+    return 0;
+}
+
+int main(void)
+{
+    struct packet packet = {0};
+    struct sockaddr_in peer = {0};
+    for (unsigned int sequence = 0; sequence < 16; ++sequence) {
+        packet.sequence = sequence;
+        for (int attempt = 0; attempt < 2; ++attempt) {
+            assert(send_data(0, &packet, &peer, 0, 1, 3, 1) == 0);
+            assert(observed_tos == ((sequence % 4 < 3 ? 8 : 16) << 2));
+            assert(current_tos == 0); /* Control traffic must not inherit DATA's mark. */
+        }
+    }
+    assert(send_data(0, &packet, &peer, 0, 0, 1, 1) == 0);
+    assert(observed_tos == 0);
+    assert(send_data(0, &packet, &peer, 0, 1, 0, 1) == -1);
+    return 0;
+}
diff --git a/tests/test_local.py b/tests/test_local.py
index d4508f2..26d98eb 100644
--- a/tests/test_local.py
+++ b/tests/test_local.py
@@ -60,6 +60,24 @@ def stop(process):
 
 
 class LocalTests(unittest.TestCase):
+    def test_multipath_client_transfers_with_weights(self):
+        with tempfile.TemporaryDirectory() as folder:
+            source, output = Path(folder)/'input', Path(folder)/'output'
+            data = os.urandom(32769)
+            source.write_bytes(data)
+            receiver, address = server(output)
+            try:
+                result = subprocess.run(
+                    [str(CLIENT), address[0], str(address[1]), str(source),
+                     '--multipath', '--upper-weight', '3', '--lower-weight', '1'],
+                    capture_output=True, text=True, timeout=15)
+                self.assertEqual(result.returncode, 0, result.stderr)
+                _, stderr = receiver.communicate(timeout=12)
+                self.assertEqual(receiver.returncode, 0, stderr)
+                self.assertEqual(output.read_bytes(), data)
+            finally:
+                stop(receiver)
+
     def test_binary_sizes(self):
         def transfer(size):
             with tempfile.TemporaryDirectory() as folder:
diff --git a/tests/test_multipath.py b/tests/test_multipath.py
index a67dc5f..b61e0cc 100644
--- a/tests/test_multipath.py
+++ b/tests/test_multipath.py
@@ -1,71 +1,52 @@
+"""Exercise the policy used by the real DSCP rule installer."""
+from pathlib import Path
+import sys
 import unittest
-
-from controller.multipath import MultipathScheduler
+import os
+import shlex
+import subprocess
+import tempfile
+sys.path.insert(0, str(Path(__file__).resolve().parents[1]))
+from controller.multipath import eligible_paths
+from test_controller import connected_policy
 
 
 class MultipathTests(unittest.TestCase):
-
-    def test_alternates_between_healthy_paths(self):
-        scheduler = MultipathScheduler()
-
-        selected = [scheduler.select_path() for _ in range(6)]
-
-        self.assertEqual(
-            selected,
-            ["upper", "lower", "upper", "lower", "upper", "lower"],
-        )
-
-    def test_unhealthy_path_is_not_selected(self):
-        scheduler = MultipathScheduler()
-        scheduler.update_path("upper", healthy=False)
-
-        selected = [scheduler.select_path() for _ in range(4)]
-
-        self.assertEqual(
-            selected,
-            ["lower", "lower", "lower", "lower"],
-        )
-
-    def test_weighted_schedule(self):
-        scheduler = MultipathScheduler()
-        scheduler.update_path("upper", weight=2)
-
-        selected = [scheduler.select_path() for _ in range(6)]
-
-        self.assertEqual(selected.count("upper"), 4)
-        self.assertEqual(selected.count("lower"), 2)
-
-    def test_all_paths_unhealthy(self):
-        scheduler = MultipathScheduler()
-        scheduler.update_path("upper", healthy=False)
-        scheduler.update_path("lower", healthy=False)
-
-        self.assertIsNone(scheduler.select_path())
-
-    def test_weighted_schedule_respects_configured_ratio(self):
-        scheduler = MultipathScheduler()
-        scheduler.update_path("upper", weight=3)
-        scheduler.update_path("lower", weight=1)
-
-        selected = [scheduler.select_path() for _ in range(8)]
-
-        self.assertEqual(selected.count("upper"), 6)
-        self.assertEqual(selected.count("lower"), 2)
-
-    def test_recovered_path_returns_to_schedule(self):
-        scheduler = MultipathScheduler()
-        scheduler.update_path("upper", healthy=False)
-
-        self.assertEqual(scheduler.select_path(), "lower")
-
-        scheduler.update_path("upper", healthy=True)
-
-        selected = [scheduler.select_path() for _ in range(4)]
-
-        self.assertCountEqual(
-            selected,
-            ["upper", "upper", "lower", "lower"],
-        )
+    def setUp(self):
+        self.policy = connected_policy()
+        self.policy.health = {"upper": .01, "lower": .01}
+
+    def test_real_sender_marks_weights_retries_and_resets(self):
+        root = Path(__file__).resolve().parents[1]
+        with tempfile.TemporaryDirectory() as folder:
+            binary = Path(folder)/'sender-test'
+            subprocess.run(shlex.split(os.environ.get('CC', 'cc')) + [
+                '-std=c11', '-D_POSIX_C_SOURCE=200809L', '-Wall', '-Wextra',
+                '-Wpedantic', '-Werror', str(root/'tests/multipath_sender.c'),
+                str(root/'src/utils.c'), '-o', str(binary)], check=True)
+            subprocess.run([str(binary)], check=True, timeout=5)
+
+    def test_healthy_paths_share_data(self):
+        self.assertEqual(eligible_paths(self.policy, "upper"), {"upper", "lower"})
+
+    def test_degraded_path_is_excluded(self):
+        self.policy.health["upper"] = .5
+        self.assertEqual(eligible_paths(self.policy, "lower"), {"lower"})
+
+    def test_unmeasured_alternate_is_excluded(self):
+        del self.policy.health["lower"]
+        self.assertEqual(eligible_paths(self.policy, "upper"), {"upper"})
+
+    def test_failed_path_is_excluded(self):
+        self.policy.ports[(1, 2)] = False
+        self.assertEqual(eligible_paths(self.policy, "lower"), {"lower"})
+
+    def test_static_mode_has_no_overrides(self):
+        self.policy.mode = "static"
+        self.assertFalse(eligible_paths(self.policy, "upper"))
+
+    def test_no_selected_route_has_no_overrides(self):
+        self.assertFalse(eligible_paths(self.policy, None))
 
 
 if __name__ == "__main__":
diff --git a/tests/test_openflow.py b/tests/test_openflow.py
index 252dc7c..30b7f7f 100644
--- a/tests/test_openflow.py
+++ b/tests/test_openflow.py
@@ -65,6 +65,49 @@ def test_soft_hold_down_never_blocks_hard_failure(self):
             c._reconcile('link_down')
             self.assertEqual(c.path, 'lower')
 
+    def test_multipath_rules_follow_health_without_route_change(self):
+        c = self.controller
+        c.policy = connected_policy()
+        c.policy.health = {'upper': .01, 'lower': .01}
+        c.started = True
+        parser = MagicMock()
+        parser.OFPMatch.side_effect = lambda **kwargs: kwargs
+        dp = types.SimpleNamespace(
+            id=1, ofproto_parser=parser,
+            ofproto=types.SimpleNamespace(OFPIT_APPLY_ACTIONS=4, OFPFC_DELETE=3,
+                                         OFPP_ANY=0xffffffff, OFPG_ANY=0xffffffff),
+            send_msg=MagicMock(), set_xid=MagicMock())
+        c.datapaths = {1: dp}
+
+        def installed_dscps():
+            return {call.kwargs['match']['ip_dscp']
+                    for call in parser.OFPFlowMod.call_args_list
+                    if call.kwargs.get('priority') == 150}
+
+        c._reconcile('startup')
+        self.assertEqual(installed_dscps(), {8, 16})
+        parser.reset_mock()
+        c.pending.clear()
+        # The alternate degrades while the selected route remains upper.
+        c.policy.health['lower'] = .5
+        c._reconcile('health_update')
+        self.assertEqual(c.path, 'upper')
+        self.assertEqual(installed_dscps(), {8})
+        self.assertTrue(any(call.kwargs.get('cookie') == self.module.MULTIPATH_COOKIE
+                            and call.kwargs.get('command') == 3
+                            for call in parser.OFPFlowMod.call_args_list))
+        parser.reset_mock()
+        c.pending.clear()
+        c.last_route_change_ns = 0
+        c.policy.health = {'upper': .5, 'lower': .01}
+        c._reconcile('health_update')
+        self.assertEqual(c.path, 'lower')
+        self.assertEqual(installed_dscps(), {16})
+        parser.reset_mock()
+        c.policy.mode = 'static'
+        c._reconcile('startup', force=True)
+        self.assertFalse(installed_dscps())
+
     def test_old_barrier_never_marks_new_generation_ready(self):
         c = self.controller; c.generation = 2; c.pending = {(1, 10): 1}; c.path = 'lower'
         c.barrier_reply(types.SimpleNamespace(msg=types.SimpleNamespace(datapath=types.SimpleNamespace(id=1), xid=10)))
diff --git a/tests/test_probe_schedule.py b/tests/test_probe_schedule.py
index 14ba95b..bf0af0b 100644
--- a/tests/test_probe_schedule.py
+++ b/tests/test_probe_schedule.py
@@ -1,4 +1,7 @@
+from pathlib import Path
+import sys
 import unittest
+sys.path.insert(0, str(Path(__file__).resolve().parents[1]))
 
 from controller.probe_schedule import AdaptiveProbeSchedule
 
@@ -43,6 +46,12 @@ def test_degraded_interval_is_capped_at_250ms(self):
         scheduler.interval = 1.0
         self.assertEqual(scheduler.interval_seconds(0), 0.25)
 
+    def test_maximum_spacing_leaves_time_for_response(self):
+        from controller.probes import ProbeTracker
+        schedule, tracker = AdaptiveProbeSchedule(), ProbeTracker()
+        self.assertLessEqual(schedule.max_interval * 1e9 + tracker.timeout_ns,
+                             tracker.freshness_ns)
+
     def test_initial_interval_is_within_bounds(self):
         scheduler = AdaptiveProbeSchedule()
         self.assertGreaterEqual(scheduler.interval, scheduler.min_interval)