ovpn: construct and process GRO batches

index 998906ec3b32..2a0c45ddd7b1 100644
@@ -67,6 +67,11 @@ enum {
 	UDP_FLAGS_ACCEPT_FRAGLIST,
 	UDP_FLAGS_ACCEPT_L4,
 	UDP_FLAGS_ENCAP_ENABLED, /* This socket enabled encap */
+	/* encap_rcv accepts UDP L4 frag-list GSO. If it does not consume such
+	 * an aggregate, it must leave packet contents unchanged. Layout-only
+	 * pulls are permitted.
+	 */
+	UDP_FLAGS_ENCAP_ACCEPT_FRAGLIST,
 };
 
 /* per NUMA structure for lockless producer usage. */

@@ -211,6 +216,13 @@ static inline bool udp_encap_needed(void)
 	return false;
 }
 
+static inline bool udp_encap_accepts_fraglist(struct sock *sk,
+					      struct sk_buff *skb)
+{
+	return skb_is_gso(skb) && udp_test_bit(ENCAP_ACCEPT_FRAGLIST, sk) &&
+	       skb_shinfo(skb)->gso_type == (SKB_GSO_UDP_L4 | SKB_GSO_FRAGLIST);
+}
+
 static inline bool udp_unexpected_gso(struct sock *sk, struct sk_buff *skb)
 {
 	if (!skb_is_gso(skb))

index 498b7b262fa9..a540e86d5f84 100644
@@ -85,6 +85,7 @@ struct udp_tunnel_sock_cfg {
 	void *sk_user_data;     /* user data used by encap_rcv call back */
 	/* Used for setting up udp_sock fields, see udp.h for details */
 	__u8  encap_type;
+	unsigned long udp_flags;
 	udp_tunnel_encap_rcv_t encap_rcv;
 	udp_tunnel_encap_err_lookup_t encap_err_lookup;
 	udp_tunnel_encap_err_rcv_t encap_err_rcv;

index ff828a3dce3c..069df5d96075 100644
@@ -266,6 +266,7 @@ int skb_gro_receive_list(struct sk_buff *p, struct sk_buff *skb)
 
 	return 0;
 }
+EXPORT_SYMBOL(skb_gro_receive_list);
 
 static void gro_complete(struct gro_node *gro, struct sk_buff *skb)
 {

index 303d0102519a..9b17ae823e56 100644
@@ -2391,7 +2391,8 @@ static int __udp_queue_rcv_skb(struct sock *sk, struct sk_buff *skb)
  * Note that in the success and error cases, the skb is assumed to
  * have either been requeued or freed.
  */
-static int udp_queue_rcv_one_skb(struct sock *sk, struct sk_buff *skb)
+static int udp_queue_rcv_one_skb(struct sock *sk, struct sk_buff *skb,
+				 bool encap_fraglist)
 {
 	enum skb_drop_reason drop_reason = SKB_DROP_REASON_NOT_SPECIFIED;
 	struct udp_sock *up = udp_sk(sk);

@@ -2424,14 +2425,25 @@ static int udp_queue_rcv_one_skb(struct sock *sk, struct sk_buff *skb)
 		/* if we're overly short, let UDP handle it */
 		encap_rcv = READ_ONCE(up->encap_rcv);
 		if (encap_rcv) {
+			unsigned int segs = encap_fraglist ?
+					    skb_shinfo(skb)->gso_segs : 1;
 			int ret;
 
-			/* Verify checksum before giving to encap */
-			if (udp_lib_checksum_complete(skb))
+			/* GRO validated every datagram before constructing an
+			 * accepted frag-list. Its aggregate checksum state
+			 * does not describe the complete skb, so validate only
+			 * after a fallback to individual datagrams.
+			 */
+			if (!encap_fraglist && udp_lib_checksum_complete(skb))
 				goto csum_error;
 
 			ret = encap_rcv(sk, skb);
-			if (ret <= 0) {
+			if (!ret) {
+				__SNMP_ADD_STATS(net->mib.udp_statistics,
+						 UDP_MIB_INDATAGRAMS, segs);
+				return 0;
+			}
+			if (ret < 0 && !encap_fraglist) {
 				__UDP_INC_STATS(net, UDP_MIB_INDATAGRAMS);
 				return -ret;
 			}

@@ -2440,6 +2452,13 @@ static int udp_queue_rcv_one_skb(struct sock *sk, struct sk_buff *skb)
 		/* FALLTHROUGH -- it's a UDP Packet */
 	}
 
+	/* The encapsulator did not consume an aggregate it had opted into so
+	 * we ask udp_queue_rcv_skb to retry after segmentation rather than
+	 * exposing multiple datagrams to the socket as one.
+	 */
+	if (encap_fraglist)
+		return -EAGAIN;
+
 	prefetch(&sk->sk_rmem_alloc);
 	if (rcu_access_pointer(sk->sk_filter) &&
 	    udp_lib_checksum_complete(skb))

@@ -2466,11 +2485,15 @@ static int udp_queue_rcv_one_skb(struct sock *sk, struct sk_buff *skb)
 
 static int udp_queue_rcv_skb(struct sock *sk, struct sk_buff *skb)
 {
+	bool encap_fraglist = udp_encap_accepts_fraglist(sk, skb);
 	struct sk_buff *next, *segs;
 	int ret;
 
-	if (likely(!udp_unexpected_gso(sk, skb)))
-		return udp_queue_rcv_one_skb(sk, skb);
+	if (likely(encap_fraglist || !udp_unexpected_gso(sk, skb))) {
+		ret = udp_queue_rcv_one_skb(sk, skb, encap_fraglist);
+		if (likely(ret != -EAGAIN))
+			return ret;
+	}
 
 	BUILD_BUG_ON(sizeof(struct udp_skb_cb) > SKB_GSO_CB_OFFSET);
 	__skb_push(skb, -skb_mac_offset(skb));

@@ -2479,7 +2502,7 @@ static int udp_queue_rcv_skb(struct sock *sk, struct sk_buff *skb)
 		__skb_pull(skb, skb_transport_offset(skb));
 
 		udp_post_segment_fix_csum(skb);
-		ret = udp_queue_rcv_one_skb(sk, skb);
+		ret = udp_queue_rcv_one_skb(sk, skb, false);
 		if (ret > 0)
 			ip_protocol_deliver_rcu(dev_net(skb->dev), skb, ret);
 	}

index a128fe85620d..411073068075 100644
@@ -71,6 +71,11 @@ static bool sk_saddr_any(struct sock *sk)
 void setup_udp_tunnel_sock(struct net *net, struct sock *sk,
 			   struct udp_tunnel_sock_cfg *cfg)
 {
+	const unsigned long allowed = BIT(UDP_FLAGS_ENCAP_ACCEPT_FRAGLIST);
+	unsigned long flags = cfg->udp_flags & allowed;
+
+	WARN_ON_ONCE(cfg->udp_flags & ~allowed);
+
 	/* Disable multicast loopback */
 	inet_clear_bit(MC_LOOP, sk);
 

@@ -87,6 +92,11 @@ void setup_udp_tunnel_sock(struct net *net, struct sock *sk,
 	udp_sk(sk)->gro_receive = cfg->gro_receive;
 	udp_sk(sk)->gro_complete = cfg->gro_complete;
 
+	/* Publish frag-list acceptance only after the callbacks which consume
+	 * that representation
+	 */
+	set_mask_bits(&udp_sk(sk)->udp_flags, allowed, flags);
+
 	udp_tunnel_encap_enable(sk);
 
 	udp_tunnel_update_gro_rcv(sk, true);

index 59f9872c3552..1b3d3adb3f44 100644
@@ -810,7 +810,8 @@ static int __udpv6_queue_rcv_skb(struct sock *sk, struct sk_buff *skb)
 	return 0;
 }
 
-static int udpv6_queue_rcv_one_skb(struct sock *sk, struct sk_buff *skb)
+static int udpv6_queue_rcv_one_skb(struct sock *sk, struct sk_buff *skb,
+				   bool encap_fraglist)
 {
 	enum skb_drop_reason drop_reason = SKB_DROP_REASON_NOT_SPECIFIED;
 	struct udp_sock *up = udp_sk(sk);

@@ -840,14 +841,25 @@ static int udpv6_queue_rcv_one_skb(struct sock *sk, struct sk_buff *skb)
 		/* if we're overly short, let UDP handle it */
 		encap_rcv = READ_ONCE(up->encap_rcv);
 		if (encap_rcv) {
+			unsigned int segs = encap_fraglist ?
+					    skb_shinfo(skb)->gso_segs : 1;
 			int ret;
 
-			/* Verify checksum before giving to encap */
-			if (udp_lib_checksum_complete(skb))
+			/* GRO validated every datagram before constructing an
+			 * accepted frag-list. Its aggregate checksum state
+			 * does not describe the complete skb, so validate only
+			 * after a fallback to individual datagrams.
+			 */
+			if (!encap_fraglist && udp_lib_checksum_complete(skb))
 				goto csum_error;
 
 			ret = encap_rcv(sk, skb);
-			if (ret <= 0) {
+			if (!ret) {
+				__SNMP_ADD_STATS(net->mib.udp_stats_in6,
+						 UDP_MIB_INDATAGRAMS, segs);
+				return 0;
+			}
+			if (ret < 0 && !encap_fraglist) {
 				__UDP6_INC_STATS(net, UDP_MIB_INDATAGRAMS);
 				return -ret;
 			}

@@ -856,6 +868,13 @@ static int udpv6_queue_rcv_one_skb(struct sock *sk, struct sk_buff *skb)
 		/* FALLTHROUGH -- it's a UDP Packet */
 	}
 
+	/* The encapsulator did not consume an aggregate it had opted into so
+	 * we ask udpv6_queue_rcv_skb to retry after segmentation rather than
+	 * exposing multiple datagrams to the socket as one.
+	 */
+	if (encap_fraglist)
+		return -EAGAIN;
+
 	prefetch(&sk->sk_rmem_alloc);
 	if (rcu_access_pointer(sk->sk_filter) &&
 	    udp_lib_checksum_complete(skb))

@@ -883,11 +902,15 @@ static int udpv6_queue_rcv_one_skb(struct sock *sk, struct sk_buff *skb)
 
 static int udpv6_queue_rcv_skb(struct sock *sk, struct sk_buff *skb)
 {
+	bool encap_fraglist = udp_encap_accepts_fraglist(sk, skb);
 	struct sk_buff *next, *segs;
 	int ret;
 
-	if (likely(!udp_unexpected_gso(sk, skb)))
-		return udpv6_queue_rcv_one_skb(sk, skb);
+	if (likely(encap_fraglist || !udp_unexpected_gso(sk, skb))) {
+		ret = udpv6_queue_rcv_one_skb(sk, skb, encap_fraglist);
+		if (likely(ret != -EAGAIN))
+			return ret;
+	}
 
 	__skb_push(skb, -skb_mac_offset(skb));
 	segs = udp_rcv_segment(sk, skb, false);

@@ -895,7 +918,7 @@ static int udpv6_queue_rcv_skb(struct sock *sk, struct sk_buff *skb)
 		__skb_pull(skb, skb_transport_offset(skb));
 
 		udp_post_segment_fix_csum(skb);
-		ret = udpv6_queue_rcv_one_skb(sk, skb);
+		ret = udpv6_queue_rcv_one_skb(sk, skb, false);
 		if (ret > 0)
 			ip6_protocol_deliver_rcu(dev_net(skb->dev), skb, ret,
 						 true);

index 2a0c45ddd7b1..e81ed7248ea0 100644
@@ -72,6 +72,7 @@ enum {
 	 * pulls are permitted.
 	 */
 	UDP_FLAGS_ENCAP_ACCEPT_FRAGLIST,
+	UDP_FLAGS_GRO_RCV_DETACHABLE, /* gro_receive may be detached */
 };
 
 /* per NUMA structure for lockless producer usage. */

index a540e86d5f84..3a137901c88b 100644
@@ -211,10 +211,12 @@ static inline int udp_tunnel_handle_offloads(struct sk_buff *skb, bool udp_csum)
 #if IS_ENABLED(CONFIG_NET_UDP_TUNNEL)
 void udp_tunnel_update_gro_lookup(struct net *net, struct sock *sk, bool add);
 void udp_tunnel_update_gro_rcv(struct sock *sk, bool add);
+void udp_tunnel_detach_gro(struct sock *sk);
 #else
 static inline void udp_tunnel_update_gro_lookup(struct net *net,
 						struct sock *sk, bool add) {}
 static inline void udp_tunnel_update_gro_rcv(struct sock *sk, bool add) {}
+static inline void udp_tunnel_detach_gro(struct sock *sk) {}
 #endif
 
 static inline void udp_tunnel_cleanup_gro(struct sock *sk)

125 lines not on this page · raw message

index 411073068075..f8fe5f673441 100644
@@ -71,10 +71,22 @@ static bool sk_saddr_any(struct sock *sk)
 void setup_udp_tunnel_sock(struct net *net, struct sock *sk,
 			   struct udp_tunnel_sock_cfg *cfg)
 {
-	const unsigned long allowed = BIT(UDP_FLAGS_ENCAP_ACCEPT_FRAGLIST);
+	const unsigned long allowed = BIT(UDP_FLAGS_GRO_RCV_DETACHABLE) |
+				      BIT(UDP_FLAGS_ENCAP_ACCEPT_FRAGLIST);
 	unsigned long flags = cfg->udp_flags & allowed;
 
 	WARN_ON_ONCE(cfg->udp_flags & ~allowed);
+	set_mask_bits(&udp_sk(sk)->udp_flags, BIT(UDP_FLAGS_GRO_RCV_DETACHABLE),
+		      flags & BIT(UDP_FLAGS_GRO_RCV_DETACHABLE));
+	if (flags & BIT(UDP_FLAGS_GRO_RCV_DETACHABLE)) {
+		WARN_ON_ONCE(cfg->gro_complete ||
+			     !(flags & BIT(UDP_FLAGS_ENCAP_ACCEPT_FRAGLIST)));
+		/* The socket may already be receiving packets so drain readers
+		 * which could have observed the old dispatch mode before
+		 * publishing the detachable callback.
+		 */
+		synchronize_net();
+	}
 
 	/* Disable multicast loopback */
 	inet_clear_bit(MC_LOOP, sk);

@@ -89,13 +101,18 @@ void setup_udp_tunnel_sock(struct net *net, struct sock *sk,
 	udp_sk(sk)->encap_err_rcv = cfg->encap_err_rcv;
 	udp_sk(sk)->encap_err_lookup = cfg->encap_err_lookup;
 	udp_sk(sk)->encap_destroy = cfg->encap_destroy;
-	udp_sk(sk)->gro_receive = cfg->gro_receive;
-	udp_sk(sk)->gro_complete = cfg->gro_complete;
+	WRITE_ONCE(udp_sk(sk)->gro_complete, cfg->gro_complete);
+	/* Publish receive after the completion callback which its aggregates
+	 * may later require
+	 */
+	WRITE_ONCE(udp_sk(sk)->gro_receive, cfg->gro_receive);
 
 	/* Publish frag-list acceptance only after the callbacks which consume
 	 * that representation
 	 */
-	set_mask_bits(&udp_sk(sk)->udp_flags, allowed, flags);
+	set_mask_bits(&udp_sk(sk)->udp_flags,
+		      BIT(UDP_FLAGS_ENCAP_ACCEPT_FRAGLIST),
+		      flags & BIT(UDP_FLAGS_ENCAP_ACCEPT_FRAGLIST));
 
 	udp_tunnel_encap_enable(sk);
 

index 9b17ae823e56..db0b60ae6859 100644
@@ -2999,6 +2999,9 @@ int udp_lib_setsockopt(struct sock *sk, int level, int optname,
 		case 0:
 #ifdef CONFIG_XFRM
 		case UDP_ENCAP_ESPINUDP:
+			if (udp_test_bit(GRO_RCV_DETACHABLE, sk))
+				udp_tunnel_detach_gro(sk);
+
 			set_xfrm_gro_udp_encap_rcv(val, sk->sk_family, sk);
 #if IS_ENABLED(CONFIG_IPV6)
 			if (sk->sk_family == AF_INET6)

@@ -3011,6 +3014,9 @@ int udp_lib_setsockopt(struct sock *sk, int level, int optname,
 #endif
 			fallthrough;
 		case UDP_ENCAP_L2TPINUDP:
+			if (udp_test_bit(GRO_RCV_DETACHABLE, sk))
+				udp_tunnel_detach_gro(sk);
+
 			WRITE_ONCE(up->encap_type, val);
 			udp_tunnel_encap_enable(sk);
 			break;

index 055cdb1bee13..98f55f178a95 100644
@@ -8,11 +8,14 @@
 
 #include <linux/netdevice.h>
 #include <linux/inetdevice.h>
+#include <linux/if_vlan.h>
 #include <linux/skbuff.h>
 #include <linux/socket.h>
 #include <linux/udp.h>
+#include <linux/unaligned.h>
 #include <net/addrconf.h>
 #include <net/dst_cache.h>
+#include <net/gro.h>
 #include <net/route.h>
 #include <net/transp_v6.h>
 #include <net/udp.h>

@@ -27,6 +30,235 @@
 #include "socket.h"
 #include "udp.h"
 
+/* like UDP and TCP frag-list GRO */
+#define OVPN_UDP_GRO_CNT_MAX 64
+
+static bool ovpn_udp_gro_header(struct sk_buff *skb, u32 *header)
+{
+	const unsigned int offset = skb_gro_offset(skb);
+
+	/* GRO replaces its frag0 pointer after holding an skb, so keep the
+	 * openvpn header linear for later candidate comparisons
+	 */
+	if (!pskb_may_pull(skb, offset + OVPN_OPCODE_SIZE))
+		return false;
+
+	*header = get_unaligned_be32(skb->data + offset);
+	return true;
+}
+
+static struct sk_buff *ovpn_udp_gro_receive_fraglist(struct sock *sk,
+						     struct list_head *head,
+						     struct sk_buff *skb)
+{
+	const unsigned int gso_size = skb_gro_len(skb);
+	struct sk_buff *p, *pp = NULL;
+	u32 header, header2;
+	int ret = 0, nhoff;
+	bool flush;
+
+	/* A packet which cannot join an aggregate must not overtake records of
+	 * the same flow which GRO is still holding. Complete those first, as
+	 * the other GRO handlers do.
+	 */
+	if (NAPI_GRO_CB(skb)->flush || skb->encapsulation ||
+	    !ovpn_udp_gro_header(skb, &header) ||
+	    FIELD_GET(OVPN_OPCODE_PKTTYPE_MASK, header) != OVPN_DATA_V2 ||
+	    skb_is_gso(skb)) {
+		NAPI_GRO_CB(skb)->flush = 1;
+		list_for_each_entry(p, head, list)
+			if (NAPI_GRO_CB(p)->same_flow)
+				return p;
+		return NULL;
+	}
+
+	/* udp_gro_receive sets encap_mark before invoking a tunnel callback,
+	 * normally because the callback recurses into an inner protocol. ovpn
+	 * retains complete outer UDP records instead, so clear it and use
+	 * ordinary UDP frag-list completion.
+	 */
+	NAPI_GRO_CB(skb)->encap_mark = 0;
+	NAPI_GRO_CB(skb)->is_flist = 1;
+
+	list_for_each_entry(p, head, list) {
+		if (!NAPI_GRO_CB(p)->same_flow)
+			continue;
+
+		/* a different representation, opcode, key or peer ID completes
+		 * the held aggregate; this record can start the next one
+		 */
+		if (NAPI_GRO_CB(p)->encap_mark || !NAPI_GRO_CB(p)->is_flist ||
+		    !ovpn_udp_gro_header(p, &header2) || header != header2)
+			return p;
+
+		/* GRO has already matched the outer addresses and UDP ports;
+		 * check the remaining outer IP fields
+		 */
+		flush = gro_receive_network_flush(udp_hdr(skb), udp_hdr(p), p);
+
+		/* The first record determines the nominal GSO size. A shorter
+		 * final record may follow it, but a larger record cannot.
+		 * Checksum metadata must also be uniform because the aggregate
+		 * exposes only one checksum state.
+		 */
+		if (gso_size > skb_shinfo(p)->gso_size || flush ||
+		    skb->ip_summed != p->ip_summed ||
+		    skb->csum_level != p->csum_level) {
+			pp = p;
+		} else {
+			/* skb_gro_receive_list pulls the headers already
+			 * processed by GRO before linking this skb to the
+			 * record list so we have to manually preserve the
+			 * outer network header location for later handling
+			 */
+			nhoff = NAPI_GRO_CB(skb)->network_offset;
+			skb_set_network_header(skb, nhoff);
+			ret = skb_gro_receive_list(p, skb);
+		}
+
+		/* complete the aggregate if the append failed, or after
+		 * appending a shorter final record, or after reaching the
+		 * record-count limit
+		 */
+		if (ret || gso_size != skb_shinfo(p)->gso_size ||
+		    NAPI_GRO_CB(p)->count >= OVPN_UDP_GRO_CNT_MAX)
+			pp = p;
+
+		return pp;
+	}
+
+	return NULL;
+}
+
+static bool ovpn_udp_outer_match(const struct sk_buff *head,
+				 const struct sk_buff *frag)
+{
+	const struct udphdr *uhf = udp_hdr(frag);
+	const struct udphdr *uhh = udp_hdr(head);
+
+	/* NAT might have rewritten endpoints after GRO */
+	if (unlikely(uhh->source != uhf->source || uhh->dest != uhf->dest))
+		return false;
+
+	switch (head->protocol) {
+	case htons(ETH_P_IP):
+		return ip_hdr(head)->saddr == ip_hdr(frag)->saddr &&
+		       ip_hdr(head)->daddr == ip_hdr(frag)->daddr;
+	case htons(ETH_P_IPV6):
+		return ipv6_addr_equal(&ipv6_hdr(head)->saddr,
+				       &ipv6_hdr(frag)->saddr) &&
+		       ipv6_addr_equal(&ipv6_hdr(head)->daddr,
+				       &ipv6_hdr(frag)->daddr);
+	default:
+		return false;
+	}
+}
+
+/* validate an skb and turn it into a list of uniform OpenVPN records or return
+ * NULL without modifying an aggregate which UDP must segment
+ */
+static struct sk_buff *ovpn_udp_gso_prepare(struct sk_buff *skb, u32 header)
+{
+	unsigned int len = 0, truesize = 0;
+	struct sk_buff *list, *record;
+
+	if (!skb_is_gso(skb)) {
+		skb_mark_not_on_list(skb);
+		goto out;
+	}
+
+	list = skb_shinfo(skb)->frag_list;
+	if (unlikely(!list))
+		return NULL;
+
+	/* if any of the skbs is not private, ask the stack to segment */
+	if (unlikely(skb_shared(skb) || skb_cloned(skb)))
+		return NULL;
+
+	/* transformations after GRO only update the aggregate head: if its
+	 * outer tuple changed, ask UDP to propagate it while segmenting
+	 */
+	if (unlikely(skb_pagelen(skb) != sizeof(struct udphdr) +
+					 skb_shinfo(skb)->gso_size ||
+		     !ovpn_udp_outer_match(skb, list)))
+		return NULL;
+
+	skb_walk_frags(skb, record) {
+		/* if the OpenVPN header is not uniform (batch created by
+		 * generic GRO by coalescing unrelated packets) ask the stack
+		 * to segment and handle each segment independently
+		 */
+		if (unlikely(skb_shared(record) || skb_cloned(record) ||
+			     skb_headlen(record) < OVPN_OPCODE_SIZE ||
+			     get_unaligned_be32(record->data) != header))
+			return NULL;
+
+		len += record->len;
+		truesize += record->truesize;
+	}
+
+	if (unlikely(skb->data_len < len || skb->len < len ||
+		     skb->truesize < truesize))
+		return NULL;
+
+	/* detach the exclusively owned fraglist */
+	skb_shinfo(skb)->frag_list = NULL;
+	skb->data_len -= len;
+	skb->truesize -= truesize;
+	skb->len -= len;
+	skb->next = list;
+	skb_gso_reset(skb);
+
+out:
+	/* skb_gro_receive_list has already pulled the UDP headers from child
+	 * records, while the aggregate head still starts at UDP, so pull its
+	 * header too before returning the list
+	 */
+	__skb_pull(skb, sizeof(struct udphdr));
+	return skb;
+}
+
+static void ovpn_udp_recv(struct ovpn_peer *peer, struct sk_buff *list)
+{
+	const __be16 proto = list->protocol;
+	const u32 priority = list->priority;
+#ifdef CONFIG_NET_SCHED
+	const u16 tc_index = list->tc_index;
+#endif
+#ifdef CONFIG_NETWORK_SECMARK
+	const u32 secmark = list->secmark;
+#endif
+	const int skb_iif = list->skb_iif;
+	struct sk_buff *skb, *next;
+
+	skb_list_walk_safe(list, skb, next) {
+		/* propagate post-GRO metadata from the list head (its outer
+		 * VLAN tag was already consumed by the receive stack)
+		 */
+		skb->protocol = proto;
+		skb->skb_iif = skb_iif;
+		skb->priority = priority;
+		__vlan_hwaccel_clear_tag(skb);
+#ifdef CONFIG_NETWORK_SECMARK
+		skb->secmark = secmark;
+#endif
+#ifdef CONFIG_NET_SCHED
+		skb->tc_index = tc_index;
+#endif
+		skb_mark_not_on_list(skb);
+
+		/* keep the current reference alive for the next record before
+		 * handing this one to crypto
+		 */
+		if (next && unlikely(!ovpn_peer_hold(peer))) {
+			DEBUG_NET_WARN_ON_ONCE(1);
+			kfree_skb_list(next);
+			next = NULL;
+		}
+		ovpn_recv(peer, skb);
+	}
+}
+
 /* Retrieve the corresponding ovpn object from a UDP socket
  * rcu_read_lock must be held on entry
  */

@@ -69,7 +301,9 @@ static int ovpn_udp_encap_recv(struct sock *sk, struct sk_buff *skb)
 	struct ovpn_socket *ovpn_sock;
 	struct ovpn_priv *ovpn;
 	struct ovpn_peer *peer;
-	u32 peer_id;
+	bool aggregate = false;
+	struct sk_buff *list;
+	u32 header, peer_id;
 	u8 opcode;
 
 	ovpn_sock = ovpn_socket_from_udp_sock(sk);

@@ -85,6 +319,16 @@ static int ovpn_udp_encap_recv(struct sock *sk, struct sk_buff *skb)
 		goto drop_noovpn;
 	}
 
+	/* only an accepted frag-list aggregate can carry OpenVPN records, and
+	 * UDP can segment it if we hand it back
+	 */
+	if (skb_is_gso(skb)) {
+		if (unlikely(skb_shinfo(skb)->gso_type !=
+			     (SKB_GSO_UDP_L4 | SKB_GSO_FRAGLIST)))
+			goto drop;
+		aggregate = true;
+	}
+
 	/* Make sure the first 4 bytes of the skb data buffer after the UDP
 	 * header are accessible.
 	 * They are required to fetch the OP code, the key ID and the peer ID.

@@ -96,7 +340,8 @@ static int ovpn_udp_encap_recv(struct sock *sk, struct sk_buff *skb)
 		goto drop;
 	}
 
-	opcode = ovpn_opcode_from_skb(skb, sizeof(struct udphdr));
+	header = get_unaligned_be32(skb->data + sizeof(struct udphdr));
+	opcode = FIELD_GET(OVPN_OPCODE_PKTTYPE_MASK, header);
 	if (unlikely(opcode != OVPN_DATA_V2)) {
 		/* DATA_V1 is not supported */
 		if (opcode == OVPN_DATA_V1)

@@ -106,7 +351,7 @@ static int ovpn_udp_encap_recv(struct sock *sk, struct sk_buff *skb)
 		return 1;
 	}
 
-	peer_id = ovpn_peer_id_from_skb(skb, sizeof(struct udphdr));
+	peer_id = FIELD_GET(OVPN_OPCODE_PEERID_MASK, header);
 	/* some OpenVPN server implementations send data packets with the
 	 * peer-id set to UNDEF. In this case we skip the peer lookup by peer-id
 	 * and we try with the transport address

@@ -119,12 +364,27 @@ static int ovpn_udp_encap_recv(struct sock *sk, struct sk_buff *skb)
 	if (unlikely(!peer))
 		goto drop;
 
-	/* pop off outer UDP header */
-	__skb_pull(skb, sizeof(struct udphdr));
-	ovpn_recv(peer, skb);
+	/* A generic aggregate may contain records for different consumers and
+	 * post-GRO transformations may have changed only its head. Prepare
+	 * only a private, unmodified aggregate with uniform opcode, key ID and
+	 * peer ID; otherwise return it to UDP for segmentation.
+	 */
+	list = ovpn_udp_gso_prepare(skb, header);
+	if (unlikely(!list)) {
+		ovpn_peer_put(peer);
+		return 1;
+	}
+
+	ovpn_udp_recv(peer, list);
 	return 0;
 
 drop:
+	/* let UDP segment an accepted aggregate so that every datagram gets
+	 * its own verdict
+	 */
+	if (aggregate)
+		return 1;
+
 	ovpn_dev_dstats_rx_dropped(ovpn->dev);
 drop_noovpn:
 	kfree_skb(skb);

@@ -512,8 +772,11 @@ int ovpn_udp_socket_attach(struct ovpn_socket *ovpn_sock, struct socket *sock,
 	struct udp_tunnel_sock_cfg cfg = {
 		.sk_user_data = ovpn_sock,
 		.encap_type = UDP_ENCAP_OVPNINUDP,
+		.udp_flags = BIT(UDP_FLAGS_GRO_RCV_DETACHABLE) |
+			     BIT(UDP_FLAGS_ENCAP_ACCEPT_FRAGLIST),
 		.encap_rcv = ovpn_udp_encap_recv,
 		.encap_destroy = ovpn_udp_encap_destroy,
+		.gro_receive = ovpn_udp_gro_receive_fraglist,
 	};
 	struct ovpn_socket *old_data;
 	int ret;

@@ -558,16 +821,35 @@ int ovpn_udp_socket_attach(struct ovpn_socket *ovpn_sock, struct socket *sock,
  */
 void ovpn_udp_socket_detach(struct ovpn_socket *ovpn_sock)
 {
+	struct udp_sock *up = udp_sk(ovpn_sock->sk);
 	struct sock *sk = ovpn_sock->sk;
 
+	/* Stop accepting aggregates while encap_rcv is still available. A
+	 * reader which sampled acceptance before it was cleared can either be
+	 * consumed by ovpn or fall back to UDP segmentation.
+	 */
+	if (udp_test_bit(GRO_RCV_DETACHABLE, sk) &&
+	    READ_ONCE(up->gro_receive) == ovpn_udp_gro_receive_fraglist)
+		udp_tunnel_detach_gro(sk);
+
+	/* UDP_ENCAP may have replaced encap_rcv while leaving other ovpn state
+	 * attached. Clear only callbacks still owned by ovpn. The final RCU
+	 * grace period in ovpn_socket_release drains readers which captured
+	 * encap_rcv before this point.
+	 */
+	if (READ_ONCE(up->encap_rcv) == ovpn_udp_encap_recv)
+		WRITE_ONCE(up->encap_rcv, NULL);
+	if (READ_ONCE(up->encap_destroy) == ovpn_udp_encap_destroy)
+		WRITE_ONCE(up->encap_destroy, NULL);
+
 	/* Re-enable multicast loopback */
 	inet_set_bit(MC_LOOP, sk);
 	/* Disable CHECKSUM_UNNECESSARY to CHECKSUM_COMPLETE conversion */
 	inet_dec_convert_csum(sk);
 
-	WRITE_ONCE(udp_sk(sk)->encap_type, 0);
-	WRITE_ONCE(udp_sk(sk)->encap_rcv, NULL);
-	WRITE_ONCE(udp_sk(sk)->encap_destroy, NULL);
+	/* UDP_ENCAP might have rewritten the encap_type */
+	if (READ_ONCE(up->encap_type) == UDP_ENCAP_OVPNINUDP)
+		WRITE_ONCE(up->encap_type, 0);
 
 	rcu_assign_sk_user_data(sk, NULL);
 }

index 98f55f178a95..5e5e48c02e13 100644
@@ -6,9 +6,11 @@
  *  Author:	Antonio Quartulli <[email protected]>
  */
 
+#include <linux/cache.h>
 #include <linux/netdevice.h>
 #include <linux/inetdevice.h>
 #include <linux/if_vlan.h>
+#include <linux/prefetch.h>
 #include <linux/skbuff.h>
 #include <linux/socket.h>
 #include <linux/udp.h>

@@ -33,6 +35,21 @@
 /* like UDP and TCP frag-list GRO */
 #define OVPN_UDP_GRO_CNT_MAX 64
 
+/* leave enough work between a cache hint and the record which consumes it */
+#define OVPN_UDP_GRO_PREFETCH_DISTANCE 2
+
+static void ovpn_udp_prefetchw_linear(const struct sk_buff *skb)
+{
+	unsigned long addr = (unsigned long)skb->data;
+	unsigned long end = addr + skb_headlen(skb);
+
+	/* crypto overwrites data in place, so request write ownership of each
+	 * linear cache line before the AEAD implementation reaches it
+	 */
+	for (; addr < end; addr = ALIGN(addr + 1, L1_CACHE_BYTES))
+		prefetchw((void *)addr);
+}
+
 static bool ovpn_udp_gro_header(struct sk_buff *skb, u32 *header)
 {
 	const unsigned int offset = skb_gro_offset(skb);

@@ -220,6 +237,7 @@ static struct sk_buff *ovpn_udp_gso_prepare(struct sk_buff *skb, u32 header)
 
 static void ovpn_udp_recv(struct ovpn_peer *peer, struct sk_buff *list)
 {
+	struct sk_buff *skb, *next, *prefetch;
 	const __be16 proto = list->protocol;
 	const u32 priority = list->priority;
 #ifdef CONFIG_NET_SCHED

@@ -229,9 +247,23 @@ static void ovpn_udp_recv(struct ovpn_peer *peer, struct sk_buff *list)
 	const u32 secmark = list->secmark;
 #endif
 	const int skb_iif = list->skb_iif;
-	struct sk_buff *skb, *next;
+	int i;
+
+	/* a fraglist GRO aggregate makes later ciphertext visible before the
+	 * current record is decrypted, so we prime the first two records, then
+	 * keep the cache hints the same distance ahead while draining the list
+	 */
+	prefetch = list->next ? list : NULL;
+	for (i = 0; i < OVPN_UDP_GRO_PREFETCH_DISTANCE && prefetch; i++) {
+		ovpn_udp_prefetchw_linear(prefetch);
+		prefetch = prefetch->next;
+	}
 
 	skb_list_walk_safe(list, skb, next) {
+		if (prefetch) {
+			ovpn_udp_prefetchw_linear(prefetch);
+			prefetch = prefetch->next;
+		}
 		/* propagate post-GRO metadata from the list head (its outer
 		 * VLAN tag was already consumed by the receive stack)
 		 */