[RFC PATCH net-next 4/9] amt: offload TX handoff, ingress punt rule and counters

@@ -319,6 +319,8 @@ static void amt_ofl_member_del(struct amt_dev *amt,
 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 bool amt_offload_tx(struct amt_dev *amt, struct sk_buff *skb,
+			   const union amt_addr *group, bool v6);
 
 static void amt_del_group(struct amt_dev *amt, struct amt_group_node *gnode)
 {

@@ -1510,6 +1512,11 @@ static netdev_tx_t amt_dev_xmit(struct sk_buff *skb, struct net_device *dev)
 	} else if (amt->mode == AMT_MODE_RELAY) {
 		if (!data)
 			goto free;
+		/* A bound offload provider replicates it, or declines; see
+		 * amt_offload_tx() for the cutover rules.
+		 */
+		if (amt_offload_tx(amt, skb, &group, v6))
+			return NETDEV_TX_OK;
 		key.group_addr = group;
 		key.v6 = v6;
 		list_for_each_entry_rcu(tunnel, &amt->tunnel_list, list) {

@@ -4087,6 +4094,7 @@ struct amt_offload_bound {
 	struct amt_offload_binding	pub;
 	const struct amt_offload_ops	*ops;
 	struct amt_offload_caps		caps;
+	enum amt_offload_path		path;
 	/* Cleared first on teardown; ops->xmit is only called while set. */
 	bool				admitting;
 };

@@ -4209,6 +4217,10 @@ struct amt_offload_state {
 	struct list_head		parked;
 	struct list_head		ep_gc;
 	bool				desync;
+	/* The binding is on the ingress path: a member change must not
+	 * clear hw_active, which there gates the punt-drop. Set at bind.
+	 */
+	bool				ingress;
 	/* Reconciler. bind and teardown also take the mutex. */
 	struct mutex			mutex;
 	struct work_struct		work;

@@ -4286,9 +4298,9 @@ static u32 amt_offload_outer_family(const struct amt_dev *amt)
 }
 
 /* The data path's way in: a binding is usable only while admitting. The
- * caller holds rcu_read_lock(). The later TX handoff is its first caller.
+ * caller holds rcu_read_lock().
  */
-static __maybe_unused struct amt_offload_bound *
+static struct amt_offload_bound *
 amt_offload_tx_admit(struct amt_dev *amt)
 {
 	struct amt_offload_bound *b = rcu_dereference(amt->offload);

@@ -4402,13 +4414,15 @@ static void amt_ofl_kick(struct amt_offload_state *st)
 }
 
 /* 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.
+ * stale member set, so the relay leaves offload for good. On TX that
+ * takes effect at once: the data path stops handing packets to the
+ * provider. 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. On an
+ * ingress binding the provider replicates to its committed sets until
+ * then. Caller holds st->lock.
  */
 static void amt_ofl_desync(struct amt_offload_state *st)
 {

@@ -4562,12 +4576,16 @@ static void amt_ofl_member_unlink(struct amt_offload_state *st,
 	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.
+	/* TX: 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. Ingress: hardware replicates
+	 * on its own until the recommit whatever the host does, and
+	 * hw_active there only drops the punted copies, so clearing it
+	 * would send the remaining members every packet twice.
 	 */
-	WRITE_ONCE(g->hw_active, false);
+	if (!READ_ONCE(st->ingress))
+		WRITE_ONCE(g->hw_active, false);
 	amt_ofl_grp_dirty(st, g);
 	ep->refs--;
 	amt_ofl_ep_unused(st, ep);

@@ -4966,7 +4984,10 @@ static unsigned int amt_ofl_grp_program(struct amt_offload_state *st,
 	 * 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.
+	 * re-queued @g; the next pass commits it. On ingress, hardware is
+	 * already replicating to the committed set whatever the host does,
+	 * and hw_active only drops the punted copies, so withholding it there
+	 * protects no one and sends the remaining members every packet twice.
 	 */
 	spin_lock_bh(&st->lock);
 	for (i = 0; i < n; i++)

@@ -4978,7 +4999,7 @@ static unsigned int amt_ofl_grp_program(struct amt_offload_state *st,
 	swap(g->ceps, eps);	/* out: frees the old set */
 	g->nr_ceps = n;
 	WRITE_ONCE(g->hw_gen, gen);
-	if (g->left == left)
+	if (READ_ONCE(st->ingress) || g->left == left)
 		smp_store_release(&g->hw_active, true);
 	spin_unlock_bh(&st->lock);
 	goto out;

@@ -5238,6 +5259,135 @@ static void amt_ofl_state_free(struct amt_dev *amt)
 	amt->offload_st = NULL;
 }
 
+/* Cutover, per packet; exactly one actor handles each one.
+ *
+ *   relay data packet (inner IP; Ethernet pulled)
+ *    |
+ *    +-- bound, admitting, desired state in sync, and (family, group)
+ *    |   hw_active?  no --> software fan-out
+ *    |
+ *    +-- TX path: ops->xmit(grp, hw_gen)
+ *    |     CONSUMED --> the provider handled every copy and owns the skb
+ *    |     FALLBACK --> skb untouched --> software fan-out
+ *    |     BUSY     --> skb untouched --> software fan-out, counted
+ *    |                  (never NETDEV_TX_BUSY: the relay is IFF_NO_QUEUE)
+ *    |
+ *    +-- ingress path: a punted copy of an hw_active group reaching here
+ *          is dropped and counted; one that took the "no" branch is
+ *          forwarded in software. A provider cannot tell which (amt
+ *          publishes after group_commit returns, and withdraws before
+ *          the callback on a reset, park or teardown and with no
+ *          callback on desync), so it must never punt a packet it
+ *          replicates.
+ *
+ * hw_gen is read once per packet. The reconciler writes it after
+ * group_commit returns 0, before it publishes hw_active, and also when a
+ * member left meanwhile and it withholds hw_active. So a packet already
+ * past the hw_active test may carry that new generation, and one that
+ * read hw_gen earlier any older committed one; the provider serves or
+ * declines it.
+ *
+ * Returns true when the skb is gone (consumed, or a punted copy dropped);
+ * false sends it to the software fan-out untouched.
+ */
+static bool amt_offload_tx(struct amt_dev *amt, struct sk_buff *skb,
+			   const union amt_addr *group, bool v6)
+{
+	struct amt_offload_group_key key;
+	enum amt_offload_result res;
+	struct amt_offload_bound *b;
+	struct amt_ofl_grp *g;
+	bool gone = false;
+
+	rcu_read_lock();
+	b = amt_offload_tx_admit(amt);
+	if (!b || READ_ONCE(amt->offload_st->desync))
+		goto out;
+	amt_ofl_grp_key(&key, group, v6);
+	g = amt_ofl_grp_find(amt->offload_st, &key);
+	/* Pairs with the release in amt_ofl_grp_program(). */
+	if (!g || !smp_load_acquire(&g->hw_active))
+		goto out;
+
+	if (b->path == AMT_OFFLOAD_PATH_INGRESS) {
+		amt_offload_count(amt, AMT_OFFLOAD_CNT_PUNT_DROPPED);
+		consume_skb(skb);
+		gone = true;
+		goto out;
+	}
+
+	res = b->ops->xmit(&b->pub, g->hw, skb, READ_ONCE(g->hw_gen));
+	switch (res) {
+	case AMT_OFFLOAD_CONSUMED:
+		gone = true;
+		break;
+	case AMT_OFFLOAD_BUSY:
+		amt_offload_count(amt, AMT_OFFLOAD_CNT_BUSY_FALLBACK);
+		break;
+	case AMT_OFFLOAD_FALLBACK:
+		break;
+	default:
+		WARN_ONCE(1, "amt: offload xmit returned %d\n", res);
+		break;
+	}
+out:
+	rcu_read_unlock();
+	return gone;
+}
+
+/* One gateway's copy for a simulated provider, built and sent exactly
+ * as amt_send_multicast_data() builds and sends a software copy: the
+ * same copy, AMT Multicast Data header, UDP ports, TOS and hop limit,
+ * through amt_udp_xmit(). Called from ops->xmit, in softirq context
+ * under rcu_read_lock().
+ */
+int amt_offload_encap_xmit(struct amt_offload_binding *b,
+			   const struct amt_offload_endpoint_key *key,
+			   struct sk_buff *oskb)
+{
+	struct amt_dev *amt = netdev_priv(b->dev);
+	struct amt_header_mcast_data *amtmd;
+	bool v6 = ip_hdr(oskb)->version == 6;
+	union amt_addr daddr = {0,};
+	struct socket *sock;
+	struct sk_buff *skb;
+
+	sock = rcu_dereference_bh(amt->sock);
+	if (!sock)
+		return -ENETDOWN;
+	if (key->family != (amt_v6(amt) ? AF_INET6 : AF_INET))
+		return -EAFNOSUPPORT;
+
+	skb = skb_copy_expand(oskb, sizeof(*amtmd) + amt_ip_hlen(amt) +
+			      sizeof(struct udphdr), 0, GFP_ATOMIC);
+	if (!skb)
+		goto drop;
+
+	skb_reset_inner_headers(skb);
+	amtmd = skb_push(skb, sizeof(*amtmd));
+	amtmd->version = 0;
+	amtmd->reserved = 0;
+	amtmd->type = AMT_MSG_MULTICAST_DATA;
+	skb_set_inner_protocol(skb, htons(v6 ? ETH_P_IPV6 : ETH_P_IP));
+
+#if IS_ENABLED(CONFIG_IPV6)
+	if (key->family == AF_INET6)
+		daddr.ip6 = key->addr.ip6;
+	else
+#endif
+		daddr.ip4 = key->addr.ip4;
+	if (amt_udp_xmit(amt, sock->sk, skb, &daddr, amt->relay_port,
+			 key->port, 0))
+		goto free;
+	return 0;
+free:
+	kfree_skb(skb);
+drop:
+	DEV_STATS_INC(amt->dev, tx_dropped);
+	return -ENETUNREACH;
+}
+EXPORT_SYMBOL_GPL(amt_offload_encap_xmit);
+
 static void amt_offload_set_carrier(struct amt_dev *amt, bool on)
 {
 	if (amt->offload_mode != AMT_OFFLOAD_MODE_REQUIRED)

@@ -5285,9 +5435,27 @@ static int amt_offload_bind(struct amt_dev *amt,
 		kfree(b);
 		return -EOPNOTSUPP;
 	}
+	/* TX when offered. Ingress only for a required relay: a packet
+	 * consumed on ingress never reaches the host, so nothing on its
+	 * forwarding path sees it, and auto never chooses that.
+	 */
+	if ((b->caps.paths & BIT(AMT_OFFLOAD_PATH_TX)) && b->ops->xmit) {
+		b->path = AMT_OFFLOAD_PATH_TX;
+	} else if ((b->caps.paths & BIT(AMT_OFFLOAD_PATH_INGRESS)) &&
+		   amt->offload_mode == AMT_OFFLOAD_MODE_REQUIRED) {
+		b->path = AMT_OFFLOAD_PATH_INGRESS;
+	} else {
+		b->ops->unbind(&b->pub);
+		amt_offload_count(amt, AMT_OFFLOAD_CNT_PATH_UNSUPPORTED);
+		kfree(b);
+		return -EOPNOTSUPP;
+	}
 
 	b->admitting = true;
 	mutex_lock(&amt->offload_st->mutex);
+	/* Before any group can hold hw under this binding. */
+	WRITE_ONCE(amt->offload_st->ingress,
+		   b->path == AMT_OFFLOAD_PATH_INGRESS);
 	rcu_assign_pointer(amt->offload, b);
 	/* Replay desired state: every group is programmed afresh. */
 	amt_ofl_requeue_all(amt->offload_st);

@@ -5506,6 +5674,9 @@ AMT_OFFLOAD_CNT_ATTR(sw_remainder, AMT_OFFLOAD_CNT_SW_REMAINDER);
 AMT_OFFLOAD_CNT_ATTR(resets, AMT_OFFLOAD_CNT_RESET);
 AMT_OFFLOAD_CNT_ATTR(stale_completions, AMT_OFFLOAD_CNT_STALE);
 AMT_OFFLOAD_CNT_ATTR(desync, AMT_OFFLOAD_CNT_DESYNC);
+AMT_OFFLOAD_CNT_ATTR(path_unsupported, AMT_OFFLOAD_CNT_PATH_UNSUPPORTED);
+AMT_OFFLOAD_CNT_ATTR(busy_fallback, AMT_OFFLOAD_CNT_BUSY_FALLBACK);
+AMT_OFFLOAD_CNT_ATTR(punt_dropped, AMT_OFFLOAD_CNT_PUNT_DROPPED);
 
 static struct attribute *amt_offload_attrs[] = {
 	&dev_attr_offload_mode.attr,

@@ -5521,6 +5692,9 @@ static struct attribute *amt_offload_attrs[] = {
 	&dev_attr_offload_resets.attr,
 	&dev_attr_offload_stale_completions.attr,
 	&dev_attr_offload_desync.attr,
+	&dev_attr_offload_path_unsupported.attr,
+	&dev_attr_offload_busy_fallback.attr,
+	&dev_attr_offload_punt_dropped.attr,
 	NULL,
 };
 

@@ -389,7 +389,9 @@ struct amt_offload_state;
 
 /* Per-relay offload counters, exported under offload/ in sysfs. */
 enum amt_offload_counter {
-	/* bind() succeeded and the caps covered the outer family */
+	/* bind() succeeded, and the caps covered the outer family and offered
+	 * a data path this relay may use
+	 */
 	AMT_OFFLOAD_CNT_BIND,
 	/* ... of which: bound when a provider registered on the stream_dev
 	 * this relay already uses

@@ -418,6 +420,15 @@ enum amt_offload_counter {
 	AMT_OFFLOAD_CNT_STALE,
 	/* desired state lost an update (allocation failure); software only */
 	AMT_OFFLOAD_CNT_DESYNC,
+	/* no data path this relay may use: the caps offered none, or TX
+	 * without ops->xmit (ingress counts only in required mode); it stayed
+	 * unbound
+	 */
+	AMT_OFFLOAD_CNT_PATH_UNSUPPORTED,
+	/* xmit returned BUSY: one synchronous software pass ran instead */
+	AMT_OFFLOAD_CNT_BUSY_FALLBACK,
+	/* ingress: a punted copy of a group the provider replicates, dropped */
+	AMT_OFFLOAD_CNT_PUNT_DROPPED,
 	AMT_OFFLOAD_CNT_MAX,
 };
 

@@ -42,8 +42,16 @@ struct sk_buff;
  *	group without a callback when a member leaves, until it publishes the
  *	recommit. Every other packet goes straight to software.
  * @AMT_OFFLOAD_PATH_INGRESS: the provider replicates on ingress and the
- *	host only sees punted packets. Semantics only until a real provider
- *	exists; no simulated provider implements it.
+ *	host only sees punted packets. From &amt_offload_ops.group_commit
+ *	until &amt_offload_ops.group_disable or &amt_offload_ops.flush the
+ *	provider replicates every packet of the group, and it must never also
+ *	punt one. amt drops, and counts, a punted copy of a group only while
+ *	the group's commit is published: from some time after group_commit
+ *	returns until amt withdraws it, which it does before any group_disable
+ *	or flush and, with no callback, once its desired state is lost. A
+ *	punted copy outside that window is forwarded in software as well.
+ *	Semantics only until a real provider exists; no simulated provider
+ *	implements it, and only a relay in required mode binds on it.
  */
 enum amt_offload_path {
 	AMT_OFFLOAD_PATH_TX,

@@ -203,7 +211,12 @@ struct amt_offload_stats {
  * @group_disable: stop forwarding @grp and free it. amt has already
  *	stopped submitting packets for @grp and waited out @xmit callers.
  * @xmit: TX path. Replicate @skb for @grp under @gen or decline; see
- *	&enum amt_offload_result. Softirq context, under rcu_read_lock().
+ *	&enum amt_offload_result. @skb is the relay data packet with its
+ *	data at the inner IP header. On %AMT_OFFLOAD_CONSUMED the provider
+ *	owns @skb and must have handled every copy under @gen alone: each one
+ *	handed to the stack, or dropped and counted (see
+ *	%AMT_OFFLOAD_CONSUMED). Otherwise it must not have sent any copy, nor
+ *	modified or freed @skb. Softirq context, under rcu_read_lock().
  * @flush: drop every endpoint and group (reset after a timeout). amt then
  *	replays from its own desired state under a new epoch, except when
  *	the op that overran was a @group_disable or @endpoint_del of amt

@@ -318,4 +331,23 @@ int amt_offload_register(struct net_device *lower,
  */
 void amt_offload_unregister(struct net_device *lower);
 
+/**
+ * amt_offload_encap_xmit - emit one gateway's copy of a relay data packet
+ * @b: the binding @skb was handed to the provider under.
+ * @key: the gateway endpoint, as amt passed it to endpoint_add.
+ * @skb: the packet given to &amt_offload_ops.xmit; not consumed.
+ *
+ * For simulated providers: encapsulates a copy of @skb as AMT Multicast
+ * Data to @key and transmits it on the relay's stream device, byte for
+ * byte as amt's own software fan-out would. Which endpoints get a copy
+ * is the provider's decision, made from its own tables. Call only from
+ * &amt_offload_ops.xmit.
+ *
+ * Return: 0 if the copy was handed to the stack, else a negative errno
+ * (counted in the relay's tx_dropped when the copy or route failed).
+ */
+int amt_offload_encap_xmit(struct amt_offload_binding *b,
+			   const struct amt_offload_endpoint_key *key,
+			   struct sk_buff *skb);
+
 #endif /* _NET_AMT_OFFLOAD_H_ */