The port enable and disable predicates spell out the same condition, that the port is in the selected aggregator and the partner is synchronized, once plainly and once negated. The advertised DISTRIBUTING bit is that same condition a third time.
Factor the condition into lacp_port_synced() and derive lacp_port_should_be_disabled() from lacp_port_should_be_enabled(). lacp_port_agg_selected() already implies a non-NULL agg_lead, so the separate lacp_port_selected() test is dropped.
In preparation for making the enable predicate mode dependent, take the advertised DISTRIBUTING bit from lacp_port_should_be_enabled() instead of computing it alongside COLLECTING, so the advertised state and the kernel port enable decision stay a single decision once they are no longer the same expression.
No functional change intended.
Signed-off-by: Anisha Halwai anisha@nexthop.ai --- New in v5: split out of the feature patch, as requested in review. teamd/teamd_runner_lacp.c | 27 +++++++++++---------------- 1 file changed, 11 insertions(+), 16 deletions(-)
diff --git a/teamd/teamd_runner_lacp.c b/teamd/teamd_runner_lacp.c index 3b4f7b1..b8a4c30 100644 --- a/teamd/teamd_runner_lacp.c +++ b/teamd/teamd_runner_lacp.c @@ -337,26 +337,20 @@ static bool lacp_port_selected(struct lacp_port *lacp_port) return lacp_port->agg_lead; }
-static int lacp_port_should_be_enabled(struct lacp_port *lacp_port) +static bool lacp_port_synced(struct lacp_port *lacp_port) { - struct lacp *lacp = lacp_port->lacp; + return lacp_port_agg_selected(lacp_port) && + (lacp_port->partner.state & INFO_STATE_SYNCHRONIZATION); +}
- if (lacp_port_selected(lacp_port) && - lacp_port->agg_lead == lacp->selected_agg_lead && - lacp_port->partner.state & INFO_STATE_SYNCHRONIZATION) - return true; - return false; +static int lacp_port_should_be_enabled(struct lacp_port *lacp_port) +{ + return lacp_port_synced(lacp_port); }
static int lacp_port_should_be_disabled(struct lacp_port *lacp_port) { - struct lacp *lacp = lacp_port->lacp; - - if (!lacp_port_selected(lacp_port) || - lacp_port->agg_lead != lacp->selected_agg_lead || - !(lacp_port->partner.state & INFO_STATE_SYNCHRONIZATION)) - return true; - return false; + return !lacp_port_should_be_enabled(lacp_port); }
static int lacp_port_update_enabled(struct lacp_port *lacp_port) @@ -929,8 +923,9 @@ static void lacp_port_actor_update(struct lacp_port *lacp_port) state |= INFO_STATE_SYNCHRONIZATION; state &= ~(INFO_STATE_COLLECTING | INFO_STATE_DISTRIBUTING); if (lacp_port->partner.state & INFO_STATE_SYNCHRONIZATION) - state |= INFO_STATE_COLLECTING | - INFO_STATE_DISTRIBUTING; + state |= INFO_STATE_COLLECTING; + if (lacp_port_should_be_enabled(lacp_port)) + state |= INFO_STATE_DISTRIBUTING; } if (lacp_port->state == PORT_STATE_EXPIRED) state |= INFO_STATE_EXPIRED;
Add a "col_dist_decoupled" option to the LACP runner that decouples the COLLECTING and DISTRIBUTING states so an external control plane can drive the data path (make-before-break).
With collecting/distributing decoupled a member advertises COLLECTING only after the control plane confirms local ingress is programmed, and DISTRIBUTING only once the partner is collecting. The confirm handshake is fenced by a monotonic request epoch (collecting_request_id / collecting_confirm): the control plane echoes back the request id it programmed for, and teamd latches the confirm only if it matches the current epoch, so a confirm computed for a superseded selection episode (e.g. one that raced a link flap) is rejected. The confirmed epoch persists as the request id advances across self-heal so a stale confirmation stays visible to the control plane.
The selected aggregator is input to every port's decoupled decision, so an aggregator switch refreshes the actor state (and request epoch) of all ports before the kernel port states are updated; without this a port deselected by the switch would keep a stale collecting_requested, letting a confirmation computed for the superseded selection latch and survive into a later reselection.
Signed-off-by: Anisha Halwai anisha@nexthop.ai --- v4 -> v5: the lacp_port_synced() refactor moved to patch 1/3. man/teamd.conf.5 | 12 ++ teamd/teamd_runner_lacp.c | 243 +++++++++++++++++++++++++++++++++++++- 2 files changed, 252 insertions(+), 3 deletions(-)
diff --git a/man/teamd.conf.5 b/man/teamd.conf.5 index 8c65c33..a74efe9 100644 --- a/man/teamd.conf.5 +++ b/man/teamd.conf.5 @@ -282,6 +282,18 @@ Default: .BR "lacp_prio" .RE .TP +.BR "runner.col_dist_decoupled " (bool) +If set, COLLECTING and DISTRIBUTING states are decoupled so an external control plane can drive the data path (make-before-break). A port advertises COLLECTING only after the control plane confirms it by writing the current +.BR "collecting_request_id " +state value back to the +.BR "collecting_confirm " +state value, and advertises DISTRIBUTING (mirrored into the kernel port "enabled" flag) only once confirmed and the link partner is collecting. Writing 0 withdraws the confirmation; a confirm carrying a stale request id is rejected. The confirmation is cleared when the port's own link goes down. +.RS 7 +.PP +Default: +.BR "false" +.RE +.TP .BR "ports.PORTIFNAME.lacp_prio " (int) Port priority according to LACP standard. The lower number means higher priority. .RS 7 diff --git a/teamd/teamd_runner_lacp.c b/teamd/teamd_runner_lacp.c index b8a4c30..ea09ae4 100644 --- a/teamd/teamd_runner_lacp.c +++ b/teamd/teamd_runner_lacp.c @@ -151,6 +151,8 @@ struct lacp { #define LACP_CFG_DFLT_MIN_PORTS_MAX 1024 enum lacp_agg_select_policy agg_select_policy; #define LACP_CFG_DFLT_AGG_SELECT_POLICY LACP_AGG_SELECT_LACP_PRIO + bool col_dist_decoupled; +#define LACP_CFG_DFLT_COL_DIST_DECOUPLED false } cfg; struct teamd_balancer *tb; }; @@ -194,6 +196,13 @@ struct lacp_port { bool sticky; #define LACP_PORT_CFG_DFLT_STICKY false } cfg; + bool collecting_requested; + /* Request->confirm epoch: the control plane echoes + * collecting_request_id back as the confirm; collecting_confirmed_id + * is the confirmed epoch (0 = latch off). + */ + uint32_t collecting_request_id; + uint32_t collecting_confirmed_id; };
static struct lacp_port *lacp_port_get(struct lacp *lacp, @@ -281,6 +290,13 @@ static int lacp_load_config(struct teamd_context *ctx, struct lacp *lacp) lacp->cfg.fast_rate = LACP_CFG_DFLT_FAST_RATE; teamd_log_dbg(ctx, "Using fast_rate "%d".", lacp->cfg.fast_rate);
+ err = teamd_config_bool_get(ctx, &lacp->cfg.col_dist_decoupled, + "$.runner.col_dist_decoupled"); + if (err) + lacp->cfg.col_dist_decoupled = LACP_CFG_DFLT_COL_DIST_DECOUPLED; + teamd_log_dbg(ctx, "Using col_dist_decoupled "%d".", + lacp->cfg.col_dist_decoupled); + err = teamd_config_int_get(ctx, &tmp, "$.runner.min_ports"); if (err) { lacp->cfg.min_ports = LACP_CFG_DFLT_MIN_PORTS; @@ -345,7 +361,14 @@ static bool lacp_port_synced(struct lacp_port *lacp_port)
static int lacp_port_should_be_enabled(struct lacp_port *lacp_port) { - return lacp_port_synced(lacp_port); + struct lacp *lacp = lacp_port->lacp; + + if (!lacp_port_synced(lacp_port)) + return false; + if (lacp->cfg.col_dist_decoupled) + return lacp_port->collecting_confirmed_id != 0 && + (lacp_port->partner.state & INFO_STATE_COLLECTING); + return true; }
static int lacp_port_should_be_disabled(struct lacp_port *lacp_port) @@ -710,18 +733,34 @@ static int lacp_ports_update_enabled(struct lacp *lacp) return 0; }
+static int lacp_ports_actor_refresh(struct lacp *lacp); + static int lacp_selected_agg_update(struct lacp *lacp, struct lacp_port *next_agg_lead) { + bool agg_changed; int err;
if (!next_agg_lead) next_agg_lead = lacp_get_next_agg(lacp); - if (lacp->selected_agg_lead != next_agg_lead) + agg_changed = lacp->selected_agg_lead != next_agg_lead; + if (agg_changed) teamd_log_dbg(lacp->ctx, "Selecting aggregator %u", lacp_agg_id(next_agg_lead)); lacp->selected_agg_lead = next_agg_lead;
+ /* The selected aggregator is input to every port's decoupled + * collecting/distributing decision: refresh actor state (and the + * request epoch) on all ports before mirroring it to the kernel, + * so a stale collecting_requested cannot satisfy a confirm for a + * superseded selection. + */ + if (agg_changed && lacp->cfg.col_dist_decoupled) { + err = lacp_ports_actor_refresh(lacp); + if (err) + return err; + } + err = lacp_ports_update_enabled(lacp); if (err) return err; @@ -918,12 +957,30 @@ static void lacp_port_actor_update(struct lacp_port *lacp_port) state |= INFO_STATE_LACP_ACTIVITY; if (lacp_port->lacp->cfg.fast_rate) state |= INFO_STATE_LACP_TIMEOUT; + if (lacp_port->lacp->cfg.col_dist_decoupled) { + bool new_request = lacp_port_synced(lacp_port); + + if (new_request && !lacp_port->collecting_requested) { + /* New episode: bump the epoch (rejecting stale + * confirms); the latch persists across self-heal, + * cleared only on own-link-down. + */ + if (++lacp_port->collecting_request_id == 0) + lacp_port->collecting_request_id = 1; /* 0 == withdraw sentinel */ + } + lacp_port->collecting_requested = new_request; + } if (lacp_port_selected(lacp_port) && lacp_port_agg_selected(lacp_port)) { state |= INFO_STATE_SYNCHRONIZATION; state &= ~(INFO_STATE_COLLECTING | INFO_STATE_DISTRIBUTING); - if (lacp_port->partner.state & INFO_STATE_SYNCHRONIZATION) + if (lacp_port->lacp->cfg.col_dist_decoupled) { + if (lacp_port->collecting_requested && + lacp_port->collecting_confirmed_id != 0) + state |= INFO_STATE_COLLECTING; + } else if (lacp_port->partner.state & INFO_STATE_SYNCHRONIZATION) { state |= INFO_STATE_COLLECTING; + } if (lacp_port_should_be_enabled(lacp_port)) state |= INFO_STATE_DISTRIBUTING; } @@ -940,6 +997,26 @@ static void lacp_port_actor_update(struct lacp_port *lacp_port)
static int lacpdu_send(struct lacp_port *lacp_port);
+static int lacp_ports_actor_refresh(struct lacp *lacp) +{ + struct teamd_port *tdport; + struct lacp_port *lacp_port; + uint8_t old_state; + int err; + + teamd_for_each_tdport(tdport, lacp->ctx) { + lacp_port = lacp_port_get(lacp, tdport); + old_state = lacp_port->actor.state; + lacp_port_actor_update(lacp_port); + if (lacp_port->actor.state != old_state) { + err = lacpdu_send(lacp_port); + if (err) + return err; + } + } + return 0; +} + static int lacp_port_set_state(struct lacp_port *lacp_port, enum lacp_port_state new_state) { @@ -988,6 +1065,12 @@ static int lacp_port_set_state(struct lacp_port *lacp_port, err = lacp_port_partner_update(lacp_port); if (err) return err; + /* Own link down: clear the latch so a later bring-up needs a + * fresh control-plane confirm (agg (de)selection must not + * clear it). + */ + if (lacp_port->lacp->cfg.col_dist_decoupled) + lacp_port->collecting_confirmed_id = 0; break; }
@@ -1115,6 +1198,19 @@ static int lacpdu_recv(struct lacp_port *lacp_port)
lacp_port_actor_update(lacp_port);
+ /* collecting_requested is (re)computed in actor_update above and the + * partner COLLECTING bit does not affect aggregation, so re-evaluate + * enable + carrier now. + */ + if (lacp_port->lacp->cfg.col_dist_decoupled) { + err = lacp_port_update_enabled(lacp_port); + if (err) + return err; + err = lacp_update_carrier(lacp_port->lacp); + if (err) + return err; + } + /* Check if the other side has correct info about us */ if (memcmp(&lacpdu.partner, &lacp_port->actor, sizeof(struct lacpdu_info))) { @@ -1519,6 +1615,16 @@ static int lacp_state_select_policy_get(struct teamd_context *ctx, return 0; }
+static int lacp_state_col_dist_decoupled_get(struct teamd_context *ctx, + struct team_state_gsc *gsc, + void *priv) +{ + struct lacp *lacp = priv; + + gsc->data.bool_val = lacp->cfg.col_dist_decoupled; + return 0; +} + static const struct teamd_state_val lacp_state_vals[] = { { .subpath = "active", @@ -1540,6 +1646,11 @@ static const struct teamd_state_val lacp_state_vals[] = { .type = TEAMD_STATE_ITEM_TYPE_STRING, .getter = lacp_state_select_policy_get, }, + { + .subpath = "col_dist_decoupled", + .type = TEAMD_STATE_ITEM_TYPE_BOOL, + .getter = lacp_state_col_dist_decoupled_get, + }, };
static struct lacp_port *lacp_port_gsc(struct team_state_gsc *gsc, @@ -1839,6 +1950,106 @@ static int lacp_port_state_prio_get(struct teamd_context *ctx, return 0; }
+static int lacp_port_state_collecting_requested_get(struct teamd_context *ctx, + struct team_state_gsc *gsc, + void *priv) +{ + gsc->data.bool_val = lacp_port_gsc(gsc, priv)->collecting_requested; + return 0; +} + +static int lacp_port_state_collecting_request_id_get(struct teamd_context *ctx, + struct team_state_gsc *gsc, + void *priv) +{ + gsc->data.int_val = lacp_port_gsc(gsc, priv)->collecting_request_id; + return 0; +} + +static int lacp_port_state_collecting_confirm_get(struct teamd_context *ctx, + struct team_state_gsc *gsc, + void *priv) +{ + struct lacp_port *lacp_port = lacp_port_gsc(gsc, priv); + + /* Epoch the control plane last confirmed (0 = none); persists as + * request_id advances so a stale confirm stays visible. + */ + gsc->data.int_val = (int) lacp_port->collecting_confirmed_id; + return 0; +} + +/* Advertised COLLECTING bit (from actor.state), mirroring "distributing"; + * distinct from collecting_confirm (CP latch) and collecting_requested. + */ +static int lacp_port_state_collecting_get(struct teamd_context *ctx, + struct team_state_gsc *gsc, + void *priv) +{ + struct lacp_port *lacp_port = lacp_port_gsc(gsc, priv); + + gsc->data.bool_val = !!(lacp_port->actor.state & INFO_STATE_COLLECTING); + return 0; +} + +static int lacp_port_state_distributing_get(struct teamd_context *ctx, + struct team_state_gsc *gsc, + void *priv) +{ + struct lacp_port *lacp_port = lacp_port_gsc(gsc, priv); + + gsc->data.bool_val = !!(lacp_port->actor.state & INFO_STATE_DISTRIBUTING); + return 0; +} + +/* Control-plane confirm, fenced by the request epoch: writing the current + * request id latches collecting, 0 withdraws, a stale id is rejected. + */ +static int lacp_port_state_collecting_confirm_set(struct teamd_context *ctx, + struct team_state_gsc *gsc, + void *priv) +{ + struct lacp_port *lacp_port = lacp_port_gsc(gsc, priv); + const char *ifname = lacp_port->tdport->ifname; + uint32_t id = (uint32_t) gsc->data.int_val; + int err; + + if (!lacp_port->lacp->cfg.col_dist_decoupled) { + teamd_log_warn("%s: collecting_confirm ignored (not col_dist_decoupled)", + ifname); + return -EOPNOTSUPP; + } + + if (id == 0) { + lacp_port->collecting_confirmed_id = 0; + teamd_log_info("%s: collecting withdrawn by control plane", + ifname); + } else if (id == lacp_port->collecting_request_id && + lacp_port->collecting_requested) { + lacp_port->collecting_confirmed_id = id; + teamd_log_info("%s: collecting confirmed for epoch %u", + ifname, id); + } else { + /* stale epoch or not currently requesting: reject */ + teamd_log_warn("%s: rejecting collecting confirm (epoch %u, " + "current %u, requested %d)", ifname, id, + lacp_port->collecting_request_id, + lacp_port->collecting_requested); + return -EINVAL; + } + + lacp_port_actor_update(lacp_port); + err = lacpdu_send(lacp_port); + if (err) + teamd_log_warn("%s: lacpdu_send after collecting confirm " + "failed: %d", ifname, err); + + err = lacp_port_update_enabled(lacp_port); + if (err) + return err; + return lacp_update_carrier(lacp_port->lacp); +} + static const struct teamd_state_val lacp_port_state_vals[] = { { .subpath = "selected", @@ -1871,6 +2082,32 @@ static const struct teamd_state_val lacp_port_state_vals[] = { .type = TEAMD_STATE_ITEM_TYPE_INT, .getter = lacp_port_state_prio_get, }, + { + .subpath = "collecting_requested", + .type = TEAMD_STATE_ITEM_TYPE_BOOL, + .getter = lacp_port_state_collecting_requested_get, + }, + { + .subpath = "collecting_request_id", + .type = TEAMD_STATE_ITEM_TYPE_INT, + .getter = lacp_port_state_collecting_request_id_get, + }, + { + .subpath = "collecting_confirm", + .type = TEAMD_STATE_ITEM_TYPE_INT, + .getter = lacp_port_state_collecting_confirm_get, + .setter = lacp_port_state_collecting_confirm_set, + }, + { + .subpath = "collecting", + .type = TEAMD_STATE_ITEM_TYPE_BOOL, + .getter = lacp_port_state_collecting_get, + }, + { + .subpath = "distributing", + .type = TEAMD_STATE_ITEM_TYPE_BOOL, + .getter = lacp_port_state_distributing_get, + }, { .subpath = "actor_lacpdu_info", .vals = lacp_port_actor_state_vals,
Add a self-contained test suite (needs root, uses veth pairs and real teamd instances) exercising the col_dist_decoupled make-before-break lifecycle: the collecting confirm handshake against both coupled and decoupled partners, multi-member aggregator behaviour across member flaps, rejection of stale confirms via the request epoch, and an aggregator switch racing a confirmation computed before the switch.
The whole suite runs inside a private, uniquely named network namespace so it cannot touch host networking resources, and confirm acceptance and rejection assert the teamdctl exit status explicitly.
Signed-off-by: Anisha Halwai anisha@nexthop.ai --- v4 -> v5: fixed the copyright header. scripts/team_lacp_col_dist_decoupled_test.py | 619 +++++++++++++++++++ 1 file changed, 619 insertions(+) create mode 100755 scripts/team_lacp_col_dist_decoupled_test.py
diff --git a/scripts/team_lacp_col_dist_decoupled_test.py b/scripts/team_lacp_col_dist_decoupled_test.py new file mode 100755 index 0000000..9ea9025 --- /dev/null +++ b/scripts/team_lacp_col_dist_decoupled_test.py @@ -0,0 +1,619 @@ +#! /usr/bin/env python3 +""" +LACP collecting/distributing decoupling test (self-contained, needs root). + +The col_dist_decoupled option decouples COLLECTING from DISTRIBUTING so an external control +plane can drive the data path (make-before-break): + + * the actor advertises COLLECTING only after the control plane confirms local + ingress is programmed (the "latch": non-zero collecting_confirm), and + * advertises DISTRIBUTING (and mirrors it into the kernel "enabled" flag) only + once the latch is set AND the partner is collecting. + +This suite brings real teamd instances up over veth pairs and drives them +through the decoupled collecting/distributing lifecycle. It is organised into scenarios, each of +which owns its topology. + + Copyright 2026 Nexthop Systems Inc. All rights reserved. + + This library is free software; you can redistribute it and/or + modify it under the terms of the GNU Lesser General Public + License as published by the Free Software Foundation; either + version 2.1 of the License, or (at your option) any later version. + + This library is distributed in the hope that it will be useful, + but WITHOUT ANY WARRANTY; without even the implied warranty of + MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU + Lesser General Public License for more details. + + You should have received a copy of the GNU Lesser General Public + License along with this library; if not, write to the Free Software + Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA +""" + +import sys +import os +import getopt +import json +import subprocess +import time + +# LACP actor/partner state bit used to observe COLLECTING in a partner's view. +INFO_STATE_COLLECTING = 0x10 + +# Everything runs inside a private, uniquely named network namespace so the +# suite cannot touch (or tear down) host networking resources. +NETNS = "lacpdc%d" % os.getpid() + + +class CmdExecFailedException(Exception): + def __init__(self, retval, cmd): + self.retval = retval + self.cmd = cmd + + def __str__(self): + return "Command failed (%d): %s" % (self.retval, self.cmd) + + +def cmd_exec(cmd, cleaner=False, expect_fail=False): + """Run a shell command inside the test netns and return stripped stdout. + cleaner=True silences the echo and tolerates a non-zero exit (teardown). + expect_fail=True asserts the command exits non-zero.""" + cmd = cmd.rstrip(" ") + if not cmd.startswith("ip netns"): + cmd = "ip netns exec %s %s" % (NETNS, cmd) + if not cleaner: + print("# "%s"" % cmd) + subp = subprocess.Popen(cmd, shell=True, stdout=subprocess.PIPE, + stderr=subprocess.PIPE, universal_newlines=True) + (out, err) = subp.communicate() + if expect_fail: + assert subp.returncode != 0, \ + "expected non-zero exit, got success: %s" % cmd + elif subp.returncode and not cleaner: + if out: + print(out) + if err: + print(err) + raise CmdExecFailedException(subp.returncode, cmd) + return out.rstrip() + + +def usage(): + print("Usage: team_lacp_col_dist_decoupled_test.py [OPTION...]") + print("") + print(" -h, --help print this message") + sys.exit() + + +def wait_until(func, timeout=30, interval=0.5, description=""): + """Poll func() -> (matched, detail) until matched, else raise with last detail.""" + deadline = time.time() + timeout + detail = None + while time.time() < deadline: + (matched, detail) = func() + if matched: + return detail + time.sleep(interval) + raise AssertionError("timed out waiting for %s (last=%r)" % (description, detail)) + + +class LacpColDistDecoupledTest: + + def __init__(self): + # Scenario A: decoupled actor with a coupled partner. + self.actor_team = "team_actor" + self.actor_port = "veth_actor" + self.partner_team = "team_partner" + self.partner_port = "veth_partner" + + # Scenario B: two decoupled instances. + self.team_a = "team_a" + self.port_a = "veth_a" + self.team_b = "team_b" + self.port_b = "veth_b" + + # ----------------------------------------------------------------------- + # teamd / kernel plumbing + # ----------------------------------------------------------------------- + + def _build_config(self, team, port, decoupled, fast_rate=True): + runner = {"name": "lacp", "active": True, "fast_rate": fast_rate, "min_ports": 1} + if decoupled: + runner["col_dist_decoupled"] = True + return {"device": team, "runner": runner, + "link_watch": {"name": "ethtool"}, "ports": {port: {}}} + + def _start_teamd(self, team, port, decoupled, fast_rate=True): + config = json.dumps(self._build_config(team, port, decoupled, + fast_rate)) + cmd_exec("teamd -c '%s' -d" % config) + + def _stop_teamd(self, team): + cmd_exec("teamd -t %s -k" % team, cleaner=True) + + def _teamd_state_dump(self, team): + return json.loads(cmd_exec("teamdctl %s state dump" % team, cleaner=True)) + + def _runner(self, team, port): + return self._teamd_state_dump(team)["ports"][port]["runner"] + + def _is_kernel_port_enabled(self, team, port): + output = cmd_exec("teamnl -p %s %s getoption enabled" % (port, team), + cleaner=True) + return output.strip().lower() in ("true", "1") + + def _is_carrier_up(self, team): + return cmd_exec("cat /sys/class/net/%s/carrier" % team, + cleaner=True).strip() == "1" + + # ----------------------------------------------------------------------- + # Small predicates for wait_until(); each returns (matched, detail). + # ----------------------------------------------------------------------- + + def _port_field_is(self, team, port, field, expected): + try: + value = self._runner(team, port)[field] + return (value == expected, value) + except Exception as e: + return (False, str(e)) + + def _port_state_is_current(self, team, port): + return self._port_field_is(team, port, "state", "current") + + def _confirm_epoch(self, team, port, epoch, expect_reject=False): + # Echo an explicit request epoch back to teamd (0 withdraws collecting). + # An accepted confirm must exit zero (cmd_exec raises otherwise); a + # stale epoch must be rejected with a non-zero teamdctl exit status, + # asserted via expect_fail. + cmd_exec("teamdctl %s state item set ports.%s.runner.collecting_confirm %s" % + (team, port, epoch), cleaner=expect_reject, + expect_fail=expect_reject) + + def _set_collecting_confirmed(self, team, port, value): + # The control-plane confirm is fenced by the request epoch: read the + # current collecting_request_id and echo it back. Confirm 0 + # withdraws collecting. + if value: + epoch = self._runner(team, port)["collecting_request_id"] + self._confirm_epoch(team, port, epoch) + else: + self._confirm_epoch(team, port, 0) + + def _confirmed_is(self, team, port, expected): + # (matched, detail) predicate for wait_until: collecting is confirmed + # iff collecting_confirm (the confirmed epoch) is non-zero. + try: + v = self._runner(team, port)["collecting_confirm"] + return ((v != 0) == expected, "collecting_confirm=%r" % v) + except Exception as e: + return (False, str(e)) + + # ----------------------------------------------------------------------- + # Topology setup / teardown + # ----------------------------------------------------------------------- + + def _setup_pair(self, port_a, port_b, team_a, team_b, + team_a_decoupled, team_b_decoupled): + cmd_exec("ip link add %s type veth peer name %s" % (port_a, port_b)) + self._start_teamd(team_a, port_a, team_a_decoupled) + self._start_teamd(team_b, port_b, team_b_decoupled) + cmd_exec("ip link set %s up" % team_a) + cmd_exec("ip link set %s up" % team_b) + wait_until(lambda: self._port_state_is_current(team_a, port_a), + description="%s/%s reaching 'current'" % (team_a, port_a)) + wait_until(lambda: self._port_state_is_current(team_b, port_b), + description="%s/%s reaching 'current'" % (team_b, port_b)) + + def _setup_scenario_a(self): + self._setup_pair(self.actor_port, self.partner_port, + self.actor_team, self.partner_team, + team_a_decoupled=True, team_b_decoupled=False) + + def _setup_scenario_b(self): + self._setup_pair(self.port_a, self.port_b, self.team_a, self.team_b, + team_a_decoupled=True, team_b_decoupled=True) + + def _cleanup(self): + for team in (self.actor_team, self.partner_team, + self.team_a, self.team_b): + self._stop_teamd(team) + # deleting one end of each veth pair removes the whole pair + for port in (self.actor_port, self.port_a): + cmd_exec("ip link del %s" % port, cleaner=True) + + # ======================================================================= + # Scenario A: decoupled actor <-> coupled partner + # + # Walks the full make-before-break handshake: nothing forwards until the + # control plane confirms collecting, distributing then follows the (coupled) + # partner, and an own-link flap clears the latch until re-confirmed. + # ======================================================================= + + def _test_a_col_dist_decoupled_advertised(self): + # Test: the runner reports collecting and distributing are decoupled. + runner = self._teamd_state_dump(self.actor_team)["runner"] + assert runner.get("col_dist_decoupled") is True, \ + "runner.col_dist_decoupled=%r" % runner.get("col_dist_decoupled") + + def _test_a_collecting_requested_by_default(self): + # Test: once the port is up and synced, the actor *requests* collecting + # (selected + agg-selected + partner in sync). + wait_until(lambda: self._port_field_is(self.actor_team, self.actor_port, + "collecting_requested", True), + description="collecting_requested becoming true") + + def _test_a_collecting_inactive_until_confirmed(self): + # COLLECTING is not advertised until the control plane confirms. + assert self._runner(self.actor_team, self.actor_port)["collecting"] is False, \ + "COLLECTING advertised before confirmation" + + def _test_a_kernel_disabled_before_confirm(self): + # Kernel "enabled" is off before collecting is confirmed. + assert not self._is_kernel_port_enabled(self.actor_team, self.actor_port), \ + "kernel port enabled before confirmation while decoupled" + + def _test_a_confirm_advertises_collecting(self): + # Confirming makes the actor advertise COLLECTING, which the partner observes. + self._set_collecting_confirmed(self.actor_team, self.actor_port, True) + wait_until(lambda: (bool(self._runner(self.partner_team, self.partner_port) + ["partner_lacpdu_info"]["state"] & INFO_STATE_COLLECTING), + None), + description="partner observing actor COLLECTING after confirm") + + def _test_a_distributing_follows_partner(self): + # The coupled partner is already collecting, so once confirmed the actor + # starts distributing. + wait_until(lambda: self._port_field_is(self.actor_team, self.actor_port, + "distributing", True), + description="actor distributing after confirm") + + def _test_a_kernel_enabled_after_distributing(self): + # Test: kernel "enabled" mirrors distributing when decoupled. + wait_until(lambda: (self._is_kernel_port_enabled(self.actor_team, + self.actor_port), None), + description="kernel port enabled after collecting/distributing") + + def _test_a_carrier_up_from_distributing(self): + # Test: team carrier is driven by distributing ports, not merely synced + # (kernel-enabled) ports. + wait_until(lambda: (self._is_carrier_up(self.actor_team), None), + description="team carrier up from distributing") + + def _test_a_own_flap_clears_latch(self): + # An own-link flap clears the latch and it stays cleared until the + # control plane re-confirms (no COLLECTING/DISTRIBUTING on reselect). + assert self._runner(self.actor_team, self.actor_port)["collecting_confirm"] != 0, \ + "expected latch set before flap" + cmd_exec("ip link set %s down" % self.actor_port) + wait_until(lambda: self._confirmed_is(self.actor_team, self.actor_port, False), + description="latch cleared on link down") + cmd_exec("ip link set %s up" % self.actor_port) + wait_until(lambda: self._port_field_is(self.actor_team, self.actor_port, + "collecting_requested", True), + description="collecting requested again after link up") + runner = self._runner(self.actor_team, self.actor_port) + assert runner["collecting_confirm"] == 0, \ + "latch re-set on reselect without re-confirm (got %r)" % runner["collecting_confirm"] + assert runner["collecting"] is False and runner["distributing"] is False, \ + "COLLECTING/DISTRIBUTING re-advertised after flap without re-confirm" + + def _test_a_deconfirm_clears_collecting(self): + # Withdrawing the confirmation stops advertising COLLECTING. + self._set_collecting_confirmed(self.actor_team, self.actor_port, True) + wait_until(lambda: self._confirmed_is(self.actor_team, self.actor_port, True), + description="latch re-set before de-confirm") + self._set_collecting_confirmed(self.actor_team, self.actor_port, False) + wait_until(lambda: self._port_field_is(self.actor_team, self.actor_port, + "collecting", False), + description="COLLECTING withdrawn after de-confirm") + + # ======================================================================= + # Scenario B: decoupled actor <-> decoupled partner + # + # With a decoupled partner, distributing must wait for the partner to + # advertise collecting too. + # ======================================================================= + + def _test_b_distributing_off_without_partner(self): + # Confirm our own collecting; distributing stays off because the + # decoupled partner is not collecting. + self._set_collecting_confirmed(self.team_a, self.port_a, True) + wait_until(lambda: self._confirmed_is(self.team_a, self.port_a, True), + description="latch becoming true") + assert self._runner(self.team_a, self.port_a)["distributing"] is False, \ + "distributing should be false while partner not collecting" + + def _test_b_distributing_on_with_partner(self): + # Once the partner confirms collecting too, distributing turns on. + self._set_collecting_confirmed(self.team_b, self.port_b, True) + wait_until(lambda: self._port_field_is(self.team_a, self.port_a, + "distributing", True), + description="distributing after partner collects") + + # ======================================================================= + # Scenario C: multi-member, single aggregator + # ======================================================================= + + def _test_c_member_flap_does_not_disturb_peer(self): + # Test: flapping one member must not disturb the OTHER member's latch + # or forwarding (agg_selected stays stable for the healthy member). + tm, tp = "team_mm", "team_mmp" + a1, a2, b1, b2 = "veth_mm1", "veth_mm2", "veth_mmb1", "veth_mmb2" + M = json.dumps({"device": tm, "runner": {"name": "lacp", "active": True, + "fast_rate": True, "min_ports": 1, "col_dist_decoupled": True}, + "link_watch": {"name": "ethtool"}, "ports": {a1: {}, a2: {}}}) + P = json.dumps({"device": tp, "runner": {"name": "lacp", "active": True, + "fast_rate": True, "min_ports": 1}, + "link_watch": {"name": "ethtool"}, "ports": {b1: {}, b2: {}}}) + cmd_exec("ip link add %s type veth peer name %s" % (a1, b1)) + cmd_exec("ip link add %s type veth peer name %s" % (a2, b2)) + try: + cmd_exec("teamd -c '%s' -d" % M) + cmd_exec("teamd -c '%s' -d" % P) + cmd_exec("ip link set %s up" % tm) + cmd_exec("ip link set %s up" % tp) + for p in (a1, a2): + wait_until(lambda p=p: self._port_state_is_current(tm, p), + description="%s reaching 'current'" % p) + self._set_collecting_confirmed(tm, p, True) + for p in (a1, a2): + wait_until(lambda p=p: self._port_field_is(tm, p, "distributing", True), + description="%s distributing" % p) + # Flap member a2; member a1 must never lose collecting/distributing. + cmd_exec("ip link set %s down" % a2) + cmd_exec("ip link set %s up" % a2) + for _ in range(6): + r = self._runner(tm, a1) + assert r["collecting_confirm"] != 0 and r["distributing"] is True, \ + "peer member a1 disturbed by a2 flap: confirm=%r dist=%r" % \ + (r["collecting_confirm"], r["distributing"]) + time.sleep(1) + finally: + self._stop_teamd(tm) + self._stop_teamd(tp) + cmd_exec("ip link del %s" % a1, cleaner=True) + cmd_exec("ip link del %s" % a2, cleaner=True) + + # ======================================================================= + # Scenario D: request->confirm epoch fencing + # + # The control plane reads (collecting_requested, collecting_request_id), + # programs ingress, then echoes the id back as the confirm. teamd advances + # the request id whenever a new selection episode begins, and latches a + # confirm only if its id matches the current one. A confirm carrying a + # superseded (pre-flap) id must therefore be rejected, so a stale confirm + # racing a link flap cannot re-enable forwarding for an episode whose + # ingress is not programmed. + # ======================================================================= + + def _test_d_stale_confirm_rejected_by_epoch(self): + tm, tp = "team_ep", "team_epp" + a, b = "veth_ep", "veth_epp" + M = json.dumps({"device": tm, "runner": {"name": "lacp", "active": True, + "fast_rate": True, "min_ports": 1, "col_dist_decoupled": True}, + "link_watch": {"name": "ethtool"}, "ports": {a: {}}}) + P = json.dumps({"device": tp, "runner": {"name": "lacp", "active": True, + "fast_rate": True, "min_ports": 1}, + "link_watch": {"name": "ethtool"}, "ports": {b: {}}}) + cmd_exec("ip link add %s type veth peer name %s" % (a, b)) + try: + cmd_exec("teamd -c '%s' -d" % M) + cmd_exec("teamd -c '%s' -d" % P) + cmd_exec("ip link set %s up" % tm) + cmd_exec("ip link set %s up" % tp) + wait_until(lambda: self._port_state_is_current(tm, a), + description="member reaching 'current'") + + # Episode 1: confirm collecting so the member forwards, and capture + # the epoch the control plane would have programmed for. + wait_until(lambda: self._port_field_is(tm, a, "collecting_requested", True), + description="collecting requested (episode 1)") + old_epoch = self._runner(tm, a)["collecting_request_id"] + self._confirm_epoch(tm, a, old_epoch) + wait_until(lambda: self._port_field_is(tm, a, "distributing", True), + description="member forwarding in episode 1") + + # Own-link flap: clears the latch and begins a new selection episode. + cmd_exec("ip link set %s down" % a) + wait_until(lambda: self._port_field_is(tm, a, "collecting_requested", False), + description="collecting request cleared on link down") + cmd_exec("ip link set %s up" % a) + wait_until(lambda: self._port_field_is(tm, a, "collecting_requested", True), + description="collecting requested again (episode 2)") + new_epoch = self._runner(tm, a)["collecting_request_id"] + assert new_epoch > old_epoch, \ + "request epoch must advance across a flap (old=%r new=%r)" % \ + (old_epoch, new_epoch) + + # A confirm carrying the stale (episode-1) epoch must be rejected: + # teamdctl fails and the latch stays clear. + self._confirm_epoch(tm, a, old_epoch, expect_reject=True) + runner = self._runner(tm, a) + assert runner["collecting_confirm"] == 0, \ + "stale confirm (old epoch) must not latch collecting (got %r)" % \ + runner["collecting_confirm"] + assert runner["distributing"] is False, \ + "stale confirm must not enable distributing" + + # A confirm with the current epoch is honored. + self._confirm_epoch(tm, a, new_epoch) + wait_until(lambda: self._port_field_is(tm, a, "distributing", True), + description="member forwards after fresh (current-epoch) confirm") + finally: + self._stop_teamd(tm) + self._stop_teamd(tp) + cmd_exec("ip link del %s" % a, cleaner=True) + + # ======================================================================= + # Runner + # ======================================================================= + + + # ======================================================================= + # Scenario E: aggregator switch refreshes decoupled state + # + # The selected aggregator is input to the decoupled collecting decision; + # switching aggregators (with no event on the deselected port itself) + # must refresh every port's requested state and epoch, so a confirmation + # computed before the switch is rejected after it. + # ======================================================================= + + def _test_e_agg_switch_rejects_racing_confirm(self): + td, tpa, tpb = "team_dut", "team_pa", "team_pb" + d1, d2, d3 = "veth_d1", "veth_d2", "veth_d3" + pa1, pb1, pb2 = "veth_pa1", "veth_pb1", "veth_pb2" + # Slow LACP rate on purpose: periodic PDUs (each of which refreshes + # the receiving port's actor state) are 30s apart, so between the + # aggregator switch and the racing confirm below there is no + # spontaneous refresh on d1 -- only the switch itself may provide it. + D = json.dumps({"device": td, "runner": {"name": "lacp", "active": True, + "min_ports": 1, "col_dist_decoupled": True, + "agg_select_policy": "count"}, + "link_watch": {"name": "ethtool"}, + "ports": {d1: {}, d2: {}, d3: {}}}) + A = json.dumps({"device": tpa, "runner": {"name": "lacp", "active": True, + "min_ports": 1}, + "link_watch": {"name": "ethtool"}, "ports": {pa1: {}}}) + B = json.dumps({"device": tpb, "runner": {"name": "lacp", "active": True, + "min_ports": 1}, + "link_watch": {"name": "ethtool"}, "ports": {pb1: {}, pb2: {}}}) + + def agg_selected(port): + agg = self._runner(td, port).get("aggregator", {}) + return (bool(agg.get("selected")), agg) + + cmd_exec("ip link add %s type veth peer name %s" % (d1, pa1)) + cmd_exec("ip link add %s type veth peer name %s" % (d2, pb1)) + cmd_exec("ip link add %s type veth peer name %s" % (d3, pb2)) + try: + cmd_exec("teamd -c '%s' -d" % D) + cmd_exec("teamd -c '%s' -d" % A) + cmd_exec("teamd -c '%s' -d" % B) + # Keep the two-port aggregator {d2,d3} down: the single-port + # aggregator {d1} is the only candidate and gets selected. + cmd_exec("ip link set %s down" % d2) + cmd_exec("ip link set %s down" % d3) + cmd_exec("ip link set %s up" % td) + cmd_exec("ip link set %s up" % tpa) + cmd_exec("ip link set %s up" % tpb) + wait_until(lambda: agg_selected(d1), + description="aggregator {d1} selected") + wait_until(lambda: self._port_field_is(td, d1, + "collecting_requested", True), + description="d1 requesting collecting") + stale_epoch = self._runner(td, d1)["collecting_request_id"] + + # Bring the second aggregator up: with the "count" policy the + # selection switches to {d2,d3} while d1's own link never flaps. + cmd_exec("ip link set %s up" % d2) + cmd_exec("ip link set %s up" % d3) + wait_until(lambda: agg_selected(d2), + description="selection switching to two-port agg") + + # A confirmation computed for the pre-switch epoch loses the race: + # teamd must reject it (non-zero teamdctl exit) and stay + # unlatched. Fired immediately after the switch: only the + # switch-time refresh can have withdrawn d1's cached request. + self._confirm_epoch(td, d1, stale_epoch, expect_reject=True) + + # The switch alone (no event on d1) must have refreshed d1. + matched, detail = self._port_field_is(td, d1, + "collecting_requested", False) + assert matched, "d1 still requesting after switch: %r" % detail + matched, detail = self._confirmed_is(td, d1, False) + assert matched, "stale confirm latched: %s" % detail + matched, detail = self._port_field_is(td, d1, "collecting", False) + assert matched, "d1 advertising COLLECTING: %r" % detail + assert not self._is_kernel_port_enabled(td, d1), \ + "kernel port d1 enabled after rejected stale confirm" + + # The freshly selected aggregator works normally end-to-end. + self._set_collecting_confirmed(td, d2, True) + wait_until(lambda: self._port_field_is(td, d2, "collecting", True), + description="d2 collecting after fresh confirm") + finally: + for team in (td, tpa, tpb): + self._stop_teamd(team) + for port in (d1, d2, d3): + cmd_exec("ip link del %s" % port, cleaner=True) + + def _scenarios(self): + return [ + ("Scenario A: decoupled actor <-> coupled partner", + self._setup_scenario_a, [ + ("col_dist_decoupled advertised", self._test_a_col_dist_decoupled_advertised), + ("collecting requested by default", self._test_a_collecting_requested_by_default), + ("collecting inactive until confirmed", self._test_a_collecting_inactive_until_confirmed), + ("kernel disabled before confirm", self._test_a_kernel_disabled_before_confirm), + ("confirm advertises COLLECTING", self._test_a_confirm_advertises_collecting), + ("distributing follows partner", self._test_a_distributing_follows_partner), + ("kernel enabled after distributing", self._test_a_kernel_enabled_after_distributing), + ("carrier up from distributing", self._test_a_carrier_up_from_distributing), + ("own-link flap clears latch", self._test_a_own_flap_clears_latch), + ("de-confirm clears COLLECTING", self._test_a_deconfirm_clears_collecting), + ]), + ("Scenario B: decoupled actor <-> decoupled partner", + self._setup_scenario_b, [ + ("distributing off without partner", self._test_b_distributing_off_without_partner), + ("distributing on with partner", self._test_b_distributing_on_with_partner), + ]), + ("Scenario C: multi-member, single aggregator", + None, [ + ("member flap does not disturb peer", self._test_c_member_flap_does_not_disturb_peer), + ]), + ("Scenario D: request->confirm epoch fencing", + None, [ + ("stale confirm rejected by epoch", self._test_d_stale_confirm_rejected_by_epoch), + ]), + ("Scenario E: aggregator switch refreshes decoupled state", + None, [ + ("agg switch rejects racing confirm", self._test_e_agg_switch_rejects_racing_confirm), + ]), + ] + + def _run_testlet(self, name, testlet, failures): + print("--- Test: %s" % name) + try: + testlet() + print(" PASS") + except Exception as e: + print(" FAIL: %s" % e) + failures.append(name) + + def run(self): + failures = [] + cmd_exec("ip netns add %s" % NETNS) + try: + for banner, setup, testlets in self._scenarios(): + print("") + print("=== %s ===" % banner) + if setup: + setup() + for name, testlet in testlets: + self._run_testlet(name, testlet, failures) + finally: + self._cleanup() + cmd_exec("ip netns del %s" % NETNS, cleaner=True) + + print("") + if failures: + raise AssertionError("LACP col_dist_decoupled test FAILED (%d): %s" % + (len(failures), ", ".join(failures))) + print("ALL LACP COL_DIST_DECOUPLED TESTS PASSED") + + +def main(): + try: + opts, args = getopt.getopt(sys.argv[1:], "h", ["help"]) + except getopt.GetoptError as err: + print(str(err)) + usage() + for opt, arg in opts: + if opt in ("-h", "--help"): + usage() + LacpColDistDecoupledTest().run() + + +if __name__ == "__main__": + main()
Wed, Sep 23, 2026 at 10:10:38PM +0200, anisha@nexthop.ai wrote:
The port enable and disable predicates spell out the same condition, that the port is in the selected aggregator and the partner is synchronized, once plainly and once negated. The advertised DISTRIBUTING bit is that same condition a third time.
Factor the condition into lacp_port_synced() and derive lacp_port_should_be_disabled() from lacp_port_should_be_enabled(). lacp_port_agg_selected() already implies a non-NULL agg_lead, so the separate lacp_port_selected() test is dropped.
In preparation for making the enable predicate mode dependent, take the advertised DISTRIBUTING bit from lacp_port_should_be_enabled() instead of computing it alongside COLLECTING, so the advertised state and the kernel port enable decision stay a single decision once they are no longer the same expression.
No functional change intended.
Signed-off-by: Anisha Halwai anisha@nexthop.ai
all 3 applied. Thanks!
libteam@lists.fedorahosted.org
-
Anisha Halwai -
Jiri Pirko