[RFC PATCH net-next 3/9] amt: offload group object, desired state and reconciler

@@ -7,6 +7,7 @@
 #include <linux/skbuff.h>
 #include <linux/udp.h>
 #include <linux/jhash.h>
+#include <linux/hashtable.h>
 #include <linux/if_tunnel.h>
 #include <linux/net.h>
 #include <linux/igmp.h>

@@ -33,6 +34,8 @@
 #include <net/inet_dscp.h>
 #include <net/ip6_checksum.h>
 
+#include <kunit/static_stub.h>
+
 static struct workqueue_struct *amt_wq;
 
 static HLIST_HEAD(source_gc_list);

@@ -304,6 +307,19 @@ static void amt_destroy_source(struct amt_source_node *snode)
 	spin_unlock_bh(&source_gc_lock);
 }
 
+/* Offload desired-state hooks, defined with the rest of the offload code.
+ * Each is a no-op on a relay in software mode.
+ */
+static void amt_ofl_member_add(struct amt_dev *amt,
+			       struct amt_tunnel_list *tunnel,
+			       const union amt_addr *group, bool v6);
+static void amt_ofl_member_del(struct amt_dev *amt,
+			       struct amt_tunnel_list *tunnel,
+			       const union amt_addr *group, bool v6);
+static void amt_ofl_tunnel_gone(struct amt_dev *amt,
+				struct amt_tunnel_list *tunnel);
+static void amt_ofl_state_free(struct amt_dev *amt);
+
 static void amt_del_group(struct amt_dev *amt, struct amt_group_node *gnode)
 {
 	struct amt_source_node *snode;

@@ -315,6 +331,8 @@ static void amt_del_group(struct amt_dev *amt, struct amt_group_node *gnode)
 	rhltable_remove(&amt->groups_rhl, &gnode->rhlnode, amt_gnode_params);
 	list_del(&gnode->tunnel_node);
 	gnode->tunnel_list->nr_groups--;
+	amt_ofl_member_del(amt, gnode->tunnel_list, &gnode->group_addr,
+			   gnode->v6);
 
 	if (!gnode->v6)
 		netdev_dbg(amt->dev, "Leave group %pI4\n",

@@ -569,6 +587,7 @@ static struct amt_group_node *amt_add_group(struct amt_dev *amt,
 	}
 	list_add(&gnode->tunnel_node, &tunnel->groups);
 	tunnel->nr_groups++;
+	amt_ofl_member_add(amt, tunnel, group, v6);
 
 	if (!gnode->v6)
 		netdev_dbg(amt->dev, "Join group %pI4\n",

@@ -1542,6 +1561,17 @@ static void amt_clear_groups(struct amt_tunnel_list *tunnel)
 	spin_unlock_bh(&tunnel->lock);
 }
 
+/* Unlink @tunnel from the relay. Caller holds amt->lock, or RTNL with the
+ * device stopping (amt_dev_stop). The groups stay until amt_clear_groups.
+ */
+static void amt_tunnel_unlink(struct amt_dev *amt,
+			      struct amt_tunnel_list *tunnel)
+{
+	list_del_rcu(&tunnel->list);
+	amt->nr_tunnels--;
+	amt_ofl_tunnel_gone(amt, tunnel);
+}
+
 static void amt_tunnel_expire(struct work_struct *work)
 {
 	struct amt_tunnel_list *tunnel = container_of(to_delayed_work(work),

@@ -1551,8 +1581,7 @@ static void amt_tunnel_expire(struct work_struct *work)
 
 	spin_lock_bh(&amt->lock);
 	rcu_read_lock();
-	list_del_rcu(&tunnel->list);
-	amt->nr_tunnels--;
+	amt_tunnel_unlink(amt, tunnel);
 	amt_clear_groups(tunnel);
 	rcu_read_unlock();
 	spin_unlock_bh(&amt->lock);

@@ -3445,8 +3474,7 @@ static int amt_dev_stop(struct net_device *dev)
 	amt->remote_ip = 0;
 
 	list_for_each_entry_safe(tunnel, tmp, &amt->tunnel_list, list) {
-		list_del_rcu(&tunnel->list);
-		amt->nr_tunnels--;
+		amt_tunnel_unlink(amt, tunnel);
 		cancel_delayed_work_sync(&tunnel->gc_wq);
 		amt_clear_groups(tunnel);
 		kfree_rcu(tunnel, rcu);

@@ -3996,6 +4024,7 @@ static void amt_dev_uninit(struct net_device *dev)
 	/* amt_dev_stop() has removed every tunnel and its groups. */
 	rhltable_destroy(&amt->groups_rhl);
 	gro_cells_destroy(&amt->gro_cells);
+	amt_ofl_state_free(amt);
 }
 
 static const struct net_device_ops amt_netdev_ops = {

@@ -4054,20 +4083,141 @@ struct amt_offload_provider {
 	struct rcu_head			rcu;
 };
 
-/* One provider-owned handle amt must give back on teardown. */
-struct amt_offload_ref {
-	struct list_head		node;
-	void				*handle;
-};
-
 struct amt_offload_bound {
 	struct amt_offload_binding	pub;
 	const struct amt_offload_ops	*ops;
 	struct amt_offload_caps		caps;
 	/* Cleared first on teardown; ops->xmit is only called while set. */
 	bool				admitting;
-	struct list_head		groups;		/* amt_offload_ref */
-	struct list_head		endpoints;	/* amt_offload_ref */
+};
+
+/* Desired state and its reconciler.
+ *
+ *   amt_add_group / amt_del_group      tunnel unlink
+ *   (first / last gnode of a group)    (every group of the tunnel)
+ *                 |                              |
+ *                 v                              v
+ *   +-- desired state, st->lock (bh, a leaf) -------------------------+
+ *   |  grp (family, group) --< member (grp, tunnel) >-- ep (addr,port) |
+ *   |  a member change bumps grp->gen and queues grp on st->dirty      |
+ *   +------------------------------------------------------------------+
+ *                 |  st->work, process context
+ *                 v
+ *   reconciler, st->mutex, never under amt->lock or a tunnel lock:
+ *     per dirty grp: endpoint_add each new ep, group_update(gen),
+ *     group_commit(gen), then publish hw_gen and hw_active; a grp
+ *     with no members is disabled; then unused eps get endpoint_del
+ *                 |
+ *                 v
+ *   provider tables; amt_dev_xmit reads grp->hw_active and hw_gen
+ *
+ * A tick starts no new group once st->budget ops are spent, so one
+ * group may overrun it by its own ops, and endpoint GC runs one
+ * endpoint_del past it when the budget left none, so a group re-dirtied
+ * on every tick cannot starve collection. Changes that land between ticks
+ * coalesce: a group is queued at most once and programmed at its latest
+ * generation. A group whose programming fails is parked and served by
+ * software (the remainder); parked groups are retried only when the
+ * provider holds an eighth fewer objects than when it refused one
+ * (hysteresis), after a reset, or on a new binding. A programming, gone
+ * or GC op overrunning op_timeout_ms resets the provider (flush) and
+ * replays every group from desired state under a new epoch; a result
+ * from an older epoch is ignored. A release op that overruns while amt
+ * gives everything back (teardown, or desired state lost) flushes once
+ * and forgets the remaining handles, with no new epoch and no replay.
+ * Only the reconciler frees grp and ep records, so it may use them
+ * without st->lock between ops.
+ *
+ * Forwarding is per (family, group) and never per source, exactly as the
+ * software fan-out in amt_dev_xmit: a gateway with an INCLUDE(S1)
+ * membership still receives S2. AMT_OFFLOAD_F_SOURCE_FILTER has no v1
+ * meaning.
+ */
+/* Both tables are sized once, from the caps, at state allocation: eps
+ * from max_tunnels, grps from max_tunnels * max_groups. Neither cap has a
+ * changelink, so a load factor of about one holds without a resize; the
+ * upper bound only matters past 64k entries.
+ */
+#define AMT_OFL_HASH_MIN_BITS	4
+#define AMT_OFL_HASH_MAX_BITS	16
+
+/* Default ops per reconciler tick. [estimated]: not measured on hardware. */
+static unsigned int amt_offload_op_budget = 1000;
+module_param_named(offload_op_budget, amt_offload_op_budget, uint, 0444);
+MODULE_PARM_DESC(offload_op_budget,
+		 "Offload programming ops per reconciler tick (default 1000)");
+
+enum amt_ofl_grp_state {
+	AMT_OFL_IDLE,
+	AMT_OFL_DIRTY,		/* on st->dirty */
+	AMT_OFL_PARKED,		/* on st->parked: served by software */
+};
+
+struct amt_ofl_ep {
+	struct hlist_node		hnode;		/* st->eps */
+	struct list_head		gc_node;	/* st->ep_gc */
+	struct amt_offload_endpoint_key	key;
+	u32				refs;		/* members using it */
+	u32				committed;	/* committed grps naming it */
+	bool				on_gc;
+	/* Reconciler-owned, under st->mutex. */
+	struct list_head		hw_node;	/* st->hw_eps */
+	struct amt_offload_endpoint	*hw;
+};
+
+struct amt_ofl_grp {
+	struct hlist_node		hnode;		/* st->grps, RCU */
+	struct list_head		q_node;	/* st->dirty or st->parked */
+	struct amt_offload_group_key	key;
+	struct list_head		members;
+	u32				nr_members;
+	u8				state;
+	u64				gen;		/* desired */
+	u32				left;		/* members removed */
+	/* Reconciler-owned, under st->mutex; the data path reads hw and
+	 * hw_gen once it has read hw_active true (acquire).
+	 */
+	struct list_head		hw_node;	/* st->hw_grps */
+	struct amt_offload_group	*hw;
+	/* The endpoints the last successful group_commit named; each counts
+	 * in its ep->committed, under st->lock.
+	 */
+	struct amt_ofl_ep		**ceps;
+	u32				nr_ceps;
+	u64				hw_gen;
+	bool				hw_active;
+	struct rcu_head			rcu;
+};
+
+struct amt_ofl_member {
+	struct list_head		grp_node;	/* grp->members */
+	struct hlist_node		tun_node;	/* tunnel->offload_members */
+	struct amt_ofl_grp		*grp;
+	struct amt_ofl_ep		*ep;
+	u32				gnodes;	/* the tunnel's gnodes for grp */
+};
+
+struct amt_offload_state {
+	struct amt_dev			*amt;
+	/* Desired state, written from the softirq decap path. */
+	spinlock_t			lock;
+	struct hlist_head		*grps;		/* RCU */
+	struct hlist_head		*eps;
+	u32				grp_bits;
+	u32				ep_bits;
+	struct list_head		dirty;
+	struct list_head		parked;
+	struct list_head		ep_gc;
+	bool				desync;
+	/* Reconciler. bind and teardown also take the mutex. */
+	struct mutex			mutex;
+	struct work_struct		work;
+	struct list_head		hw_grps;
+	struct list_head		hw_eps;
+	u32				hw_objs;
+	bool				enospc;
+	u32				enospc_at;
+	u32				budget;
 };
 
 static LIST_HEAD(amt_offload_providers);

@@ -4146,21 +4296,948 @@ amt_offload_tx_admit(struct amt_dev *amt)
 	return b && READ_ONCE(b->admitting) ? b : NULL;
 }
 
-/* Record a handle the provider handed back; the later reconciler is the
- * production caller.
+static void amt_ofl_grp_key(struct amt_offload_group_key *key,
+			    const union amt_addr *group, bool v6)
+{
+	memset(key, 0, sizeof(*key));
+#if IS_ENABLED(CONFIG_IPV6)
+	if (v6) {
+		key->family = AF_INET6;
+		key->group.ip6 = group->ip6;
+		return;
+	}
+#endif
+	key->family = AF_INET;
+	key->group.ip4 = group->ip4;
+}
+
+static void amt_ofl_ep_key(struct amt_offload_endpoint_key *key,
+			   const struct amt_dev *amt,
+			   const struct amt_tunnel_list *tunnel)
+{
+	memset(key, 0, sizeof(*key));
+	key->port = tunnel->source_port;
+#if IS_ENABLED(CONFIG_IPV6)
+	if (amt_v6(amt)) {
+		key->family = AF_INET6;
+		key->addr.ip6 = tunnel->addr.ip6;
+		return;
+	}
+#endif
+	key->family = AF_INET;
+	key->addr.ip4 = tunnel->addr.ip4;
+}
+
+/* Keys are built zeroed, so memcmp() over the padding is sound. Seeded
+ * per device: gateways choose their groups and ports, so an unseeded hash
+ * lets them pile entries into one chain under st->lock.
+ */
+static u32 amt_ofl_hash(const struct amt_offload_state *st,
+			const void *key, size_t len, u32 bits)
+{
+	return hash_32(jhash(key, len, st->amt->hash_seed), bits);
+}
+
+static struct hlist_head *
+amt_ofl_grp_bucket(const struct amt_offload_state *st,
+		   const struct amt_offload_group_key *key)
+{
+	return &st->grps[amt_ofl_hash(st, key, sizeof(*key), st->grp_bits)];
+}
+
+static struct hlist_head *
+amt_ofl_ep_bucket(const struct amt_offload_state *st,
+		  const struct amt_offload_endpoint_key *key)
+{
+	return &st->eps[amt_ofl_hash(st, key, sizeof(*key), st->ep_bits)];
+}
+
+/* Caller holds st->lock, or rcu_read_lock() for the data path. */
+static struct amt_ofl_grp *
+amt_ofl_grp_find(struct amt_offload_state *st,
+		 const struct amt_offload_group_key *key)
+{
+	struct amt_ofl_grp *g;
+
+	hlist_for_each_entry_rcu(g, amt_ofl_grp_bucket(st, key), hnode,
+				 lockdep_is_held(&st->lock))
+		if (!memcmp(&g->key, key, sizeof(*key)))
+			return g;
+	return NULL;
+}
+
+static struct amt_ofl_ep *
+amt_ofl_ep_find(struct amt_offload_state *st,
+		const struct amt_offload_endpoint_key *key)
+{
+	struct amt_ofl_ep *ep;
+
+	hlist_for_each_entry(ep, amt_ofl_ep_bucket(st, key), hnode)
+		if (!memcmp(&ep->key, key, sizeof(*key)))
+			return ep;
+	return NULL;
+}
+
+/* Queue @g for the reconciler. A parked group stays parked, served by
+ * software, until its retry is due -- unless it lost its last member,
+ * which the reconciler must see to free it. Caller holds st->lock.
+ */
+static void amt_ofl_grp_dirty(struct amt_offload_state *st,
+			      struct amt_ofl_grp *g)
+{
+	if (g->state == AMT_OFL_DIRTY)
+		return;
+	if (g->state == AMT_OFL_PARKED) {
+		if (g->nr_members)
+			return;
+		list_del(&g->q_node);
+	}
+	list_add_tail(&g->q_node, &st->dirty);
+	g->state = AMT_OFL_DIRTY;
+}
+
+static void amt_ofl_kick(struct amt_offload_state *st)
+{
+	queue_work(system_unbound_wq, &st->work);
+}
+
+/* Desired state could not record a change. Hardware would forward to a
+ * stale member set, so the relay leaves offload for good: the data path
+ * stops consulting the provider, and the next reconciler pass gives every
+ * provider object back (group_disable, endpoint_del; the binding stays,
+ * with no unbind). Desync does not schedule that pass: a pass already
+ * queued or a later removal of a recorded member (by a leave, a timer or
+ * a tunnel removal) does, and a teardown gives everything back itself
+ * and unbinds. Caller holds st->lock.
+ */
+static void amt_ofl_desync(struct amt_offload_state *st)
+{
+	if (st->desync)
+		return;
+	WRITE_ONCE(st->desync, true);
+	amt_offload_count(st->amt, AMT_OFFLOAD_CNT_DESYNC);
+}
+
+/* The first gnode of (@group, @v6) on @tunnel makes the tunnel's
+ * endpoint a member; later ones only count. Caller holds tunnel->lock.
+ */
+static void amt_ofl_member_add(struct amt_dev *amt,
+			       struct amt_tunnel_list *tunnel,
+			       const union amt_addr *group, bool v6)
+{
+	struct amt_offload_state *st = amt->offload_st;
+	struct amt_ofl_grp *g, *new_g = NULL;
+	struct amt_ofl_ep *ep, *new_ep = NULL;
+	struct amt_offload_endpoint_key ekey;
+	struct amt_offload_group_key gkey;
+	struct amt_ofl_member *m;
+
+	if (!st)
+		return;
+	amt_ofl_grp_key(&gkey, group, v6);
+	spin_lock_bh(&st->lock);
+	if (st->desync)
+		goto out;
+	hlist_for_each_entry(m, &tunnel->offload_members, tun_node) {
+		if (!memcmp(&m->grp->key, &gkey, sizeof(gkey))) {
+			m->gnodes++;
+			goto out;
+		}
+	}
+
+	g = amt_ofl_grp_find(st, &gkey);
+	if (!g) {
+		new_g = kzalloc_obj(*new_g, GFP_ATOMIC);
+		if (!new_g)
+			goto nomem;
+		g = new_g;
+		g->key = gkey;
+		INIT_LIST_HEAD(&g->members);
+		INIT_LIST_HEAD(&g->hw_node);
+	}
+	amt_ofl_ep_key(&ekey, amt, tunnel);
+	ep = amt_ofl_ep_find(st, &ekey);
+	if (!ep) {
+		new_ep = kzalloc_obj(*new_ep, GFP_ATOMIC);
+		if (!new_ep)
+			goto nomem;
+		ep = new_ep;
+		ep->key = ekey;
+		INIT_LIST_HEAD(&ep->hw_node);
+	}
+	m = kzalloc_obj(*m, GFP_ATOMIC);
+	if (!m)
+		goto nomem;
+
+	if (new_g)
+		hlist_add_head_rcu(&g->hnode, amt_ofl_grp_bucket(st, &gkey));
+	if (new_ep)
+		hlist_add_head(&ep->hnode, amt_ofl_ep_bucket(st, &ekey));
+	m->grp = g;
+	m->ep = ep;
+	m->gnodes = 1;
+	ep->refs++;
+	list_add_tail(&m->grp_node, &g->members);
+	hlist_add_head(&m->tun_node, &tunnel->offload_members);
+	g->nr_members++;
+	g->gen++;
+	amt_ofl_grp_dirty(st, g);
+	spin_unlock_bh(&st->lock);
+	amt_ofl_kick(st);
+	return;
+nomem:
+	/* One lost allocation takes the relay out of offload for good (see
+	 * amt_ofl_desync()). Rebuilding desired state from the tunnel walk
+	 * would avoid that, if GFP_ATOMIC failures ever show up here.
+	 */
+	kfree(new_g);
+	kfree(new_ep);
+	amt_ofl_desync(st);
+out:
+	spin_unlock_bh(&st->lock);
+}
+
+/* The reconciler's member arrays: up to max_tunnels entries, so not one
+ * contiguous block.
+ */
+static void *amt_ofl_alloc_array(u32 n, size_t size)
+{
+#if IS_ENABLED(CONFIG_AMT_KUNIT_TEST)
+	KUNIT_STATIC_STUB_REDIRECT(amt_ofl_alloc_array, n, size);
+#endif
+	return kvmalloc_array(n, size, GFP_KERNEL);
+}
+
+static void amt_ofl_free_array(void *p)
+{
+#if IS_ENABLED(CONFIG_AMT_KUNIT_TEST)
+	KUNIT_STATIC_STUB_REDIRECT(amt_ofl_free_array, p);
+#endif
+	kvfree(p);
+}
+
+/* @ep may be deleted once no member uses it and no committed group names
+ * it: queue it for endpoint GC. Caller holds st->lock.
+ */
+static void amt_ofl_ep_unused(struct amt_offload_state *st,
+			      struct amt_ofl_ep *ep)
+{
+	if (ep->refs || ep->committed || ep->on_gc)
+		return;
+	list_add_tail(&ep->gc_node, &st->ep_gc);
+	ep->on_gc = true;
+}
+
+/* @g's committed member set no longer reaches hardware: its provider
+ * object is disabled or flushed. Caller holds st->mutex.
+ */
+static void amt_ofl_grp_uncommit(struct amt_offload_state *st,
+				 struct amt_ofl_grp *g)
+{
+	struct amt_ofl_ep **ceps = g->ceps;
+	u32 i;
+
+	if (!ceps)
+		return;
+	spin_lock_bh(&st->lock);
+	for (i = 0; i < g->nr_ceps; i++) {
+		ceps[i]->committed--;
+		amt_ofl_ep_unused(st, ceps[i]);
+	}
+	g->ceps = NULL;
+	g->nr_ceps = 0;
+	spin_unlock_bh(&st->lock);
+	amt_ofl_free_array(ceps);
+}
+
+/* Caller holds st->lock. */
+static void amt_ofl_member_unlink(struct amt_offload_state *st,
+				  struct amt_ofl_member *m)
+{
+	struct amt_ofl_grp *g = m->grp;
+	struct amt_ofl_ep *ep = m->ep;
+
+	hlist_del(&m->tun_node);
+	list_del(&m->grp_node);
+	g->nr_members--;
+	g->gen++;
+	g->left++;
+	/* The committed member set still names this endpoint: stop handing
+	 * the group to hardware now, not when the reconciler recommits, so
+	 * no packet admitted after this is replicated to it. Software serves
+	 * the group meanwhile.
+	 */
+	WRITE_ONCE(g->hw_active, false);
+	amt_ofl_grp_dirty(st, g);
+	ep->refs--;
+	amt_ofl_ep_unused(st, ep);
+	kfree(m);
+}
+
+/* The last gnode of (@group, @v6) on @tunnel is gone: the endpoint
+ * leaves the group. Caller holds tunnel->lock.
+ */
+static void amt_ofl_member_del(struct amt_dev *amt,
+			       struct amt_tunnel_list *tunnel,
+			       const union amt_addr *group, bool v6)
+{
+	struct amt_offload_state *st = amt->offload_st;
+	struct amt_offload_group_key gkey;
+	struct amt_ofl_member *m;
+	bool kick = false;
+
+	if (!st)
+		return;
+	amt_ofl_grp_key(&gkey, group, v6);
+	spin_lock_bh(&st->lock);
+	hlist_for_each_entry(m, &tunnel->offload_members, tun_node) {
+		if (memcmp(&m->grp->key, &gkey, sizeof(gkey)))
+			continue;
+		if (!--m->gnodes) {
+			amt_ofl_member_unlink(st, m);
+			kick = true;
+		}
+		break;
+	}
+	spin_unlock_bh(&st->lock);
+	if (kick)
+		amt_ofl_kick(st);
+}
+
+/* @tunnel is torn down or unlinked: its endpoint leaves every group at
+ * once. The gnodes deleted afterwards find no member and do nothing.
+ * Caller holds amt->lock, or RTNL with the device stopping (amt_dev_stop).
+ */
+static void amt_ofl_tunnel_gone(struct amt_dev *amt,
+				struct amt_tunnel_list *tunnel)
+{
+	struct amt_offload_state *st = amt->offload_st;
+	struct amt_ofl_member *m;
+	struct hlist_node *tmp;
+	bool kick;
+
+	if (!st)
+		return;
+	spin_lock_bh(&st->lock);
+	kick = !hlist_empty(&tunnel->offload_members);
+	hlist_for_each_entry_safe(m, tmp, &tunnel->offload_members, tun_node)
+		amt_ofl_member_unlink(st, m);
+	spin_unlock_bh(&st->lock);
+	if (kick)
+		amt_ofl_kick(st);
+}
+
+/* Queue every group with members, parked ones included, for a full
+ * replay: on a new binding and after a reset.
+ */
+static void amt_ofl_requeue_all(struct amt_offload_state *st)
+{
+	struct amt_ofl_grp *g;
+	u32 bkt;
+
+	spin_lock_bh(&st->lock);
+	for (bkt = 0; bkt < 1U << st->grp_bits; bkt++) {
+		hlist_for_each_entry(g, &st->grps[bkt], hnode) {
+			if (g->state == AMT_OFL_PARKED) {
+				list_del(&g->q_node);
+				g->state = AMT_OFL_IDLE;
+			}
+			amt_ofl_grp_dirty(st, g);
+		}
+	}
+	st->enospc = false;
+	spin_unlock_bh(&st->lock);
+}
+
+/* Wait out every ops->xmit admitted before the caller cleared hw_active or
+ * admitting.
+ */
+static void amt_ofl_sync_net(void)
+{
+#if IS_ENABLED(CONFIG_AMT_KUNIT_TEST)
+	KUNIT_STATIC_STUB_REDIRECT(amt_ofl_sync_net);
+#endif
+	synchronize_net();
+}
+
+/* Forget every provider handle; the provider was flushed. Each group is
+ * unpublished too, so a later binding never sees hw_active over a NULL
+ * handle. Caller holds st->mutex.
+ */
+static void amt_ofl_hw_forget(struct amt_offload_state *st)
+{
+	struct amt_ofl_grp *g, *gt;
+	struct amt_ofl_ep *ep, *et;
+
+	list_for_each_entry_safe(g, gt, &st->hw_grps, hw_node) {
+		WRITE_ONCE(g->hw_active, false);
+		amt_ofl_grp_uncommit(st, g);
+		g->hw = NULL;
+		list_del_init(&g->hw_node);
+	}
+	list_for_each_entry_safe(ep, et, &st->hw_eps, hw_node) {
+		ep->hw = NULL;
+		list_del_init(&ep->hw_node);
+	}
+	st->hw_objs = 0;
+}
+
+/* An op issued at @t0 overran op_timeout_ms: the provider is wedged. */
+static bool amt_ofl_op_overran(struct amt_offload_bound *b, ktime_t t0)
+{
+	u32 tmo = b->caps.op_timeout_ms;
+
+#if IS_ENABLED(CONFIG_AMT_KUNIT_TEST)
+	KUNIT_STATIC_STUB_REDIRECT(amt_ofl_op_overran, b, t0);
+#endif
+	return tmo && ktime_ms_delta(ktime_get(), t0) > tmo;
+}
+
+/* Give every provider object back with group_disable and endpoint_del,
+ * groups first. A release op that overruns op_timeout_ms ends it: the
+ * provider is flushed once and the remaining handles are forgotten, so
+ * a wedged provider costs one op plus one flush, not one op per object,
+ * under RTNL at teardown. @quiesced: the caller already stopped
+ * admission and waited out ops->xmit. Caller holds st->mutex.
  */
-static int __maybe_unused amt_offload_track(struct list_head *list,
-					    void *handle)
+static void amt_ofl_hw_release(struct amt_offload_state *st,
+			       struct amt_offload_bound *b, bool quiesced)
 {
-	struct amt_offload_ref *ref = kzalloc(sizeof(*ref), GFP_KERNEL);
+	struct amt_ofl_grp *g, *gt;
+	struct amt_ofl_ep *ep, *et;
+	bool overran;
+	ktime_t t0;
+
+	lockdep_assert_held(&st->mutex);
+	if (!quiesced && !list_empty(&st->hw_grps)) {
+		list_for_each_entry(g, &st->hw_grps, hw_node)
+			WRITE_ONCE(g->hw_active, false);
+		amt_ofl_sync_net();
+	}
+	list_for_each_entry_safe(g, gt, &st->hw_grps, hw_node) {
+		WRITE_ONCE(g->hw_active, false);
+		t0 = ktime_get();
+		b->ops->group_disable(&b->pub, g->hw);
+		overran = amt_ofl_op_overran(b, t0);
+		amt_ofl_grp_uncommit(st, g);
+		g->hw = NULL;
+		list_del_init(&g->hw_node);
+		if (overran)
+			goto wedged;
+	}
+	list_for_each_entry_safe(ep, et, &st->hw_eps, hw_node) {
+		t0 = ktime_get();
+		b->ops->endpoint_del(&b->pub, ep->hw);
+		overran = amt_ofl_op_overran(b, t0);
+		ep->hw = NULL;
+		list_del_init(&ep->hw_node);
+		if (overran)
+			goto wedged;
+	}
+	st->hw_objs = 0;
+	return;
+wedged:
+	b->ops->flush(&b->pub);
+	amt_ofl_hw_forget(st);
+	amt_offload_count(st->amt, AMT_OFFLOAD_CNT_RESET);
+}
+
+/* An op overran op_timeout_ms: the provider is treated as wedged. Stop
+ * forwarding through it, flush its tables, forget every handle, bump the
+ * epoch so anything it still returns is stale, and replay. Caller holds
+ * st->mutex.
+ */
+static void amt_ofl_reset(struct amt_offload_state *st,
+			  struct amt_offload_bound *b)
+{
+	struct amt_ofl_grp *g;
+
+	lockdep_assert_held(&st->mutex);
+	list_for_each_entry(g, &st->hw_grps, hw_node)
+		WRITE_ONCE(g->hw_active, false);
+	amt_ofl_sync_net();
+	b->ops->flush(&b->pub);
+	amt_ofl_hw_forget(st);
+	b->pub.epoch = ++st->amt->offload_epoch;
+	amt_offload_count(st->amt, AMT_OFFLOAD_CNT_RESET);
+	amt_ofl_requeue_all(st);
+}
+
+/* Completion check for an op issued under @epoch; @overran was sampled
+ * right after the provider call returned, so amt's own bookkeeping is not
+ * charged to the provider. False means its result must be ignored -- the
+ * op overran op_timeout_ms (the provider is reset here) or the epoch moved
+ * under it.
+ */
+static bool amt_ofl_op_done(struct amt_offload_state *st,
+			    struct amt_offload_bound *b, u64 epoch, bool overran)
+{
+	if (overran)
+		amt_ofl_reset(st, b);
+	if (b->pub.epoch == epoch)
+		return true;
+	amt_offload_count(st->amt, AMT_OFFLOAD_CNT_STALE);
+	return false;
+}
+
+/* Stop forwarding @g in hardware and free its provider object. The caller
+ * cleared hw_active and waited out ops->xmit, or the group never had it.
+ * Returns whether group_disable overran op_timeout_ms.
+ */
+static bool amt_ofl_grp_hw_free(struct amt_offload_state *st,
+				struct amt_offload_bound *b,
+				struct amt_ofl_grp *g)
+{
+	ktime_t t0 = ktime_get();
+	bool overran;
+
+	b->ops->group_disable(&b->pub, g->hw);
+	overran = amt_ofl_op_overran(b, t0);
+	amt_ofl_grp_uncommit(st, g);
+	g->hw = NULL;
+	list_del_init(&g->hw_node);
+	st->hw_objs--;
+	return overran;
+}
 
-	if (!ref)
+/* Retry parked groups once the provider holds an eighth fewer objects
+ * than when it refused one. Refused with nothing programmed, only a
+ * reset or a new binding retries them. Caller holds st->mutex.
+ */
+static void amt_ofl_unpark(struct amt_offload_state *st)
+{
+	struct amt_ofl_grp *g, *t;
+
+	if (!st->enospc || !st->enospc_at ||
+	    st->hw_objs + max(st->enospc_at / 8, 1U) > st->enospc_at)
+		return;
+	spin_lock_bh(&st->lock);
+	list_for_each_entry_safe(g, t, &st->parked, q_node) {
+		list_del(&g->q_node);
+		g->state = AMT_OFL_IDLE;
+		amt_ofl_grp_dirty(st, g);
+	}
+	st->enospc = false;
+	spin_unlock_bh(&st->lock);
+}
+
+/* Programming @g failed: software serves it from now until a retry. Its
+ * endpoints stay programmed; they may serve other groups.
+ */
+static unsigned int amt_ofl_grp_park(struct amt_offload_state *st,
+				     struct amt_offload_bound *b,
+				     struct amt_ofl_grp *g)
+{
+	unsigned int ops = 0;
+
+	if (g->hw) {
+		/* Wait out ops->xmit even when hw_active already reads false:
+		 * amt_ofl_member_unlink() clears it from softirq with no grace
+		 * period, so a reader admitted before may still hold g->hw.
+		 */
+		WRITE_ONCE(g->hw_active, false);
+		amt_ofl_sync_net();
+		amt_ofl_grp_hw_free(st, b, g);
+		ops++;
+	}
+	if (!st->enospc) {
+		st->enospc = true;
+		st->enospc_at = st->hw_objs;
+	}
+	amt_offload_count(st->amt, AMT_OFFLOAD_CNT_SW_REMAINDER);
+	spin_lock_bh(&st->lock);
+	if (g->state == AMT_OFL_DIRTY && !g->nr_members) {
+		/* Emptied meanwhile: let the reconciler free it. */
+	} else {
+		if (g->state == AMT_OFL_DIRTY)
+			list_del(&g->q_node);
+		list_add_tail(&g->q_node, &st->parked);
+		g->state = AMT_OFL_PARKED;
+	}
+	spin_unlock_bh(&st->lock);
+	return ops;
+}
+
+/* Program @g at its current generation: add its endpoints that are not
+ * in hardware yet, stage the member set, commit, publish. Returns the
+ * ops spent. Caller holds st->mutex; @g is off every queue.
+ */
+static unsigned int amt_ofl_grp_program(struct amt_offload_state *st,
+					struct amt_offload_bound *b,
+					struct amt_ofl_grp *g)
+{
+	struct amt_offload_endpoint **hws = NULL;
+	struct amt_ofl_ep **eps = NULL;
+	struct amt_offload_group *hw;
+	struct amt_ofl_member *m;
+	unsigned int ops = 0;
+	u32 n, i;
+	ktime_t t0;
+	u64 epoch;
+	u32 left;
+	u64 gen;
+	int err;
+
+	for (;;) {
+		spin_lock_bh(&st->lock);
+		n = g->nr_members;
+		spin_unlock_bh(&st->lock);
+		if (!n)
+			goto requeue;
+		eps = amt_ofl_alloc_array(n, sizeof(*eps));
+		hws = amt_ofl_alloc_array(n, sizeof(*hws));
+		if (!eps || !hws)
+			goto park;
+		spin_lock_bh(&st->lock);
+		if (g->nr_members == n)
+			break;
+		spin_unlock_bh(&st->lock);
+		/* Membership moved under the allocation. The next pass may
+		 * find none and leave through out:, which frees both again.
+		 */
+		amt_ofl_free_array(eps);
+		amt_ofl_free_array(hws);
+		eps = NULL;
+		hws = NULL;
+	}
+	i = 0;
+	list_for_each_entry(m, &g->members, grp_node)
+		eps[i++] = m->ep;
+	gen = g->gen;
+	left = g->left;
+	spin_unlock_bh(&st->lock);
+
+	if (g->hw && READ_ONCE(g->hw_active) && g->hw_gen == gen)
+		goto out;	/* a replay of what is already committed */
+
+	for (i = 0; i < n; i++) {
+		struct amt_offload_endpoint *ehw = NULL;
+
+		if (!eps[i]->hw) {
+			if (!READ_ONCE(b->admitting))
+				goto out;	/* teardown gives it all back */
+			epoch = b->pub.epoch;
+			t0 = ktime_get();
+			err = b->ops->endpoint_add(&b->pub, &eps[i]->key, &ehw);
+			ops++;
+			if (!amt_ofl_op_done(st, b, epoch, amt_ofl_op_overran(b, t0)))
+				goto out;	/* reset: replayed */
+			if (err)
+				goto park;
+			eps[i]->hw = ehw;
+			list_add_tail(&eps[i]->hw_node, &st->hw_eps);
+			st->hw_objs++;
+		}
+		hws[i] = eps[i]->hw;
+	}
+
+	if (!READ_ONCE(b->admitting))
+		goto out;
+	hw = g->hw;
+	epoch = b->pub.epoch;
+	t0 = ktime_get();
+	err = b->ops->group_update(&b->pub, &g->key, hws, n, gen, &hw);
+	ops++;
+	if (!amt_ofl_op_done(st, b, epoch, amt_ofl_op_overran(b, t0)))
+		goto out;
+	if (!g->hw && hw) {
+		g->hw = hw;
+		list_add_tail(&g->hw_node, &st->hw_grps);
+		st->hw_objs++;
+	}
+	if (err)
+		goto park;
+
+	if (!READ_ONCE(b->admitting))
+		goto out;
+	epoch = b->pub.epoch;
+	t0 = ktime_get();
+	err = b->ops->group_commit(&b->pub, g->hw, gen);
+	ops++;
+	if (!amt_ofl_op_done(st, b, epoch, amt_ofl_op_overran(b, t0)))
+		goto out;
+	if (err)
+		goto park;
+
+	/* Generation rule: packets submitted from here on carry @gen.
+	 * Release pairs with the acquire in amt_offload_tx(): a reader that
+	 * sees hw_active also sees hw and hw_gen. Publish only if no member
+	 * was removed since the snapshot: it is still in what was just
+	 * committed, and amt_ofl_member_unlink() already withdrew the group.
+	 * A join meanwhile does not hold the publication back: the committed
+	 * set is a subset that names no departed endpoint. Either change
+	 * re-queued @g; the next pass commits it.
+	 */
+	spin_lock_bh(&st->lock);
+	for (i = 0; i < n; i++)
+		eps[i]->committed++;
+	for (i = 0; i < g->nr_ceps; i++) {
+		g->ceps[i]->committed--;
+		amt_ofl_ep_unused(st, g->ceps[i]);
+	}
+	swap(g->ceps, eps);	/* out: frees the old set */
+	g->nr_ceps = n;
+	WRITE_ONCE(g->hw_gen, gen);
+	if (g->left == left)
+		smp_store_release(&g->hw_active, true);
+	spin_unlock_bh(&st->lock);
+	goto out;
+park:
+	ops += amt_ofl_grp_park(st, b, g);
+	goto out;
+requeue:
+	spin_lock_bh(&st->lock);
+	amt_ofl_grp_dirty(st, g);
+	spin_unlock_bh(&st->lock);
+out:
+	amt_ofl_free_array(eps);
+	amt_ofl_free_array(hws);
+	return ops;
+}
+
+/* Disable the unhashed groups on @gone: one grace period per flush, then
+ * disable. A group holds hw only while bound, so @b is set when any does.
+ */
+static void amt_ofl_gone_flush(struct amt_offload_state *st,
+			       struct amt_offload_bound *b,
+			       struct list_head *gone)
+{
+	struct amt_ofl_grp *g, *gt;
+	u64 e;
+
+	if (list_empty(gone))
+		return;
+	amt_ofl_sync_net();
+	list_for_each_entry_safe(g, gt, gone, q_node) {
+		list_del(&g->q_node);
+		/* A reset earlier in this flush or tick already flushed it. */
+		if (g->hw) {
+			e = b->pub.epoch;
+			amt_ofl_op_done(st, b, e, amt_ofl_grp_hw_free(st, b, g));
+		}
+		kfree_rcu(g, rcu);
+	}
+}
+
+/* Is a group with @key on @gone, its provider object still live? */
+static bool amt_ofl_gone_has(struct list_head *gone,
+			     const struct amt_offload_group_key *key)
+{
+	struct amt_ofl_grp *g;
+
+	list_for_each_entry(g, gone, q_node)
+		if (!memcmp(&g->key, key, sizeof(*key)))
+			return true;
+	return false;
+}
+
+/* One reconciler tick. Returns true while work remains. */
+static bool amt_ofl_reconcile(struct amt_dev *amt)
+{
+	struct amt_offload_state *st = amt->offload_st;
+	struct amt_ofl_grp *g;
+	struct amt_offload_bound *b;
+	LIST_HEAD(gone);
+	unsigned int spent = 0;
+	unsigned int budget;
+	struct amt_ofl_ep *ep;
+	bool gc_del = false;
+	u64 epoch = 0, e;
+	ktime_t t0;
+	bool more;
+
+	budget = max(READ_ONCE(st->budget), 1U);
+	mutex_lock(&st->mutex);
+	b = rcu_dereference_protected(amt->offload,
+				      lockdep_is_held(&st->mutex));
+	if (b && !READ_ONCE(b->admitting)) {
+		/* Teardown is under way and gives everything back. */
+		mutex_unlock(&st->mutex);
+		return false;
+	}
+	if (b && READ_ONCE(st->desync)) {
+		amt_ofl_hw_release(st, b, false);
+		b = NULL;
+	}
+	if (b)
+		epoch = b->pub.epoch;
+
+	while (spent < budget) {
+		/* A teardown waiting on st->mutex holds RTNL, and a reset
+		 * replays everything anyway: end the pass at either, and let
+		 * the work re-kick, rather than spend the whole budget
+		 * against a provider that may be wedged.
+		 */
+		if (b && (!READ_ONCE(b->admitting) || b->pub.epoch != epoch))
+			break;
+		spin_lock_bh(&st->lock);
+		g = list_first_entry_or_null(&st->dirty, struct amt_ofl_grp,
+					     q_node);
+		if (!g) {
+			spin_unlock_bh(&st->lock);
+			break;
+		}
+		list_del(&g->q_node);
+		g->state = AMT_OFL_IDLE;
+		if (!g->nr_members) {
+			hlist_del_init_rcu(&g->hnode);
+			spin_unlock_bh(&st->lock);
+			if (g->hw) {
+				WRITE_ONCE(g->hw_active, false);
+				list_add_tail(&g->q_node, &gone);
+				spent++;
+			} else {
+				kfree_rcu(g, rcu);
+			}
+			continue;
+		}
+		spin_unlock_bh(&st->lock);
+		if (!b)
+			continue;
+		/* A rejoin of a group emptied earlier in this tick hashed a
+		 * new object for the same key while the old one's provider
+		 * group is still live: disable that one first, or the
+		 * provider would hold two groups for one (family, group).
+		 */
+		if (amt_ofl_gone_has(&gone, &g->key)) {
+			amt_ofl_gone_flush(st, b, &gone);
+			/* A reset in that flush requeued @g: replay it. */
+			if (b->pub.epoch != epoch)
+				break;
+		}
+		spent += amt_ofl_grp_program(st, b, g);
+	}
+
+	amt_ofl_gone_flush(st, b, &gone);
+
+	/* Unused endpoints: no member uses one and no committed group names
+	 * it (ep->committed), so it goes whatever is still dirty. At least
+	 * one endpoint_del per tick, or a group re-dirtied on every pass
+	 * would spend the budget and starve it.
+	 */
+	for (;;) {
+		if (b && !READ_ONCE(b->admitting))
+			break;
+		spin_lock_bh(&st->lock);
+		ep = spent < budget || !gc_del ?
+		     list_first_entry_or_null(&st->ep_gc, struct amt_ofl_ep,
+					      gc_node) : NULL;
+		if (!ep) {
+			spin_unlock_bh(&st->lock);
+			break;
+		}
+		list_del(&ep->gc_node);
+		ep->on_gc = false;
+		if (ep->refs || ep->committed) {
+			/* Queued again when both reach zero. */
+			spin_unlock_bh(&st->lock);
+			continue;
+		}
+		hlist_del_init(&ep->hnode);
+		spin_unlock_bh(&st->lock);
+		if (ep->hw) {
+			list_del_init(&ep->hw_node);
+			st->hw_objs--;
+			e = b->pub.epoch;
+			t0 = ktime_get();
+			b->ops->endpoint_del(&b->pub, ep->hw);
+			spent++;
+			gc_del = true;
+			amt_ofl_op_done(st, b, e, amt_ofl_op_overran(b, t0));
+		}
+		kfree(ep);
+	}
+
+	if (b)
+		amt_ofl_unpark(st);
+	spin_lock_bh(&st->lock);
+	more = !list_empty(&st->dirty) || !list_empty(&st->ep_gc);
+	spin_unlock_bh(&st->lock);
+	mutex_unlock(&st->mutex);
+	return more;
+}
+
+static void amt_ofl_work(struct work_struct *work)
+{
+	struct amt_offload_state *st = container_of(work,
+						    struct amt_offload_state,
+						    work);
+
+	if (amt_ofl_reconcile(st->amt))
+		amt_ofl_kick(st);
+}
+
+static int amt_ofl_state_alloc(struct amt_dev *amt)
+{
+	struct amt_offload_state *st;
+
+	st = kzalloc_obj(*st, GFP_KERNEL);
+	if (!st)
 		return -ENOMEM;
-	ref->handle = handle;
-	list_add_tail(&ref->node, list);
+	st->ep_bits = clamp_t(u32, order_base_2(amt->max_tunnels),
+			      AMT_OFL_HASH_MIN_BITS, AMT_OFL_HASH_MAX_BITS);
+	/* Clamp in 64 bits: order_base_2() takes an unsigned long, which
+	 * would truncate the product on a 32-bit kernel.
+	 */
+	st->grp_bits = clamp_t(u32, order_base_2(min_t(u64,
+					(u64)amt->max_tunnels * amt->max_groups,
+					1U << AMT_OFL_HASH_MAX_BITS)),
+			       AMT_OFL_HASH_MIN_BITS, AMT_OFL_HASH_MAX_BITS);
+	st->grps = kvcalloc(1U << st->grp_bits, sizeof(*st->grps), GFP_KERNEL);
+	st->eps = kvcalloc(1U << st->ep_bits, sizeof(*st->eps), GFP_KERNEL);
+	if (!st->grps || !st->eps) {
+		kvfree(st->grps);
+		kvfree(st->eps);
+		kfree(st);
+		return -ENOMEM;
+	}
+	st->amt = amt;
+	spin_lock_init(&st->lock);
+	INIT_LIST_HEAD(&st->dirty);
+	INIT_LIST_HEAD(&st->parked);
+	INIT_LIST_HEAD(&st->ep_gc);
+	mutex_init(&st->mutex);
+	INIT_WORK(&st->work, amt_ofl_work);
+	INIT_LIST_HEAD(&st->hw_grps);
+	INIT_LIST_HEAD(&st->hw_eps);
+	st->budget = READ_ONCE(amt_offload_op_budget);
+	amt->offload_st = st;
 	return 0;
 }
 
+/* ndo_uninit: every tunnel is gone and the binding with it, so nothing
+ * here is in hardware. Members are freed without touching their tunnels.
+ */
+static void amt_ofl_state_free(struct amt_dev *amt)
+{
+	struct amt_offload_state *st = amt->offload_st;
+	struct amt_ofl_member *m, *mt;
+	struct hlist_node *tmp;
+	struct amt_ofl_grp *g;
+	struct amt_ofl_ep *ep;
+	u32 bkt;
+
+	if (!st)
+		return;
+	cancel_work_sync(&st->work);
+	for (bkt = 0; bkt < 1U << st->grp_bits; bkt++) {
+		hlist_for_each_entry_safe(g, tmp, &st->grps[bkt], hnode) {
+			list_for_each_entry_safe(m, mt, &g->members, grp_node)
+				kfree(m);
+			amt_ofl_free_array(g->ceps);
+			kfree(g);
+		}
+	}
+	for (bkt = 0; bkt < 1U << st->ep_bits; bkt++)
+		hlist_for_each_entry_safe(ep, tmp, &st->eps[bkt], hnode)
+			kfree(ep);
+	mutex_destroy(&st->mutex);
+	kvfree(st->grps);
+	kvfree(st->eps);
+	kfree(st);
+	amt->offload_st = NULL;
+}
+
 static void amt_offload_set_carrier(struct amt_dev *amt, bool on)
 {
 	if (amt->offload_mode != AMT_OFFLOAD_MODE_REQUIRED)