This is a very early RFC to discuss a different approch to solve the data transfer stall reported by Geliang and Gang. There are a few open points documented into the individual patches, the goal here is describe with some detail the intended architecture. Note that the diffstat is biases by the quite large patch 2/6, which contains mechanical transformation of existing code; "real" changes are noticiable smaller. Paolo Abeni (6): mptcp: move checks vs rcvbuf size earlier in the RX path mptcp: sync mptcp skb cb layout with tcp one tcp: expose the tcp_collapse_ofo_queue() helper to mptcp usage, too mptcp: implemented OoO queue pruning mptcp: refine coalescing conditions mptcp: unclone skbs before coalescing them, when needed include/net/tcp.h | 4 ++ net/ipv4/tcp_input.c | 55 ++++++++++------ net/mptcp/mib.c | 3 + net/mptcp/mib.h | 3 + net/mptcp/options.c | 35 +++++++++- net/mptcp/protocol.c | 154 +++++++++++++++++++++++++++++++------------ net/mptcp/protocol.h | 7 +- 7 files changed, 195 insertions(+), 66 deletions(-) -- 2.53.0 Currently the enforcement of the rcvbuf constraint is implemented when moving the skbs into the msk receive or OoO queue. Under significant memory pressure the above can cause permanent data transfer stalls. Move the checks early on, before landing even in the subflow queues. Signed-off-by: Paolo Abeni <pabeni@redhat.com> --- Note that: - this needs the follow-up patches to really fix the stall - the memory comparison is intentionally very rough, as the msk socket lock is not currently held where the condition is now enforced. This should require some refinement, shared as-is to avoid more latency on my side --- net/mptcp/options.c | 21 +++++++++++++++++++-- net/mptcp/protocol.c | 9 ++------- 2 files changed, 21 insertions(+), 9 deletions(-) diff --git a/net/mptcp/options.c b/net/mptcp/options.c index XXXXXXX..XXXXXXX 100644 --- a/net/mptcp/options.c +++ b/net/mptcp/options.c @@ -XXX,XX +XXX,XX @@ static bool add_addr_hmac_valid(struct mptcp_sock *msk, return hmac == mp_opt->ahmac; } -/* Return false in case of error (or subflow has been reset), - * else return true. +static bool mptcp_over_limit(const struct sock *sk, struct sk_buff *skb) +{ + int limit; + + if (!skb->len) + return false; + + limit = READ_ONCE(sk->sk_rcvbuf) << 1; + return sk_rmem_alloc_get(sk) > limit; +} + +/* Return false when the caller must to drop the packet, i.e. in case of error, + * subflow has been reset, or over memory limits. */ bool mptcp_incoming_options(struct sock *sk, struct sk_buff *skb) { @@ -XXX,XX +XXX,XX @@ bool mptcp_incoming_options(struct sock *sk, struct sk_buff *skb) __mptcp_data_acked(subflow->conn); mptcp_data_unlock(subflow->conn); + + if (mptcp_over_limit(subflow->conn, skb)) + return false; return true; } @@ -XXX,XX +XXX,XX @@ bool mptcp_incoming_options(struct sock *sk, struct sk_buff *skb) return true; } + if (mptcp_over_limit(subflow->conn, skb)) + return false; + mpext = skb_ext_add(skb, SKB_EXT_MPTCP); if (!mpext) return false; diff --git a/net/mptcp/protocol.c b/net/mptcp/protocol.c index XXXXXXX..XXXXXXX 100644 --- a/net/mptcp/protocol.c +++ b/net/mptcp/protocol.c @@ -XXX,XX +XXX,XX @@ static bool __mptcp_move_skbs_from_subflow(struct mptcp_sock *msk, mptcp_init_skb(ssk, skb, offset, len); - if (own_msk && sk_rmem_alloc_get(sk) < sk->sk_rcvbuf) { + if (own_msk) { mptcp_subflow_lend_fwdmem(subflow, skb); ret |= __mptcp_move_skb(sk, skb); } else { @@ -XXX,XX +XXX,XX @@ static bool __mptcp_move_skbs(struct sock *sk, struct list_head *skbs, u32 *delt *delta = 0; while (1) { - /* If the msk recvbuf is full stop, don't drop */ - if (sk_rmem_alloc_get(sk) > sk->sk_rcvbuf) - break; - prefetch(skb->next); list_del(&skb->list); *delta += skb->truesize; @@ -XXX,XX +XXX,XX @@ static bool mptcp_can_spool_backlog(struct sock *sk, struct list_head *skbs) mem_cgroup_from_sk(sk)); /* Don't spool the backlog if the rcvbuf is full. */ - if (list_empty(&msk->backlog_list) || - sk_rmem_alloc_get(sk) > sk->sk_rcvbuf) + if (list_empty(&msk->backlog_list)) return false; INIT_LIST_HEAD(skbs); -- 2.53.0 The MPTCP protocol uses a significantly different CB layout WRT TCP, as it includes different information and use 64 bits for the sequence numbers. As the msk-level rcvbuf buffer size is limited by the core socket code the INT_MAX, we can safely use 32 bits for MPTCP-level sequence number. This allow updating the MPTCP CB layout so that fields with a corresponding TCP-level data use the same area inside the CB itself. Add build time check the unsure the latter invariant. Signed-off-by: Paolo Abeni <pabeni@redhat.com> --- net/mptcp/protocol.c | 81 +++++++++++++++++++++++++------------------- net/mptcp/protocol.h | 5 +-- 2 files changed, 50 insertions(+), 36 deletions(-) diff --git a/net/mptcp/protocol.c b/net/mptcp/protocol.c index XXXXXXX..XXXXXXX 100644 --- a/net/mptcp/protocol.c +++ b/net/mptcp/protocol.c @@ -XXX,XX +XXX,XX @@ #include "protocol.h" #include "mib.h" -static unsigned int mptcp_inq_hint(const struct sock *sk); +static int mptcp_inq_hint(const struct sock *sk); #define CREATE_TRACE_POINTS #include <trace/events/mptcp.h> @@ -XXX,XX +XXX,XX @@ static bool __mptcp_try_coalesce(struct sock *sk, struct sk_buff *to, !skb_try_coalesce(to, from, fragstolen, delta)) return false; - pr_debug("colesced seq %llx into %llx new len %d new end seq %llx\n", + pr_debug("colesced seq %x into %x new len %d new end seq %x\n", MPTCP_SKB_CB(from)->map_seq, MPTCP_SKB_CB(to)->map_seq, to->len, MPTCP_SKB_CB(from)->end_seq); MPTCP_SKB_CB(to)->end_seq = MPTCP_SKB_CB(from)->end_seq; @@ -XXX,XX +XXX,XX @@ static void mptcp_data_queue_ofo(struct mptcp_sock *msk, struct sk_buff *skb) { struct sock *sk = (struct sock *)msk; struct rb_node **p, *parent; - u64 seq, end_seq, max_seq; + u32 seq, end_seq, max_seq; struct sk_buff *skb1; seq = MPTCP_SKB_CB(skb)->map_seq; end_seq = MPTCP_SKB_CB(skb)->end_seq; max_seq = atomic64_read(&msk->rcv_wnd_sent); - pr_debug("msk=%p seq=%llx limit=%llx empty=%d\n", msk, seq, max_seq, + pr_debug("msk=%p seq=%x limit=%x empty=%d\n", msk, seq, max_seq, RB_EMPTY_ROOT(&msk->out_of_order_queue)); - if (after64(end_seq, max_seq)) { + if (after(end_seq, max_seq)) { /* out of window */ mptcp_drop(sk, skb); - pr_debug("oow by %lld, rcv_wnd_sent %llu\n", - (unsigned long long)end_seq - (unsigned long)max_seq, + pr_debug("oow by %d, rcv_wnd_sent %llu\n", + end_seq - max_seq, (unsigned long long)atomic64_read(&msk->rcv_wnd_sent)); MPTCP_INC_STATS(sock_net(sk), MPTCP_MIB_NODSSWINDOW); return; @@ -XXX,XX +XXX,XX @@ static void mptcp_data_queue_ofo(struct mptcp_sock *msk, struct sk_buff *skb) } /* Can avoid an rbtree lookup if we are adding skb after ooo_last_skb */ - if (!before64(seq, MPTCP_SKB_CB(msk->ooo_last_skb)->end_seq)) { + if (!before(seq, MPTCP_SKB_CB(msk->ooo_last_skb)->end_seq)) { MPTCP_INC_STATS(sock_net(sk), MPTCP_MIB_OFOQUEUETAIL); parent = &msk->ooo_last_skb->rbnode; p = &parent->rb_right; @@ -XXX,XX +XXX,XX @@ static void mptcp_data_queue_ofo(struct mptcp_sock *msk, struct sk_buff *skb) while (*p) { parent = *p; skb1 = rb_to_skb(parent); - if (before64(seq, MPTCP_SKB_CB(skb1)->map_seq)) { + if (before(seq, MPTCP_SKB_CB(skb1)->map_seq)) { p = &parent->rb_left; continue; } - if (before64(seq, MPTCP_SKB_CB(skb1)->end_seq)) { - if (!after64(end_seq, MPTCP_SKB_CB(skb1)->end_seq)) { + if (before(seq, MPTCP_SKB_CB(skb1)->end_seq)) { + if (!after(end_seq, MPTCP_SKB_CB(skb1)->end_seq)) { /* All the bits are present. Drop. */ mptcp_drop(sk, skb); MPTCP_INC_STATS(sock_net(sk), MPTCP_MIB_DUPDATA); return; } - if (after64(seq, MPTCP_SKB_CB(skb1)->map_seq)) { + if (after(seq, MPTCP_SKB_CB(skb1)->map_seq)) { /* partial overlap: * | skb | * | skb1 | @@ -XXX,XX +XXX,XX @@ static void mptcp_data_queue_ofo(struct mptcp_sock *msk, struct sk_buff *skb) merge_right: /* Remove other segments covered by skb. */ while ((skb1 = skb_rb_next(skb)) != NULL) { - if (before64(end_seq, MPTCP_SKB_CB(skb1)->end_seq)) + if (before(end_seq, MPTCP_SKB_CB(skb1)->end_seq)) break; rb_erase(&skb1->rbnode, &msk->out_of_order_queue); mptcp_drop(sk, skb1); @@ -XXX,XX +XXX,XX @@ static void mptcp_init_skb(struct sock *ssk, struct sk_buff *skb, int offset, /* the skb map_seq accounts for the skb offset: * mptcp_subflow_get_mapped_dsn() is based on the current tp->copied_seq - * value + * value; note that seq numbers are truncated to 32bits */ MPTCP_SKB_CB(skb)->map_seq = mptcp_subflow_get_mapped_dsn(subflow); MPTCP_SKB_CB(skb)->end_seq = MPTCP_SKB_CB(skb)->map_seq + copy_len; MPTCP_SKB_CB(skb)->offset = offset; + MPTCP_SKB_CB(skb)->flags = 0; MPTCP_SKB_CB(skb)->has_rxtstamp = has_rxtstamp; MPTCP_SKB_CB(skb)->cant_coalesce = 0; @@ -XXX,XX +XXX,XX @@ static void mptcp_init_skb(struct sock *ssk, struct sk_buff *skb, int offset, static bool __mptcp_move_skb(struct sock *sk, struct sk_buff *skb) { - u64 copy_len = MPTCP_SKB_CB(skb)->end_seq - MPTCP_SKB_CB(skb)->map_seq; + u32 copy_len = MPTCP_SKB_CB(skb)->end_seq - MPTCP_SKB_CB(skb)->map_seq; struct mptcp_sock *msk = mptcp_sk(sk); + u32 ack_seq = msk->ack_seq; struct sk_buff *tail; mptcp_borrow_fwdmem(sk, skb); - if (MPTCP_SKB_CB(skb)->map_seq == msk->ack_seq) { + if (MPTCP_SKB_CB(skb)->map_seq == ack_seq) { /* in sequence */ msk->bytes_received += copy_len; WRITE_ONCE(msk->ack_seq, msk->ack_seq + copy_len); @@ -XXX,XX +XXX,XX @@ static bool __mptcp_move_skb(struct sock *sk, struct sk_buff *skb) skb_set_owner_r(skb, sk); __skb_queue_tail(&sk->sk_receive_queue, skb); return true; - } else if (after64(MPTCP_SKB_CB(skb)->map_seq, msk->ack_seq)) { + } else if (after(MPTCP_SKB_CB(skb)->map_seq, ack_seq)) { mptcp_data_queue_ofo(msk, skb); return false; } @@ -XXX,XX +XXX,XX @@ static bool __mptcp_move_skbs_from_subflow(struct mptcp_sock *msk, static bool __mptcp_ofo_queue(struct mptcp_sock *msk) { + u32 seq_delta, ack_seq = msk->ack_seq; struct sock *sk = (struct sock *)msk; struct sk_buff *skb, *tail; bool moved = false; struct rb_node *p; - u64 end_seq; p = rb_first(&msk->out_of_order_queue); pr_debug("msk=%p empty=%d\n", msk, RB_EMPTY_ROOT(&msk->out_of_order_queue)); while (p) { skb = rb_to_skb(p); - if (after64(MPTCP_SKB_CB(skb)->map_seq, msk->ack_seq)) + if (after(MPTCP_SKB_CB(skb)->map_seq, ack_seq)) break; p = rb_next(p); rb_erase(&skb->rbnode, &msk->out_of_order_queue); - if (unlikely(!after64(MPTCP_SKB_CB(skb)->end_seq, - msk->ack_seq))) { + if (unlikely(!after(MPTCP_SKB_CB(skb)->end_seq, ack_seq))) { mptcp_drop(sk, skb); MPTCP_INC_STATS(sock_net(sk), MPTCP_MIB_DUPDATA); continue; } - end_seq = MPTCP_SKB_CB(skb)->end_seq; + seq_delta = MPTCP_SKB_CB(skb)->end_seq - ack_seq; tail = skb_peek_tail(&sk->sk_receive_queue); if (!tail || !mptcp_ooo_try_coalesce(msk, tail, skb)) { - int delta = msk->ack_seq - MPTCP_SKB_CB(skb)->map_seq; + int delta = ack_seq - MPTCP_SKB_CB(skb)->map_seq; /* skip overlapping data, if any */ - pr_debug("uncoalesced seq=%llx ack seq=%llx delta=%d\n", - MPTCP_SKB_CB(skb)->map_seq, msk->ack_seq, - delta); + pr_debug("uncoalesced seq=%x ack seq=%x delta=%d\n", + MPTCP_SKB_CB(skb)->map_seq, ack_seq, delta); MPTCP_SKB_CB(skb)->offset += delta; MPTCP_SKB_CB(skb)->map_seq += delta; __skb_queue_tail(&sk->sk_receive_queue, skb); } - msk->bytes_received += end_seq - msk->ack_seq; - WRITE_ONCE(msk->ack_seq, end_seq); + msk->bytes_received += seq_delta; + WRITE_ONCE(msk->ack_seq, msk->ack_seq + seq_delta); moved = true; } return moved; @@ -XXX,XX +XXX,XX @@ static bool mptcp_move_skbs(struct sock *sk) return enqueued; } -static unsigned int mptcp_inq_hint(const struct sock *sk) +static int mptcp_inq_hint(const struct sock *sk) { const struct mptcp_sock *msk = mptcp_sk(sk); const struct sk_buff *skb; skb = skb_peek(&sk->sk_receive_queue); if (skb) { - u64 hint_val = READ_ONCE(msk->ack_seq) - MPTCP_SKB_CB(skb)->map_seq; + int hint_val = (u32)READ_ONCE(msk->ack_seq) - + MPTCP_SKB_CB(skb)->map_seq; - if (hint_val >= INT_MAX) - return INT_MAX; + if (hint_val < 0) + return -hint_val; - return (unsigned int)hint_val; + return hint_val; } if (sk->sk_state == TCP_CLOSE || (sk->sk_shutdown & RCV_SHUTDOWN)) @@ -XXX,XX +XXX,XX @@ static int mptcp_recvmsg(struct sock *sk, struct msghdr *msg, size_t len, tcp_recv_timestamp(msg, sk, &tss); if (cmsg_flags & MPTCP_CMSG_INQ) { - unsigned int inq = mptcp_inq_hint(sk); + int inq = mptcp_inq_hint(sk); put_cmsg(msg, SOL_TCP, TCP_CM_INQ, sizeof(inq), &inq); } @@ -XXX,XX +XXX,XX @@ static int mptcp_napi_poll(struct napi_struct *napi, int budget) return work_done; } +#define CHK_CB_FIELD(mptcp_field, tcp_field) \ + ({ \ + BUILD_BUG_ON(offsetof(struct mptcp_skb_cb, mptcp_field) != \ + offsetof(struct tcp_skb_cb, tcp_field)); \ + BUILD_BUG_ON(offsetofend(struct mptcp_skb_cb, mptcp_field) != \ + offsetofend(struct tcp_skb_cb, tcp_field)); \ + }) + void __init mptcp_proto_init(void) { struct mptcp_delegated_action *delegated; int cpu; + CHK_CB_FIELD(map_seq, seq); + CHK_CB_FIELD(end_seq, end_seq); + CHK_CB_FIELD(flags, tcp_flags); + mptcp_prot.h.hashinfo = tcp_prot.h.hashinfo; if (percpu_counter_init(&mptcp_sockets_allocated, 0, GFP_KERNEL)) diff --git a/net/mptcp/protocol.h b/net/mptcp/protocol.h index XXXXXXX..XXXXXXX 100644 --- a/net/mptcp/protocol.h +++ b/net/mptcp/protocol.h @@ -XXX,XX +XXX,XX @@ #define MPTCP_SYNC_SNDBUF 7 struct mptcp_skb_cb { - u64 map_seq; - u64 end_seq; + u32 map_seq; + u32 end_seq; u32 offset; + u16 flags; u8 has_rxtstamp; u8 cant_coalesce; }; -- 2.53.0 The end goal is to avoid duplicating the quite untrivial strategy at MPTCP level. After the previous patch, the mentioned helpers could process skbs standing in MPTCP-level queues without any CB-related adaptation. The only additional adjustment needed is explicitly providing the OoO queue reference, to cope with different sk layout. Additionally rename the helper to clearly document its hybrid nature and let it return the number of collapsed skbs, to allow proper accounting from the future MPTCP caller. Signed-off-by: Paolo Abeni <pabeni@redhat.com> --- Note: - this will need a significant amount of testing at the TCP level and explicit approval from Eric, which I can't guess if we can hope. --- include/net/tcp.h | 4 ++++ net/ipv4/tcp_input.c | 55 ++++++++++++++++++++++++++++---------------- 2 files changed, 39 insertions(+), 20 deletions(-) diff --git a/include/net/tcp.h b/include/net/tcp.h index XXXXXXX..XXXXXXX 100644 --- a/include/net/tcp.h +++ b/include/net/tcp.h @@ -XXX,XX +XXX,XX @@ extern void tcp_openreq_init_rwin(struct request_sock *req, void tcp_enter_memory_pressure(struct sock *sk); void tcp_leave_memory_pressure(struct sock *sk); +unsigned int xtcp_collapse_ofo_queue(struct sock *sk, + struct rb_root *out_of_order_queue, + struct sk_buff **ooo_last_skb, + u8 scaling_radio); static inline int keepalive_intvl_when(const struct tcp_sock *tp) { diff --git a/net/ipv4/tcp_input.c b/net/ipv4/tcp_input.c index XXXXXXX..XXXXXXX 100644 --- a/net/ipv4/tcp_input.c +++ b/net/ipv4/tcp_input.c @@ -XXX,XX +XXX,XX @@ static struct sk_buff *tcp_collapse_one(struct sock *sk, struct sk_buff *skb, /* Collapse contiguous sequence of skbs head..tail with * sequence numbers start..end. * + * sk can be either a TCP or an MPTCP socket. + * * If tail is NULL, this means until the end of the queue. * * Segments with FIN/SYN are not collapsed (only because this * simplifies code) + * + * Returns the number of collapsed skbs. */ -static void -tcp_collapse(struct sock *sk, struct sk_buff_head *list, struct rb_root *root, - struct sk_buff *head, struct sk_buff *tail, u32 start, u32 end) +static unsigned int +xtcp_collapse(struct sock *sk, struct sk_buff_head *list, struct rb_root *root, + struct sk_buff *head, struct sk_buff *tail, u32 start, u32 end, + u8 scaling_ratio) { struct sk_buff *skb = head, *n; + unsigned int collapsed = 0; struct sk_buff_head tmp; bool end_of_skbs; @@ -XXX,XX +XXX,XX @@ tcp_collapse(struct sock *sk, struct sk_buff_head *list, struct rb_root *root, /* No new bits? It is possible on ofo queue. */ if (!before(start, TCP_SKB_CB(skb)->end_seq)) { + collapsed++; skb = tcp_collapse_one(sk, skb, list, root); if (!skb) break; @@ -XXX,XX +XXX,XX @@ tcp_collapse(struct sock *sk, struct sk_buff_head *list, struct rb_root *root, * overlaps to the next one and mptcp allow collapsing. */ if (!(TCP_SKB_CB(skb)->tcp_flags & (TCPHDR_SYN | TCPHDR_FIN)) && - (tcp_win_from_space(sk, skb->truesize) > skb->len || + (__tcp_win_from_space(scaling_ratio, skb->truesize) > skb->len || before(TCP_SKB_CB(skb)->seq, start))) { end_of_skbs = false; break; @@ -XXX,XX +XXX,XX @@ tcp_collapse(struct sock *sk, struct sk_buff_head *list, struct rb_root *root, if (end_of_skbs || (TCP_SKB_CB(skb)->tcp_flags & (TCPHDR_SYN | TCPHDR_FIN)) || !skb_frags_readable(skb)) - return; + return collapsed; __skb_queue_head_init(&tmp); @@ -XXX,XX +XXX,XX @@ tcp_collapse(struct sock *sk, struct sk_buff_head *list, struct rb_root *root, start += size; } if (!before(start, TCP_SKB_CB(skb)->end_seq)) { + collapsed++; skb = tcp_collapse_one(sk, skb, list, root); if (!skb || skb == tail || @@ -XXX,XX +XXX,XX @@ tcp_collapse(struct sock *sk, struct sk_buff_head *list, struct rb_root *root, end: skb_queue_walk_safe(&tmp, skb, n) tcp_rbtree_insert(root, skb); + return collapsed; } /* Collapse ofo queue. Algorithm: select contiguous sequence of skbs - * and tcp_collapse() them until all the queue is collapsed. + * and xtcp_collapse() them until all the queue is collapsed. */ -static void tcp_collapse_ofo_queue(struct sock *sk) +unsigned int xtcp_collapse_ofo_queue(struct sock *sk, + struct rb_root *ooo_queue, + struct sk_buff **ooo_last_skb, + u8 scaling_ratio) { - struct tcp_sock *tp = tcp_sk(sk); - u32 range_truesize, sum_tiny = 0; + u32 range_truesize, sum_tiny = 0, collapsed = 0; struct sk_buff *skb, *head; u32 start, end; - skb = skb_rb_first(&tp->out_of_order_queue); + skb = skb_rb_first(ooo_queue); new_range: if (!skb) { - tp->ooo_last_skb = skb_rb_last(&tp->out_of_order_queue); - return; + *ooo_last_skb = skb_rb_last(ooo_queue); + return collapsed; } start = TCP_SKB_CB(skb)->seq; end = TCP_SKB_CB(skb)->end_seq; @@ -XXX,XX +XXX,XX @@ static void tcp_collapse_ofo_queue(struct sock *sk) /* Do not attempt collapsing tiny skbs */ if (range_truesize != head->truesize || end - start >= SKB_WITH_OVERHEAD(PAGE_SIZE)) { - tcp_collapse(sk, NULL, &tp->out_of_order_queue, - head, skb, start, end); + collapsed += xtcp_collapse(sk, NULL, ooo_queue, + head, skb, start, end, + scaling_ratio); } else { sum_tiny += range_truesize; if (sum_tiny > sk->sk_rcvbuf >> 3) - return; + return collapsed; } goto new_range; } @@ -XXX,XX +XXX,XX @@ static void tcp_collapse_ofo_queue(struct sock *sk) if (after(TCP_SKB_CB(skb)->end_seq, end)) end = TCP_SKB_CB(skb)->end_seq; } + return collapsed; } /* @@ -XXX,XX +XXX,XX @@ static int tcp_prune_queue(struct sock *sk, const struct sk_buff *in_skb) if (tcp_can_ingest(sk, in_skb)) return 0; - tcp_collapse_ofo_queue(sk); + xtcp_collapse_ofo_queue(sk, &tp->out_of_order_queue, + &tp->ooo_last_skb, tp->scaling_ratio); if (!skb_queue_empty(&sk->sk_receive_queue)) - tcp_collapse(sk, &sk->sk_receive_queue, NULL, - skb_peek(&sk->sk_receive_queue), - NULL, - tp->copied_seq, tp->rcv_nxt); + xtcp_collapse(sk, &sk->sk_receive_queue, NULL, + skb_peek(&sk->sk_receive_queue), + NULL, + tp->copied_seq, tp->rcv_nxt, + tp->scaling_ratio); if (tcp_can_ingest(sk, in_skb)) return 0; -- 2.53.0 Leverage the hybrid helper to implement the OoO queue prune at ingress time. If the msk is owned by the user-space at incoming skb time, perform the pruning in the release_cb. The prune check is additionally performed when the skb reaches the msk-level queues. Signed-off-by: Paolo Abeni <pabeni@redhat.com> --- Notes: - Similarly to path 'mptcp: move checks vs rcvbuf size earlier in the RX path', some cleanup/tuning in mptcp_over_limit() will be needed - Pruning in the release_cb() is likely not needed, should probably be removed (after more testing). --- net/mptcp/mib.c | 3 +++ net/mptcp/mib.h | 3 +++ net/mptcp/options.c | 22 +++++++++++++--- net/mptcp/protocol.c | 61 ++++++++++++++++++++++++++++++++++++++++++++ net/mptcp/protocol.h | 2 ++ 5 files changed, 87 insertions(+), 4 deletions(-) diff --git a/net/mptcp/mib.c b/net/mptcp/mib.c index XXXXXXX..XXXXXXX 100644 --- a/net/mptcp/mib.c +++ b/net/mptcp/mib.c @@ -XXX,XX +XXX,XX @@ static const struct snmp_mib mptcp_snmp_list[] = { SNMP_MIB_ITEM("SimultConnectFallback", MPTCP_MIB_SIMULTCONNFALLBACK), SNMP_MIB_ITEM("FallbackFailed", MPTCP_MIB_FALLBACKFAILED), SNMP_MIB_ITEM("WinProbe", MPTCP_MIB_WINPROBE), + SNMP_MIB_ITEM("OfoPruned", MPTCP_MIB_OFO_PRUNED), + SNMP_MIB_ITEM("RcvPruned", MPTCP_MIB_RCVPRUNED), + SNMP_MIB_ITEM("RcvCollapsed", MPTCP_MIB_RCVCOLLAPSED), }; /* mptcp_mib_alloc - allocate percpu mib counters diff --git a/net/mptcp/mib.h b/net/mptcp/mib.h index XXXXXXX..XXXXXXX 100644 --- a/net/mptcp/mib.h +++ b/net/mptcp/mib.h @@ -XXX,XX +XXX,XX @@ enum linux_mptcp_mib_field { MPTCP_MIB_SIMULTCONNFALLBACK, /* Simultaneous connect */ MPTCP_MIB_FALLBACKFAILED, /* Can't fallback due to msk status */ MPTCP_MIB_WINPROBE, /* MPTCP-level zero window probe */ + MPTCP_MIB_OFO_PRUNED, /* MPTCP-level OoO queue pruned */ + MPTCP_MIB_RCVPRUNED, /* Dropped due to memory constrains */ + MPTCP_MIB_RCVCOLLAPSED, /* Collapsed due to memory pressure */ __MPTCP_MIB_MAX }; diff --git a/net/mptcp/options.c b/net/mptcp/options.c index XXXXXXX..XXXXXXX 100644 --- a/net/mptcp/options.c +++ b/net/mptcp/options.c @@ -XXX,XX +XXX,XX @@ static bool add_addr_hmac_valid(struct mptcp_sock *msk, return hmac == mp_opt->ahmac; } -static bool mptcp_over_limit(const struct sock *sk, struct sk_buff *skb) +static bool mptcp_over_limit(struct sock *sk, struct sk_buff *skb, u32 seq) { + struct mptcp_sock *msk = mptcp_sk(sk); + bool ret = true; int limit; if (!skb->len) return false; + /* Allow some slack for backlog processing */ limit = READ_ONCE(sk->sk_rcvbuf) << 1; - return sk_rmem_alloc_get(sk) > limit; + if (sk_rmem_alloc_get(sk) < limit) + return false; + + mptcp_data_lock(sk); + if (!sock_owned_by_user(sk)) { + __mptcp_check_prune(sk, seq); + ret = sk_rmem_alloc_get(sk) > READ_ONCE(sk->sk_rcvbuf); + } else { + __set_bit(MPTCP_PRUNE, &msk->cb_flags); + } + mptcp_data_unlock(sk); + return ret; } /* Return false when the caller must to drop the packet, i.e. in case of error, @@ -XXX,XX +XXX,XX @@ bool mptcp_incoming_options(struct sock *sk, struct sk_buff *skb) __mptcp_data_acked(subflow->conn); mptcp_data_unlock(subflow->conn); - if (mptcp_over_limit(subflow->conn, skb)) + if (mptcp_over_limit(subflow->conn, skb, msk->ack_seq)) return false; return true; } @@ -XXX,XX +XXX,XX @@ bool mptcp_incoming_options(struct sock *sk, struct sk_buff *skb) return true; } - if (mptcp_over_limit(subflow->conn, skb)) + if (mptcp_over_limit(subflow->conn, skb, mp_opt.use_map ? mp_opt.data_seq : msk->ack_seq)) return false; mpext = skb_ext_add(skb, SKB_EXT_MPTCP); diff --git a/net/mptcp/protocol.c b/net/mptcp/protocol.c index XXXXXXX..XXXXXXX 100644 --- a/net/mptcp/protocol.c +++ b/net/mptcp/protocol.c @@ -XXX,XX +XXX,XX @@ static void mptcp_init_skb(struct sock *ssk, struct sk_buff *skb, int offset, skb_dst_drop(skb); } +/* "Inspiered" from the TCP version */ +static void mptcp_prune_ofo_queue(struct sock *sk, u32 seq) +{ + struct mptcp_sock *msk = mptcp_sk(sk); + struct rb_node *node, *prev; + bool pruned = false; + + if (RB_EMPTY_ROOT(&msk->out_of_order_queue)) + return; + + node = &msk->ooo_last_skb->rbnode; + + do { + struct sk_buff *skb = rb_to_skb(node); + + /* If incoming skb would land last in ofo queue, stop pruning. */ + if (after(seq, MPTCP_SKB_CB(skb)->map_seq)) + break; + + pruned = true; + prev = rb_prev(node); + rb_erase(node, &msk->out_of_order_queue); + mptcp_drop(sk, skb); + msk->ooo_last_skb = rb_to_skb(prev); + if (atomic_read(&sk->sk_rmem_alloc) < sk->sk_rcvbuf) + break; + + node = prev; + } while (node); + + if (pruned) + NET_INC_STATS(sock_net(sk), LINUX_MIB_OFOPRUNED); +} + +bool __mptcp_check_prune(struct sock *sk, u32 seq) +{ + struct mptcp_sock *msk = mptcp_sk(sk); + unsigned int dropped; + + if (likely(atomic_read(&sk->sk_rmem_alloc) < sk->sk_rcvbuf)) + return false; + + dropped = xtcp_collapse_ofo_queue(sk, &msk->out_of_order_queue, + &msk->ooo_last_skb, msk->scaling_ratio); + if (dropped) + MPTCP_ADD_STATS(sock_net(sk), MPTCP_MIB_RCVCOLLAPSED, dropped); + if (likely(atomic_read(&sk->sk_rmem_alloc) < sk->sk_rcvbuf)) + return false; + + mptcp_prune_ofo_queue(sk, seq); + return atomic_read(&sk->sk_rmem_alloc) >= sk->sk_rcvbuf; +} + static bool __mptcp_move_skb(struct sock *sk, struct sk_buff *skb) { u32 copy_len = MPTCP_SKB_CB(skb)->end_seq - MPTCP_SKB_CB(skb)->map_seq; @@ -XXX,XX +XXX,XX @@ static bool __mptcp_move_skb(struct sock *sk, struct sk_buff *skb) mptcp_borrow_fwdmem(sk, skb); + if (__mptcp_check_prune(sk, MPTCP_SKB_CB(skb)->map_seq)) { + MPTCP_INC_STATS(sock_net(sk), MPTCP_MIB_RCVPRUNED); + mptcp_drop(sk, skb); + return false; + } + if (MPTCP_SKB_CB(skb)->map_seq == ack_seq) { /* in sequence */ msk->bytes_received += copy_len; @@ -XXX,XX +XXX,XX @@ static void mptcp_release_cb(struct sock *sk) __mptcp_error_report(sk); if (__test_and_clear_bit(MPTCP_SYNC_SNDBUF, &msk->cb_flags)) __mptcp_sync_sndbuf(sk); + if (__test_and_clear_bit(MPTCP_PRUNE, &msk->cb_flags)) + __mptcp_check_prune(sk, msk->ack_seq - 1); } } diff --git a/net/mptcp/protocol.h b/net/mptcp/protocol.h index XXXXXXX..XXXXXXX 100644 --- a/net/mptcp/protocol.h +++ b/net/mptcp/protocol.h @@ -XXX,XX +XXX,XX @@ #define MPTCP_FLUSH_JOIN_LIST 5 #define MPTCP_SYNC_STATE 6 #define MPTCP_SYNC_SNDBUF 7 +#define MPTCP_PRUNE 8 struct mptcp_skb_cb { u32 map_seq; @@ -XXX,XX +XXX,XX @@ bool __mptcp_close(struct sock *sk, long timeout); void mptcp_cancel_work(struct sock *sk); void __mptcp_unaccepted_force_close(struct sock *sk); void mptcp_set_state(struct sock *sk, int state); +bool __mptcp_check_prune(struct sock *sk, u32 seq); bool mptcp_addresses_equal(const struct mptcp_addr_info *a, const struct mptcp_addr_info *b, bool use_port); -- 2.53.0 The current conditions prevent any coalescing when the receive buffer is small. Ensure that MPTCP can always aggregate at least at max GSO size. Signed-off-by: Paolo Abeni <pabeni@redhat.com> --- Note: - or we can drop entirely the rcvbuf-related check, to be verified vs simult_flows tests --- net/mptcp/protocol.c | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/net/mptcp/protocol.c b/net/mptcp/protocol.c index XXXXXXX..XXXXXXX 100644 --- a/net/mptcp/protocol.c +++ b/net/mptcp/protocol.c @@ -XXX,XX +XXX,XX @@ static bool __mptcp_try_coalesce(struct sock *sk, struct sk_buff *to, if (unlikely(MPTCP_SKB_CB(to)->cant_coalesce) || MPTCP_SKB_CB(from)->offset || - ((to->len + from->len) > (limit >> 3)) || + ((to->len + from->len) > max(U16_MAX, (limit >> 3))) || !skb_try_coalesce(to, from, fragstolen, delta)) return false; -- 2.53.0 The self-test can trigger skb coalescing on clones skb, as the forward path uses only veth devices. That in turn prevents coalescing making memory pressure scenario more extreme. Signed-off-by: Paolo Abeni <pabeni@redhat.com> --- Possibly we could obtain the same effect with some netem magic, would be better. --- net/mptcp/protocol.c | 1 + 1 file changed, 1 insertion(+) diff --git a/net/mptcp/protocol.c b/net/mptcp/protocol.c index XXXXXXX..XXXXXXX 100644 --- a/net/mptcp/protocol.c +++ b/net/mptcp/protocol.c @@ -XXX,XX +XXX,XX @@ static bool __mptcp_try_coalesce(struct sock *sk, struct sk_buff *to, if (unlikely(MPTCP_SKB_CB(to)->cant_coalesce) || MPTCP_SKB_CB(from)->offset || ((to->len + from->len) > max(U16_MAX, (limit >> 3))) || + (skb_cloned(to) && skb_unclone(to, GFP_ATOMIC)) || !skb_try_coalesce(to, from, fragstolen, delta)) return false; -- 2.53.0 This brings (hopefully) the final bits required to address the data transfer stall reported by Geliang and Gang This series improves mptcp retranmission to make them reliable: pruning can require some of them. The only change over the previous iteration is in patch 2, addressing sashiko feedback over the previous iteration (possible infinite rtx loop). Paolo Abeni (2): mptcp: move the retrans loop to a separate helper mptcp: let the retrans scheduler do its job. net/mptcp/protocol.c | 155 +++++++++++++++++++++++++++++++------------ 1 file changed, 111 insertions(+), 44 deletions(-) -- 2.54.0 This is a cleanup in order to make the next patch simpler. No functional change intended. Tested-by: Gang Yan <yangang@kylinos.cn> Tested-by: Geliang Tang <geliang@kernel.org> Acked-by: Geliang Tang <geliang@kernel.org> Signed-off-by: Paolo Abeni <pabeni@redhat.com> --- net/mptcp/protocol.c | 74 +++++++++++++++++++++++++------------------- 1 file changed, 43 insertions(+), 31 deletions(-) diff --git a/net/mptcp/protocol.c b/net/mptcp/protocol.c index XXXXXXX..XXXXXXX 100644 --- a/net/mptcp/protocol.c +++ b/net/mptcp/protocol.c @@ -XXX,XX +XXX,XX @@ static void mptcp_check_fastclose(struct mptcp_sock *msk) sk_error_report(sk); } -static void __mptcp_retrans(struct sock *sk) +/* Retransmit the specified data fragment on all the selected subflows. */ +static int __mptcp_push_retrans(struct sock *sk, struct mptcp_data_frag *dfrag) { struct mptcp_sendmsg_info info = { .data_lock_held = true, }; struct mptcp_sock *msk = mptcp_sk(sk); struct mptcp_subflow_context *subflow; - struct mptcp_data_frag *dfrag; struct sock *ssk; - int ret, err; - u16 len = 0; - - mptcp_clean_una_wakeup(sk); - - /* first check ssk: need to kick "stale" logic */ - err = mptcp_sched_get_retrans(msk); - dfrag = mptcp_rtx_head(sk); - if (!dfrag) { - if (mptcp_data_fin_enabled(msk)) { - struct inet_connection_sock *icsk = inet_csk(sk); - - WRITE_ONCE(icsk->icsk_retransmits, - icsk->icsk_retransmits + 1); - mptcp_set_datafin_timeout(sk); - mptcp_send_ack(msk); - - goto reset_timer; - } - - if (!mptcp_send_head(sk)) - goto clear_scheduled; - - goto reset_timer; - } - - if (err) - goto reset_timer; + int ret, len = 0; mptcp_for_each_subflow(msk, subflow) { if (READ_ONCE(subflow->scheduled)) { @@ -XXX,XX +XXX,XX @@ static void __mptcp_retrans(struct sock *sk) !msk->allow_subflows) { spin_unlock_bh(&msk->fallback_lock); release_sock(ssk); - goto clear_scheduled; + return -1; } while (info.sent < info.limit) { @@ -XXX,XX +XXX,XX @@ static void __mptcp_retrans(struct sock *sk) release_sock(ssk); } } + return len; +} + +static void __mptcp_retrans(struct sock *sk) +{ + struct mptcp_sock *msk = mptcp_sk(sk); + struct mptcp_subflow_context *subflow; + struct mptcp_data_frag *dfrag; + int err, len; + + mptcp_clean_una_wakeup(sk); + + /* first check ssk: need to kick "stale" logic */ + err = mptcp_sched_get_retrans(msk); + dfrag = mptcp_rtx_head(sk); + if (!dfrag) { + if (mptcp_data_fin_enabled(msk)) { + struct inet_connection_sock *icsk = inet_csk(sk); + + WRITE_ONCE(icsk->icsk_retransmits, + icsk->icsk_retransmits + 1); + mptcp_set_datafin_timeout(sk); + mptcp_send_ack(msk); + + goto reset_timer; + } + + if (!mptcp_send_head(sk)) + goto clear_scheduled; + + goto reset_timer; + } + + if (err) + goto reset_timer; + + len = __mptcp_push_retrans(sk, dfrag); + if (len < 0) + goto clear_scheduled; msk->bytes_retrans += len; dfrag->already_sent = max(dfrag->already_sent, len); -- 2.54.0 Currently the MPTCP core enforces that when MPTCP-level retrans timer fires, at most a single dfrag is retransmitted. If some corner-cases it may be necessary retransmit multiple dfrags, and the MPTCP socket will need to wait multiple retrans timeout to accomplish that. Remove the mentioned constraint, allowing to transmit multiple dfrags per retrans period, as long as the scheduler keeps selecting subflows for retransmissions and pending data is available in the rtx queue. The default scheduler will transmit a dfrag per available subflow. Tested-by: Gang Yan <yangang@kylinos.cn> Tested-by: Geliang Tang <geliang@kernel.org> Acked-by: Geliang Tang <geliang@kernel.org> Signed-off-by: Paolo Abeni <pabeni@redhat.com> --- v11 -> v12: - avoid infinite loop (sashiko) v10 -> v11: - avoid WARNING when retransmitting dfrag with 0 already_sent, as such status can happend, as reported by Geliang. Instead explicitly check for the bad condition and skip. v9 -> v10: - simpler handling for data-fin rtx v7 -> v8 - fix corner-case retrans_seq update v4 -> v5: - fixed already_sent update v3 -> v4: - avoid quadratic behavior, fix retrans_seq update - fix rtx timer re-schedule miss v2 -> v3: - fix infinite loop issue (should address tls tests failures) v1 -> v2: - fix retrans sequence update (sashiko) Note: - sashiko may see missing data-fin rtx when the initial `dfrag` is not NULL. data-fin RTX is NOT needed in such scenario. --- net/mptcp/protocol.c | 119 +++++++++++++++++++++++++++++++------------ 1 file changed, 87 insertions(+), 32 deletions(-) diff --git a/net/mptcp/protocol.c b/net/mptcp/protocol.c index XXXXXXX..XXXXXXX 100644 --- a/net/mptcp/protocol.c +++ b/net/mptcp/protocol.c @@ -XXX,XX +XXX,XX @@ static void __mptcp_clean_una_wakeup(struct sock *sk) mptcp_write_space(sk); } -static void mptcp_clean_una_wakeup(struct sock *sk) -{ - mptcp_data_lock(sk); - __mptcp_clean_una_wakeup(sk); - mptcp_data_unlock(sk); -} - static void mptcp_enter_memory_pressure(struct sock *sk) { struct mptcp_subflow_context *subflow; @@ -XXX,XX +XXX,XX @@ static void mptcp_check_fastclose(struct mptcp_sock *msk) sk_error_report(sk); } -/* Retransmit the specified data fragment on all the selected subflows. */ -static int __mptcp_push_retrans(struct sock *sk, struct mptcp_data_frag *dfrag) +/* + * Retransmit the specified data fragment on all the selected subflows, + * starting from the specified sequence + */ +static int __mptcp_push_retrans(struct sock *sk, struct mptcp_data_frag *dfrag, + u64 sent_seq) { struct mptcp_sendmsg_info info = { .data_lock_held = true, }; struct mptcp_sock *msk = mptcp_sk(sk); @@ -XXX,XX +XXX,XX @@ static int __mptcp_push_retrans(struct sock *sk, struct mptcp_data_frag *dfrag) mptcp_for_each_subflow(msk, subflow) { if (READ_ONCE(subflow->scheduled)) { + u16 offset = sent_seq - dfrag->data_seq; u16 copied = 0; mptcp_subflow_set_scheduled(subflow, false); @@ -XXX,XX +XXX,XX @@ static int __mptcp_push_retrans(struct sock *sk, struct mptcp_data_frag *dfrag) lock_sock(ssk); /* limit retransmission to the bytes already sent on some subflows */ - info.sent = 0; + info.sent = offset; info.limit = READ_ONCE(msk->csum_enabled) ? dfrag->data_len : dfrag->already_sent; @@ -XXX,XX +XXX,XX @@ static void __mptcp_retrans(struct sock *sk) struct mptcp_sock *msk = mptcp_sk(sk); struct mptcp_subflow_context *subflow; struct mptcp_data_frag *dfrag; + bool need_retrans; + u64 retrans_seq; int err, len; - mptcp_clean_una_wakeup(sk); - - /* first check ssk: need to kick "stale" logic */ - err = mptcp_sched_get_retrans(msk); + mptcp_data_lock(sk); + __mptcp_clean_una_wakeup(sk); + retrans_seq = msk->snd_una; dfrag = mptcp_rtx_head(sk); - if (!dfrag) { + need_retrans = !!dfrag; + mptcp_data_unlock(sk); + if (!dfrag) + goto check_data_fin; + + for (;;) { + bool already_retrans; + u64 sent_seq; + + /* The default scheduler will kick "stale" logic, that in + * turn can process incoming acks and clean the RTX queue; + * ensure that the current dfrag will still be around + * afterwards. + */ + get_page(dfrag->page); + err = mptcp_sched_get_retrans(msk); + if (err) { + put_page(dfrag->page); + break; + } + + /* Incoming acks can have moved retrans sequence after + * the current dfrag, if so try to start again from RTX head. + */ + mptcp_data_lock(sk); + already_retrans = !before64(msk->snd_una, dfrag->data_seq + + dfrag->already_sent); + put_page(dfrag->page); + if (already_retrans) { + __mptcp_clean_una_wakeup(sk); + retrans_seq = msk->snd_una; + dfrag = mptcp_rtx_head(sk); + need_retrans = !!dfrag; + } else if (after64(msk->snd_una, retrans_seq)) { + retrans_seq = msk->snd_una; + } + mptcp_data_unlock(sk); + + /* `already_sent` can be 0 for `dfrag` belonging to the RTX + * queue due to __mptcp_retransmit_pending_data(). + */ + if (!dfrag || !dfrag->already_sent) + break; + + /* Can fail only in case of fallback. */ + len = __mptcp_push_retrans(sk, dfrag, retrans_seq); + if (len < 0) + goto clear_scheduled; + + retrans_seq += len; + msk->bytes_retrans += len; + dfrag->already_sent = max_t(u16, dfrag->already_sent, + retrans_seq - dfrag->data_seq); + + /* With csum enabled retransmission can send new data. */ + sent_seq = dfrag->already_sent + dfrag->data_seq; + if (after64(sent_seq, msk->snd_nxt)) + WRITE_ONCE(msk->snd_nxt, sent_seq); + + /* Attempt the next fragment only if the current one is + * completely retransmitted. + */ + if (before64(retrans_seq, dfrag->data_seq + dfrag->data_len)) + break; + + dfrag = list_is_last(&dfrag->list, &msk->rtx_queue) ? + NULL : list_next_entry(dfrag, list); + if (!dfrag) + break; + } + + /* Attempt data-fin retransmission only when the RTX queue is empty. */ + if (!need_retrans) { +check_data_fin: if (mptcp_data_fin_enabled(msk)) { struct inet_connection_sock *icsk = inet_csk(sk); @@ -XXX,XX +XXX,XX @@ static void __mptcp_retrans(struct sock *sk) icsk->icsk_retransmits + 1); mptcp_set_datafin_timeout(sk); mptcp_send_ack(msk); - goto reset_timer; } if (!mptcp_send_head(sk)) goto clear_scheduled; - - goto reset_timer; } - if (err) - goto reset_timer; - - len = __mptcp_push_retrans(sk, dfrag); - if (len < 0) - goto clear_scheduled; - - msk->bytes_retrans += len; - dfrag->already_sent = max(dfrag->already_sent, len); - - /* With csum enabled retransmission can send new data. */ - if (after64(dfrag->already_sent + dfrag->data_seq, msk->snd_nxt)) - WRITE_ONCE(msk->snd_nxt, dfrag->already_sent + dfrag->data_seq); - reset_timer: mptcp_check_and_set_pending(sk); -- 2.54.0