ovpn: construct and process GRO batches

index cf07c3c6611a..572a65cd8977 100644
@@ -965,6 +965,14 @@ INDIRECT_CALLABLE_SCOPE int udp4_gro_complete(struct sk_buff *skb, int nhoff)
 		skb_shinfo(skb)->gso_type |= (SKB_GSO_FRAGLIST|SKB_GSO_UDP_L4);
 		skb_shinfo(skb)->gso_segs = NAPI_GRO_CB(skb)->count;
 
+		/* GRO validated every datagram, but a CHECKSUM_COMPLETE value
+		 * covers only the first one: keep the verdict instead, as for
+		 * a device which validated the checksums itself
+		 */
+		if (skb->ip_summed == CHECKSUM_COMPLETE) {
+			skb->ip_summed = CHECKSUM_UNNECESSARY;
+			skb->csum_level = 0;
+		}
 		__skb_incr_checksum_unnecessary(skb);
 
 		return 0;

index 7370bcb80332..0b7ced2b5394 100644
@@ -176,6 +176,14 @@ int udp6_gro_complete(struct sk_buff *skb, int nhoff)
 		skb_shinfo(skb)->gso_type |= (SKB_GSO_FRAGLIST|SKB_GSO_UDP_L4);
 		skb_shinfo(skb)->gso_segs = NAPI_GRO_CB(skb)->count;
 
+		/* GRO validated every datagram, but a CHECKSUM_COMPLETE value
+		 * covers only the first one: keep the verdict instead, as for
+		 * a device which validated the checksums itself
+		 */
+		if (skb->ip_summed == CHECKSUM_COMPLETE) {
+			skb->ip_summed = CHECKSUM_UNNECESSARY;
+			skb->csum_level = 0;
+		}
 		__skb_incr_checksum_unnecessary(skb);
 
 		return 0;

index 998906ec3b32..cb3fd34d56f6 100644
@@ -67,6 +67,10 @@ 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 it unmodified.
+	 */
+	UDP_FLAGS_ENCAP_ACCEPT_FRAGLIST,
 };
 
 /* per NUMA structure for lockless producer usage. */

@@ -211,6 +215,19 @@ static inline bool udp_encap_needed(void)
 	return false;
 }
 
+static inline bool udp_encap_accepts_fraglist(struct sock *sk,
+					      struct sk_buff *skb)
+{
+	/* pairs with set_mask_bits in setup_udp_tunnel_sock, so the callbacks
+	 * published before acceptance are visible once it is observed
+	 */
+	return skb_is_gso(skb) &&
+	       skb_shinfo(skb)->gso_type ==
+		       (SKB_GSO_UDP_L4 | SKB_GSO_FRAGLIST) &&
+	       test_bit_acquire(UDP_FLAGS_ENCAP_ACCEPT_FRAGLIST,
+				&udp_sk(sk)->udp_flags);
+}
+
 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 1b033855a20d..1cac10516262 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)
 {

123 lines not on this page · raw message

index a128fe85620d..c023344891a2 100644
@@ -71,6 +71,13 @@ 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);
+	/* a socket must stop accepting frag-lists before it is set up again */
+	WARN_ON_ONCE(udp_test_bit(ENCAP_ACCEPT_FRAGLIST, sk));
+
 	/* Disable multicast loopback */
 	inet_clear_bit(MC_LOOP, sk);
 

@@ -87,6 +94,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);

116 lines not on this page · raw message

index cb3fd34d56f6..3ae7ef796abb 100644
@@ -71,6 +71,7 @@ enum {
 	 * an aggregate, it must leave it unmodified.
 	 */
 	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 c023344891a2..426c5e4ac583 100644
@@ -71,12 +71,28 @@ 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);
 	/* a socket must stop accepting frag-lists before it is set up again */
 	WARN_ON_ONCE(udp_test_bit(ENCAP_ACCEPT_FRAGLIST, sk));
+	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 registry ignores detachable sockets, so a callback which
+		 * is already registered there would stay pinned
+		 */
+		WARN_ON_ONCE(udp_sk(sk)->gro_receive);
+		/* 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);

@@ -91,13 +107,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 cd209298012d..913897888bbb 100644
@@ -3023,6 +3023,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)

@@ -3035,6 +3038,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..e3a8aa83344f 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,252 @@
 #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;
+	int ret = 0, nhoff;
+	bool ip_flush;
+	u32 header;
+
+	/* FOU and GUE clear encap_mark and set is_fou before passing the inner
+	 * UDP packet to GRO, so frag-list completion would run on their outer
+	 * UDP header and skip their own completion
+	 */
+	if (unlikely(NAPI_GRO_CB(skb)->flush || NAPI_GRO_CB(skb)->is_fou ||
+		     skb->encapsulation || skb_is_gso(skb)))
+		goto flush;
+
+	if (!ovpn_udp_gro_header(skb, &header) ||
+	    FIELD_GET(OVPN_OPCODE_PKTTYPE_MASK, header) != OVPN_DATA_V2)
+		goto flush;
+
+	/* 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.
+		 * Read the held header without pulling which might change the
+		 * skb shape.
+		 */
+		if (NAPI_GRO_CB(p)->encap_mark || !NAPI_GRO_CB(p)->is_flist ||
+		    skb_headlen(p) < skb_gro_offset(p) + OVPN_OPCODE_SIZE ||
+		    get_unaligned_be32(p->data + skb_gro_offset(p)) != header)
+			return p;
+
+		/* GRO has already matched the outer addresses and UDP ports;
+		 * check the remaining outer IP fields
+		 */
+		ip_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 || ip_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;
+
+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.
+	 */
+	NAPI_GRO_CB(skb)->flush = 1;
+	list_for_each_entry(p, head, list)
+		if (NAPI_GRO_CB(p)->same_flow)
+			return p;
+	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, and clear
+		 * the state which the receive stack consumed on the head only
+		 */
+		skb->protocol = proto;
+		skb->skb_iif = skb_iif;
+		skb->priority = priority;
+#ifdef CONFIG_NETWORK_SECMARK
+		skb->secmark = secmark;
+#endif
+#ifdef CONFIG_NET_SCHED
+		skb->tc_index = tc_index;
+#endif
+		__vlan_hwaccel_clear_tag(skb);
+		skb_reset_redirect(skb);
+#ifdef CONFIG_NET_XGRESS
+		skb->tc_skip_classify = 0;
+#endif
+		skb_clear_delivery_time(skb);
+		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 +318,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);

@@ -89,14 +340,28 @@ static int ovpn_udp_encap_recv(struct sock *sk, struct sk_buff *skb)
 	 * header are accessible.
 	 * They are required to fetch the OP code, the key ID and the peer ID.
 	 */
-	if (unlikely(!pskb_may_pull(skb, sizeof(struct udphdr) +
-				    OVPN_OPCODE_SIZE))) {
+	if (skb_is_gso(skb)) {
+		/* Only an accepted frag-list aggregate can carry OpenVPN
+		 * records. It is checked without pulling, since one which we
+		 * decline must reach UDP unmodified so that it can segment it.
+		 */
+		if (unlikely(skb_shinfo(skb)->gso_type !=
+			     (SKB_GSO_UDP_L4 | SKB_GSO_FRAGLIST)))
+			goto drop;
+		aggregate = true;
+
+		if (unlikely(skb_headlen(skb) <
+			     sizeof(struct udphdr) + OVPN_OPCODE_SIZE))
+			goto drop;
+	} else if (unlikely(!pskb_may_pull(skb, sizeof(struct udphdr) +
+					   OVPN_OPCODE_SIZE))) {
 		net_dbg_ratelimited("%s: packet too small from UDP socket\n",
 				    netdev_name(ovpn->dev));
 		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 +371,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 +384,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 +792,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;

@@ -522,6 +805,18 @@ int ovpn_udp_socket_attach(struct ovpn_socket *ovpn_sock, struct socket *sock,
 	rcu_read_lock();
 	old_data = rcu_dereference_sk_user_data(ovpn_sock->sk);
 	if (!old_data) {
+		/* another encapsulation (such as ESP-in-UDP) may own the
+		 * socket without setting sk_user_data, and the GRO callback
+		 * which it may have registered cannot be replaced
+		 */
+		if (READ_ONCE(udp_sk(ovpn_sock->sk)->encap_type) ||
+		    READ_ONCE(udp_sk(ovpn_sock->sk)->gro_receive)) {
+			rcu_read_unlock();
+			netdev_dbg(ovpn->dev,
+				   "provided socket already taken by other user\n");
+			return -EBUSY;
+		}
+
 		/* socket is currently unused - we can take it */
 		rcu_read_unlock();
 		setup_udp_tunnel_sock(sock_net(ovpn_sock->sk), sock->sk, &cfg);

@@ -558,16 +853,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 e3a8aa83344f..2f77acc47bd9 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);

@@ -232,6 +249,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

@@ -241,9 +259,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, and clear
 		 * the state which the receive stack consumed on the head only
 		 */