diff --git a/net/tipc/socket.c b/net/tipc/socket.c index e564341e0216..73c7c966c98c 100644 --- a/net/tipc/socket.c +++ b/net/tipc/socket.c @@ -156,7 +156,8 @@ static int tipc_sk_insert(struct tipc_sock *tsk); static void tipc_sk_remove(struct tipc_sock *tsk); static int __tipc_sendstream(struct socket *sock, struct msghdr *m, size_t dsz); static int __tipc_sendmsg(struct socket *sock, struct msghdr *m, size_t dsz); -static void tipc_sk_push_backlog(struct tipc_sock *tsk, bool nagle_ack); +static void tipc_sk_push_backlog(struct tipc_sock *tsk, bool nagle_ack, + struct sk_buff_head *deferq); static int tipc_wait_for_connect(struct socket *sock, long *timeo_p); static const struct proto_ops packet_ops; @@ -560,7 +561,7 @@ static void __tipc_shutdown(struct socket *sock, int error) !tsk_conn_cong(tsk))); /* Push out delayed messages if in Nagle mode */ - tipc_sk_push_backlog(tsk, false); + tipc_sk_push_backlog(tsk, false, NULL); /* Remove pending SYN */ __skb_queue_purge(&sk->sk_write_queue); @@ -1268,7 +1269,8 @@ void tipc_sk_mcast_rcv(struct net *net, struct sk_buff_head *arrvq, /* tipc_sk_push_backlog(): send accumulated buffers in socket write queue * when socket is in Nagle mode */ -static void tipc_sk_push_backlog(struct tipc_sock *tsk, bool nagle_ack) +static void tipc_sk_push_backlog(struct tipc_sock *tsk, bool nagle_ack, + struct sk_buff_head *deferq) { struct sk_buff_head *txq = &tsk->sk.sk_write_queue; struct sk_buff *skb = skb_peek_tail(txq); @@ -1310,6 +1312,12 @@ static void tipc_sk_push_backlog(struct tipc_sock *tsk, bool nagle_ack) tsk->pkt_cnt += skb_queue_len(txq); tsk->snt_unacked += tsk->snd_backlog; tsk->snd_backlog = 0; + if (deferq) { + /* Block concurrent senders until the detached queue is sent. */ + tsk->cong_link_cnt = 1; + skb_queue_splice_tail_init(txq, deferq); + return; + } rc = tipc_node_xmit(net, txq, dnode, tsk->portid); if (rc == -ELINKCONG) tsk->cong_link_cnt = 1; @@ -1321,10 +1329,12 @@ static void tipc_sk_push_backlog(struct tipc_sock *tsk, bool nagle_ack) * @skb: pointer to message buffer. * @inputq: buffer list containing the buffers * @xmitq: output message area + * @deferq: socket write queue to transmit after releasing the socket lock */ static void tipc_sk_conn_proto_rcv(struct tipc_sock *tsk, struct sk_buff *skb, struct sk_buff_head *inputq, - struct sk_buff_head *xmitq) + struct sk_buff_head *xmitq, + struct sk_buff_head *deferq) { struct tipc_msg *hdr = buf_msg(skb); u32 onode = tsk_own_node(tsk); @@ -1367,7 +1377,7 @@ static void tipc_sk_conn_proto_rcv(struct tipc_sock *tsk, struct sk_buff *skb, goto exit; was_cong = tsk_conn_cong(tsk); - tipc_sk_push_backlog(tsk, msg_nagle_ack(hdr)); + tipc_sk_push_backlog(tsk, msg_nagle_ack(hdr), deferq); tsk->snt_unacked -= msg_conn_ack(hdr); if (tsk->peer_caps & TIPC_BLOCK_FLOWCTL) tsk->snd_win = msg_adv_win(hdr); @@ -2147,7 +2157,8 @@ static void tipc_sock_destruct(struct sock *sk) static void tipc_sk_proto_rcv(struct sock *sk, struct sk_buff_head *inputq, - struct sk_buff_head *xmitq) + struct sk_buff_head *xmitq, + struct sk_buff_head *deferq) { struct sk_buff *skb = __skb_dequeue(inputq); struct tipc_sock *tsk = tipc_sk(sk); @@ -2157,7 +2168,7 @@ static void tipc_sk_proto_rcv(struct sock *sk, switch (msg_user(hdr)) { case CONN_MANAGER: - tipc_sk_conn_proto_rcv(tsk, skb, inputq, xmitq); + tipc_sk_conn_proto_rcv(tsk, skb, inputq, xmitq, deferq); return; case SOCK_WAKEUP: tipc_dest_del(&tsk->cong_links, msg_orignode(hdr), 0); @@ -2165,7 +2176,7 @@ static void tipc_sk_proto_rcv(struct sock *sk, smp_wmb(); tsk->cong_link_cnt--; wakeup = true; - tipc_sk_push_backlog(tsk, false); + tipc_sk_push_backlog(tsk, false, deferq); break; case GROUP_PROTOCOL: tipc_group_proto_rcv(grp, &wakeup, hdr, inputq, xmitq); @@ -2189,10 +2200,12 @@ static void tipc_sk_proto_rcv(struct sock *sk, * @tsk: TIPC socket * @skb: pointer to message buffer. * @xmitq: for Nagle ACK if any + * @deferq: socket write queue to transmit after releasing the socket lock * Return: true if message should be added to receive queue, false otherwise */ static bool tipc_sk_filter_connect(struct tipc_sock *tsk, struct sk_buff *skb, - struct sk_buff_head *xmitq) + struct sk_buff_head *xmitq, + struct sk_buff_head *deferq) { struct sock *sk = &tsk->sk; struct net *net = sock_net(sk); @@ -2256,7 +2269,7 @@ static bool tipc_sk_filter_connect(struct tipc_sock *tsk, struct sk_buff *skb, return false; case TIPC_ESTABLISHED: if (!skb_queue_empty(&sk->sk_write_queue)) - tipc_sk_push_backlog(tsk, false); + tipc_sk_push_backlog(tsk, false, deferq); /* Accept only connection-based messages sent by peer */ if (likely(con_msg && !err && pport == oport && pnode == onode)) { @@ -2329,6 +2342,7 @@ static unsigned int rcvbuf_limit(struct sock *sk, struct sk_buff *skb) * @sk: socket * @skb: pointer to message. * @xmitq: output message area (FIXME) + * @deferq: socket write queue to transmit after releasing the socket lock * * Enqueues message on receive queue if acceptable; optionally handles * disconnect indication for a connected socket. @@ -2336,7 +2350,8 @@ static unsigned int rcvbuf_limit(struct sock *sk, struct sk_buff *skb) * Called with socket lock already taken */ static void tipc_sk_filter_rcv(struct sock *sk, struct sk_buff *skb, - struct sk_buff_head *xmitq) + struct sk_buff_head *xmitq, + struct sk_buff_head *deferq) { bool sk_conn = !tipc_sk_type_connectionless(sk); struct tipc_sock *tsk = tipc_sk(sk); @@ -2353,7 +2368,7 @@ static void tipc_sk_filter_rcv(struct sock *sk, struct sk_buff *skb, __skb_queue_tail(&inputq, skb); if (unlikely(!msg_isdata(hdr))) - tipc_sk_proto_rcv(sk, &inputq, xmitq); + tipc_sk_proto_rcv(sk, &inputq, xmitq, deferq); if (unlikely(grp)) tipc_group_filter_msg(grp, &inputq, xmitq); @@ -2365,7 +2380,8 @@ static void tipc_sk_filter_rcv(struct sock *sk, struct sk_buff *skb, while ((skb = __skb_dequeue(&inputq))) { hdr = buf_msg(skb); limit = rcvbuf_limit(sk, skb); - if ((sk_conn && !tipc_sk_filter_connect(tsk, skb, xmitq)) || + if ((sk_conn && + !tipc_sk_filter_connect(tsk, skb, xmitq, deferq)) || (!sk_conn && msg_connected(hdr)) || (!grp && msg_in_group(hdr))) err = TIPC_ERR_NO_PORT; @@ -2408,7 +2424,7 @@ static int tipc_sk_backlog_rcv(struct sock *sk, struct sk_buff *skb) __skb_queue_head_init(&xmitq); - tipc_sk_filter_rcv(sk, skb, &xmitq); + tipc_sk_filter_rcv(sk, skb, &xmitq, NULL); added = sk_rmem_alloc_get(sk) - before; atomic_add(added, &tipc_sk(sk)->dupl_rcvcnt); @@ -2424,11 +2440,13 @@ static int tipc_sk_backlog_rcv(struct sock *sk, struct sk_buff *skb) * @sk: socket where the buffers should be enqueued * @dport: port number for the socket * @xmitq: output queue + * @deferq: socket write queue to transmit after releasing the socket lock * * Caller must hold socket lock */ static void tipc_sk_enqueue(struct sk_buff_head *inputq, struct sock *sk, - u32 dport, struct sk_buff_head *xmitq) + u32 dport, struct sk_buff_head *xmitq, + struct sk_buff_head *deferq) { unsigned long time_limit = jiffies + usecs_to_jiffies(20000); struct sk_buff *skb; @@ -2446,7 +2464,9 @@ static void tipc_sk_enqueue(struct sk_buff_head *inputq, struct sock *sk, /* Add message directly to receive queue if possible */ if (!sock_owned_by_user(sk)) { - tipc_sk_filter_rcv(sk, skb, xmitq); + tipc_sk_filter_rcv(sk, skb, xmitq, deferq); + if (deferq && !skb_queue_empty(deferq)) + return; continue; } @@ -2483,6 +2503,7 @@ static void tipc_sk_enqueue(struct sk_buff_head *inputq, struct sock *sk, */ void tipc_sk_rcv(struct net *net, struct sk_buff_head *inputq) { + struct sk_buff_head deferq; struct sk_buff_head xmitq; u32 dnode, dport = 0; int err; @@ -2490,6 +2511,7 @@ void tipc_sk_rcv(struct net *net, struct sk_buff_head *inputq) struct sock *sk; struct sk_buff *skb; + __skb_queue_head_init(&deferq); __skb_queue_head_init(&xmitq); while (skb_queue_len(inputq)) { dport = tipc_skb_peek_port(inputq, dport); @@ -2498,9 +2520,22 @@ void tipc_sk_rcv(struct net *net, struct sk_buff_head *inputq) if (likely(tsk)) { sk = &tsk->sk; if (likely(spin_trylock_bh(&sk->sk_lock.slock))) { - tipc_sk_enqueue(inputq, sk, dport, &xmitq); + tipc_sk_enqueue(inputq, sk, dport, &xmitq, + &deferq); + if (!skb_queue_empty(&deferq)) + dnode = tsk_peer_node(tsk); spin_unlock_bh(&sk->sk_lock.slock); } + if (!skb_queue_empty(&deferq)) { + err = tipc_node_xmit(sock_net(sk), &deferq, + dnode, dport); + if (err != -ELINKCONG) { + spin_lock_bh(&sk->sk_lock.slock); + tsk->cong_link_cnt = 0; + sk->sk_write_space(sk); + spin_unlock_bh(&sk->sk_lock.slock); + } + } /* Send pending response/rejected messages, if any */ tipc_node_distr_xmit(sock_net(sk), &xmitq); sock_put(sk);