Hi Jiri,
Thanks for the review! The review comments have been addressed, here are the 2 patches. "col_dist_decoupled" was a great suggestion, adopted that instead of "independent mode". Please take a look.
v2: - split the test suite into a separate patch - renamed independent_mode -> col_dist_decoupled (option, state key, internal identifiers, docs and test)
Anisha Halwai (2): teamd: lacp: Add per-runner col_dist_decoupled option teamd: lacp: Add col_dist_decoupled test suite
man/teamd.conf.5 | 12 + scripts/team_lacp_col_dist_decoupled_test.py | 512 +++++++++++++++++++ teamd/teamd_runner_lacp.c | 231 ++++++++- 3 files changed, 754 insertions(+), 1 deletion(-) create mode 100755 scripts/team_lacp_col_dist_decoupled_test.py
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.
Signed-off-by: Anisha Halwai anisha@nexthop.ai --- man/teamd.conf.5 | 12 ++ teamd/teamd_runner_lacp.c | 231 +++++++++++++++++++++++++++++++++++++- 2 files changed, 242 insertions(+), 1 deletion(-)
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 3b4f7b1..4ccd99d 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; @@ -337,10 +353,31 @@ static bool lacp_port_selected(struct lacp_port *lacp_port) return lacp_port->agg_lead; }
+/* Whether teamd currently wants this member to collect (computed live, not + * from the cached collecting_requested field). + */ +static bool lacp_port_decoupled_collecting_requested(struct lacp_port *lacp_port) +{ + return lacp_port_selected(lacp_port) && + lacp_port_agg_selected(lacp_port) && + (lacp_port->partner.state & INFO_STATE_SYNCHRONIZATION); +} + +static bool lacp_port_decoupled_distributing_ready(struct lacp_port *lacp_port) +{ + return lacp_port_decoupled_collecting_requested(lacp_port) && + lacp_port->collecting_confirmed_id != 0 && + (lacp_port->partner.state & INFO_STATE_COLLECTING); +} + static int lacp_port_should_be_enabled(struct lacp_port *lacp_port) { struct lacp *lacp = lacp_port->lacp;
+ if (lacp->cfg.col_dist_decoupled) + /* With collecting/distributing decoupled, kernel "enabled" mirrors distributing. */ + return lacp_port_decoupled_distributing_ready(lacp_port); + if (lacp_port_selected(lacp_port) && lacp_port->agg_lead == lacp->selected_agg_lead && lacp_port->partner.state & INFO_STATE_SYNCHRONIZATION) @@ -352,6 +389,9 @@ static int lacp_port_should_be_disabled(struct lacp_port *lacp_port) { struct lacp *lacp = lacp_port->lacp;
+ if (lacp->cfg.col_dist_decoupled) + return !lacp_port_decoupled_distributing_ready(lacp_port); + if (!lacp_port_selected(lacp_port) || lacp_port->agg_lead != lacp->selected_agg_lead || !(lacp_port->partner.state & INFO_STATE_SYNCHRONIZATION)) @@ -924,7 +964,8 @@ 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_selected(lacp_port) && + if (!lacp_port->lacp->cfg.col_dist_decoupled && + lacp_port_selected(lacp_port) && lacp_port_agg_selected(lacp_port)) { state |= INFO_STATE_SYNCHRONIZATION; state &= ~(INFO_STATE_COLLECTING | INFO_STATE_DISTRIBUTING); @@ -932,6 +973,34 @@ static void lacp_port_actor_update(struct lacp_port *lacp_port) state |= INFO_STATE_COLLECTING | INFO_STATE_DISTRIBUTING; } + if (lacp_port->lacp->cfg.col_dist_decoupled) { + bool new_request = + lacp_port_decoupled_collecting_requested(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->collecting_requested && + lacp_port->collecting_confirmed_id != 0) + state |= INFO_STATE_COLLECTING; + /* Advertised DISTRIBUTING must match the kernel-enable + * decision; both use the one helper so they cannot drift. + */ + if (lacp_port_decoupled_distributing_ready(lacp_port)) + state |= INFO_STATE_DISTRIBUTING; + } + } if (lacp_port->state == PORT_STATE_EXPIRED) state |= INFO_STATE_EXPIRED; if (lacp_port->state == PORT_STATE_DEFAULTED) @@ -993,6 +1062,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; }
@@ -1120,6 +1195,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))) { @@ -1524,6 +1612,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", @@ -1545,6 +1643,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, @@ -1844,6 +1947,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", @@ -1876,6 +2079,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 "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 --- v2 -> v3 (addressing review): - an aggregator switch now refreshes actor state and the request epoch on all ports before kernel port states are updated, so a confirmation computed for a superseded selection is rejected man/teamd.conf.5 | 12 ++ teamd/teamd_runner_lacp.c | 269 +++++++++++++++++++++++++++++++++++++- 2 files changed, 279 insertions(+), 2 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 3b4f7b1..ba52993 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; @@ -337,10 +353,31 @@ static bool lacp_port_selected(struct lacp_port *lacp_port) return lacp_port->agg_lead; }
+/* Whether teamd currently wants this member to collect (computed live, not + * from the cached collecting_requested field). + */ +static bool lacp_port_decoupled_collecting_requested(struct lacp_port *lacp_port) +{ + return lacp_port_selected(lacp_port) && + lacp_port_agg_selected(lacp_port) && + (lacp_port->partner.state & INFO_STATE_SYNCHRONIZATION); +} + +static bool lacp_port_decoupled_distributing_ready(struct lacp_port *lacp_port) +{ + return lacp_port_decoupled_collecting_requested(lacp_port) && + lacp_port->collecting_confirmed_id != 0 && + (lacp_port->partner.state & INFO_STATE_COLLECTING); +} + static int lacp_port_should_be_enabled(struct lacp_port *lacp_port) { struct lacp *lacp = lacp_port->lacp;
+ if (lacp->cfg.col_dist_decoupled) + /* With collecting/distributing decoupled, kernel "enabled" mirrors distributing. */ + return lacp_port_decoupled_distributing_ready(lacp_port); + if (lacp_port_selected(lacp_port) && lacp_port->agg_lead == lacp->selected_agg_lead && lacp_port->partner.state & INFO_STATE_SYNCHRONIZATION) @@ -352,6 +389,9 @@ static int lacp_port_should_be_disabled(struct lacp_port *lacp_port) { struct lacp *lacp = lacp_port->lacp;
+ if (lacp->cfg.col_dist_decoupled) + return !lacp_port_decoupled_distributing_ready(lacp_port); + if (!lacp_port_selected(lacp_port) || lacp_port->agg_lead != lacp->selected_agg_lead || !(lacp_port->partner.state & INFO_STATE_SYNCHRONIZATION)) @@ -716,18 +756,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; @@ -924,7 +980,8 @@ 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_selected(lacp_port) && + if (!lacp_port->lacp->cfg.col_dist_decoupled && + lacp_port_selected(lacp_port) && lacp_port_agg_selected(lacp_port)) { state |= INFO_STATE_SYNCHRONIZATION; state &= ~(INFO_STATE_COLLECTING | INFO_STATE_DISTRIBUTING); @@ -932,6 +989,34 @@ static void lacp_port_actor_update(struct lacp_port *lacp_port) state |= INFO_STATE_COLLECTING | INFO_STATE_DISTRIBUTING; } + if (lacp_port->lacp->cfg.col_dist_decoupled) { + bool new_request = + lacp_port_decoupled_collecting_requested(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->collecting_requested && + lacp_port->collecting_confirmed_id != 0) + state |= INFO_STATE_COLLECTING; + /* Advertised DISTRIBUTING must match the kernel-enable + * decision; both use the one helper so they cannot drift. + */ + if (lacp_port_decoupled_distributing_ready(lacp_port)) + state |= INFO_STATE_DISTRIBUTING; + } + } if (lacp_port->state == PORT_STATE_EXPIRED) state |= INFO_STATE_EXPIRED; if (lacp_port->state == PORT_STATE_DEFAULTED) @@ -945,6 +1030,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) { @@ -993,6 +1098,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; }
@@ -1120,6 +1231,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))) { @@ -1524,6 +1648,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", @@ -1545,6 +1679,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, @@ -1844,6 +1983,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", @@ -1876,6 +2115,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,
Wed, Sep 02, 2026 at 07:48:58PM +0200, anisha@nexthop.ai wrote:
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
v2 -> v3 (addressing review):
- an aggregator switch now refreshes actor state and the request epoch on
all ports before kernel port states are updated, so a confirmation computed for a superseded selection is rejected man/teamd.conf.5 | 12 ++ teamd/teamd_runner_lacp.c | 269 +++++++++++++++++++++++++++++++++++++- 2 files changed, 279 insertions(+), 2 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 3b4f7b1..ba52993 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;
@@ -337,10 +353,31 @@ static bool lacp_port_selected(struct lacp_port *lacp_port) return lacp_port->agg_lead; }
+/* Whether teamd currently wants this member to collect (computed live, not
- from the cached collecting_requested field).
Some AI leftover I assume.
- */
+static bool lacp_port_decoupled_collecting_requested(struct lacp_port *lacp_port) +{
- return lacp_port_selected(lacp_port) &&
lacp_port_agg_selected(lacp_port) &&(lacp_port->partner.state & INFO_STATE_SYNCHRONIZATION);+}
+static bool lacp_port_decoupled_distributing_ready(struct lacp_port *lacp_port) +{
- return lacp_port_decoupled_collecting_requested(lacp_port) &&
lacp_port->collecting_confirmed_id != 0 &&(lacp_port->partner.state & INFO_STATE_COLLECTING);+}
static int lacp_port_should_be_enabled(struct lacp_port *lacp_port) { struct lacp *lacp = lacp_port->lacp;
- if (lacp->cfg.col_dist_decoupled)
/* With collecting/distributing decoupled, kernel "enabled" mirrors distributing. */return lacp_port_decoupled_distributing_ready(lacp_port);
The function lacp_port_decoupled_collecting_requested is duplicating some of the checks here. Could the whole check be folded here instead of having a separate path in decoupled case?
- if (lacp_port_selected(lacp_port) && lacp_port->agg_lead == lacp->selected_agg_lead && lacp_port->partner.state & INFO_STATE_SYNCHRONIZATION)
@@ -352,6 +389,9 @@ static int lacp_port_should_be_disabled(struct lacp_port *lacp_port) { struct lacp *lacp = lacp_port->lacp;
- if (lacp->cfg.col_dist_decoupled)
return !lacp_port_decoupled_distributing_ready(lacp_port);
Same here.
- if (!lacp_port_selected(lacp_port) || lacp_port->agg_lead != lacp->selected_agg_lead || !(lacp_port->partner.state & INFO_STATE_SYNCHRONIZATION))
@@ -716,18 +756,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;
@@ -924,7 +980,8 @@ 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_selected(lacp_port) &&
- if (!lacp_port->lacp->cfg.col_dist_decoupled &&
lacp_port_selected(lacp_port) && lacp_port_agg_selected(lacp_port)) {@@ -932,6 +989,34 @@ static void lacp_port_actor_update(struct lacp_port *lacp_port) state |= INFO_STATE_COLLECTING | INFO_STATE_DISTRIBUTING; }
- if (lacp_port->lacp->cfg.col_dist_decoupled) {
bool new_request =lacp_port_decoupled_collecting_requested(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->collecting_requested &&lacp_port->collecting_confirmed_id != 0)state |= INFO_STATE_COLLECTING;/* Advertised DISTRIBUTING must match the kernel-enable* decision; both use the one helper so they cannot drift.*/if (lacp_port_decoupled_distributing_ready(lacp_port))state |= INFO_STATE_DISTRIBUTING;}
Again, lots of duplications. Could be folded?
- } if (lacp_port->state == PORT_STATE_EXPIRED) state |= INFO_STATE_EXPIRED; if (lacp_port->state == PORT_STATE_DEFAULTED)
@@ -945,6 +1030,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) { @@ -993,6 +1098,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;@@ -1120,6 +1231,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))) {
@@ -1524,6 +1648,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", @@ -1545,6 +1679,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, @@ -1844,6 +1983,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", @@ -1876,6 +2115,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,- },
How does user know request_id changed? He polls?
- {
.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,
-- 2.43.0
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, and rejection of stale confirms via the request epoch.
Signed-off-by: Anisha Halwai anisha@nexthop.ai --- scripts/team_lacp_col_dist_decoupled_test.py | 512 +++++++++++++++++++ 1 file changed, 512 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..957caf3 --- /dev/null +++ b/scripts/team_lacp_col_dist_decoupled_test.py @@ -0,0 +1,512 @@ +#! /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 (C) 2012-2015 Jiri Pirko jiri@resnulli.us + + 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 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 + + +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): + """Run a shell command and return stripped stdout. cleaner=True silences + the echo and tolerates a non-zero exit (teardown / expected failures).""" + cmd = cmd.rstrip(" ") + 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 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). + # A stale epoch is rejected (teamdctl exits non-zero); pass + # expect_reject=True so that expected failure does not abort the test. + cmd_exec("teamdctl %s state item set ports.%s.runner.collecting_confirm %s" % + (team, port, epoch), cleaner=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 + # ======================================================================= + + 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), + ]), + ] + + 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 = [] + self._cleanup() + 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() + + 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()
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 --- v2 -> v3 (addressing review): - the suite runs inside a private, uniquely named network namespace and no longer pre-cleans (or can touch) host networking resources - confirm acceptance/rejection now assert the teamdctl exit status - new scenario: aggregator switch racing a confirmation computed before the switch 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..ce68ac9 --- /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 (C) 2012-2015 Jiri Pirko jiri@resnulli.us + + 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, Aug 26, 2026 at 04:20:20AM +0200, anisha@nexthop.ai wrote:
Hi Jiri,
Thanks for the review! The review comments have been addressed, here are the 2 patches. "col_dist_decoupled" was a great suggestion, adopted that instead of "independent mode". Please take a look.
Hi Anisha,
I ran the v2 series through a Codex-assisted review. It identified the following issues that look worth investigating:
1. Aggregator changes may not refresh the per-port decoupled state.
`collecting_requested`, `collecting_request_id`, and the actor COLLECTING/DISTRIBUTING bits are updated by `lacp_port_actor_update()`. However, `lacp_selected_agg_update()` can change the selected aggregator and update the kernel enabled state for every port without calling `lacp_port_actor_update()` for all affected ports.
Codex suggests this could temporarily make the advertised state disagree with the kernel state and could leave a stale `collecting_requested` value, allowing an obsolete confirmation to pass the setter's cached state check. A test switching between two distinct aggregators, with a confirmation racing the switch, would help confirm or dismiss this.
2. The test cleanup may affect networking resources it did not create.
The test uses fixed interface and team names, and calls `_cleanup()` before creating its topology. That cleanup unconditionally stops matching teamd instances and deletes matching veth devices. Since the test runs as root, Codex flagged a possible collision with unrelated host networking.
Using a uniquely named network namespace, or tracking and removing only resources created by this invocation, would avoid that risk.
3. The stale-confirm test may not verify the rejection return status.
`_confirm_epoch(..., expect_reject=True)` selects `cleaner=True`, which tolerates command failure but does not expose the exit status. The later state checks verify that the confirmation did not latch, but not that teamdctl actually returned an error as intended.
It may be worth explicitly checking for a non-zero status on stale confirmations and zero on accepted confirmations.
Thanks, Jiri
Hi Jiri,
Thanks for the review.
Some AI leftover I assume.
I assume this was for the comment? I added it there for readability initially, but it's better if it's removed.
The function lacp_port_decoupled_collecting_requested is duplicating some of the checks here. Could the whole check be folded here instead of having a separate path in decoupled case?
Yes. In version 4 the enable check is one path with a decoupled tail on a new mode-neutral helper, lacp_port_synced(), and lacp_port_should_be_disabled() reduces to !lacp_port_should_be_enabled() (the two were exact complements in both modes already). Both decoupled_* helpers are gone.
Again, lots of duplications. Could be folded?
Folded: a single selected/agg_selected block now sets SYNCHRONIZATION, only the COLLECTING condition branches by mode, and DISTRIBUTING is derived from lacp_port_should_be_enabled() in both modes, so the advertised bit and the kernel enable decision are the same function.
How does user know request_id changed? He polls?
Yes, the teamd state dump is periodically polled by the control plane to get the latest port changes. This is the same model as the existing lacp_port_state_vals items. The epoch is designed for an asynchronous, possibly-stale observer: a confirm against a superseded epoch is rejected, and the control plane re-reads the current epoch and confirms again.
The version 4 source patch is inline below; the test patch is unchanged since version 3.
Thanks, Anisha
-- >8 -- From: Anisha Halwai anisha@nexthop.ai Date: Tue, 1 Sep 2026 22:39:49 +0000 Subject: [PATCH v4] teamd: lacp: Add per-runner col_dist_decoupled option
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 --- v3 -> v4 (addressing review): - removed the leftover comment - folded the enable/disable checks and actor state generation; both decoupled_* helpers replaced by one mode-neutral lacp_port_synced() man/teamd.conf.5 | 12 ++ teamd/teamd_runner_lacp.c | 264 +++++++++++++++++++++++++++++++++++--- 2 files changed, 260 insertions(+), 16 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 3b4f7b1..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; @@ -337,26 +353,27 @@ static bool lacp_port_selected(struct lacp_port *lacp_port) return lacp_port->agg_lead; }
+static bool lacp_port_synced(struct lacp_port *lacp_port) +{ + return lacp_port_agg_selected(lacp_port) && + (lacp_port->partner.state & INFO_STATE_SYNCHRONIZATION); +} + static int lacp_port_should_be_enabled(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; + 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) { - 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) @@ -716,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; @@ -924,13 +957,32 @@ 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) - state |= INFO_STATE_COLLECTING | - INFO_STATE_DISTRIBUTING; + 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; } if (lacp_port->state == PORT_STATE_EXPIRED) state |= INFO_STATE_EXPIRED; @@ -945,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) { @@ -993,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; }
@@ -1120,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))) { @@ -1524,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", @@ -1545,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, @@ -1844,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", @@ -1876,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,
Wed, Sep 16, 2026 at 02:45:12AM +0200, anisha@nexthop.ai wrote:
Hi Jiri,
Thanks for the review.
Some AI leftover I assume.
I assume this was for the comment? I added it there for readability initially, but it's better if it's removed.
Don't assume. Use some sane email client.
The function lacp_port_decoupled_collecting_requested is duplicating some of the checks here. Could the whole check be folded here instead of having a separate path in decoupled case?
Yes. In version 4 the enable check is one path with a decoupled tail on a new mode-neutral helper, lacp_port_synced(), and lacp_port_should_be_disabled() reduces to !lacp_port_should_be_enabled() (the two were exact complements in both modes already). Both decoupled_* helpers are gone.
Again, lots of duplications. Could be folded?
Folded: a single selected/agg_selected block now sets SYNCHRONIZATION, only the COLLECTING condition branches by mode, and DISTRIBUTING is derived from lacp_port_should_be_enabled() in both modes, so the advertised bit and the kernel enable decision are the same function.
How does user know request_id changed? He polls?
Yes, the teamd state dump is periodically polled by the control plane to get the latest port changes. This is the same model as the existing lacp_port_state_vals items. The epoch is designed for an asynchronous, possibly-stale observer: a confirm against a superseded epoch is rejected, and the control plane re-reads the current epoch and confirms again.
The version 4 source patch is inline below; the test patch is unchanged since version 3.
Send it on the list the nomal way. Please not as a reply.
Thanks, Anisha
-- >8 -- From: Anisha Halwai anisha@nexthop.ai Date: Tue, 1 Sep 2026 22:39:49 +0000 Subject: [PATCH v4] teamd: lacp: Add per-runner col_dist_decoupled option
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
v3 -> v4 (addressing review):
- removed the leftover comment
- folded the enable/disable checks and actor state generation; both
decoupled_* helpers replaced by one mode-neutral lacp_port_synced() man/teamd.conf.5 | 12 ++ teamd/teamd_runner_lacp.c | 264 +++++++++++++++++++++++++++++++++++--- 2 files changed, 260 insertions(+), 16 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 3b4f7b1..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;
@@ -337,26 +353,27 @@ static bool lacp_port_selected(struct lacp_port *lacp_port) return lacp_port->agg_lead; }
+static bool lacp_port_synced(struct lacp_port *lacp_port) +{
- return lacp_port_agg_selected(lacp_port) &&
(lacp_port->partner.state & INFO_STATE_SYNCHRONIZATION);+}
static int lacp_port_should_be_enabled(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;
- 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) {
- 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) @@ -716,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;
@@ -924,13 +957,32 @@ 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)state |= INFO_STATE_COLLECTING |INFO_STATE_DISTRIBUTING;
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;@@ -945,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) { @@ -993,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;@@ -1120,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))) {
@@ -1524,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", @@ -1545,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, @@ -1844,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", @@ -1876,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,
-- 2.43.0
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 --- v3 -> v4 (addressing review): - removed the leftover comment - folded the enable/disable checks and actor state generation; both decoupled_* helpers replaced by one mode-neutral lacp_port_synced() man/teamd.conf.5 | 12 ++ teamd/teamd_runner_lacp.c | 264 +++++++++++++++++++++++++++++++++++--- 2 files changed, 260 insertions(+), 16 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 3b4f7b1..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; @@ -337,26 +353,27 @@ static bool lacp_port_selected(struct lacp_port *lacp_port) return lacp_port->agg_lead; }
+static bool lacp_port_synced(struct lacp_port *lacp_port) +{ + return lacp_port_agg_selected(lacp_port) && + (lacp_port->partner.state & INFO_STATE_SYNCHRONIZATION); +} + static int lacp_port_should_be_enabled(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; + 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) { - 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) @@ -716,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; @@ -924,13 +957,32 @@ 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) - state |= INFO_STATE_COLLECTING | - INFO_STATE_DISTRIBUTING; + 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; } if (lacp_port->state == PORT_STATE_EXPIRED) state |= INFO_STATE_EXPIRED; @@ -945,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) { @@ -993,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; }
@@ -1120,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))) { @@ -1524,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", @@ -1545,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, @@ -1844,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", @@ -1876,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,
libteam@lists.fedorahosted.org
-
Anisha Halwai -
Jiri Pirko