877 lines · 21.7 KB · C
RawDownload
// SPDX-License-Identifier: GPL-2.0-or-later/* AF_RXRPC sendmsg() implementation. * * Copyright (C) 2007, 2016 Red Hat, Inc. All Rights Reserved. * Written by David Howells (dhowells@redhat.com) */ #define pr_fmt(fmt) KBUILD_MODNAME ": " fmt #include <linux/net.h>#include <linux/gfp.h>#include <linux/skbuff.h>#include <linux/export.h>#include <linux/sched/signal.h> #include <net/sock.h>#include <net/af_rxrpc.h>#include "ar-internal.h" /* * Propose an abort to be made in the I/O thread. */bool rxrpc_propose_abort(struct rxrpc_call *call, s32 abort_code, int error,			 enum rxrpc_abort_reason why){	_enter("{%d},%d,%d,%u", call->debug_id, abort_code, error, why); 	if (!call->send_abort && !rxrpc_call_is_complete(call)) {		call->send_abort_why = why;		call->send_abort_err = error;		call->send_abort_seq = 0;		trace_rxrpc_abort_call(call, abort_code);		/* Request abort locklessly vs rxrpc_input_call_event(). */		smp_store_release(&call->send_abort, abort_code);		rxrpc_poke_call(call, rxrpc_call_poke_abort);		return true;	} 	return false;} /* * Wait for a call to become connected.  Interruption here doesn't cause the * call to be aborted. */static int rxrpc_wait_to_be_connected(struct rxrpc_call *call, long *timeo){	DECLARE_WAITQUEUE(myself, current);	int ret = 0; 	_enter("%d", call->debug_id); 	if (rxrpc_call_state(call) != RXRPC_CALL_CLIENT_AWAIT_CONN)		goto no_wait; 	add_wait_queue_exclusive(&call->waitq, &myself); 	for (;;) {		switch (call->interruptibility) {		case RXRPC_INTERRUPTIBLE:		case RXRPC_PREINTERRUPTIBLE:			set_current_state(TASK_INTERRUPTIBLE);			break;		case RXRPC_UNINTERRUPTIBLE:		default:			set_current_state(TASK_UNINTERRUPTIBLE);			break;		} 		if (rxrpc_call_state(call) != RXRPC_CALL_CLIENT_AWAIT_CONN)			break;		if ((call->interruptibility == RXRPC_INTERRUPTIBLE ||		     call->interruptibility == RXRPC_PREINTERRUPTIBLE) &&		    signal_pending(current)) {			ret = sock_intr_errno(*timeo);			break;		}		*timeo = schedule_timeout(*timeo);	} 	remove_wait_queue(&call->waitq, &myself);	__set_current_state(TASK_RUNNING); no_wait:	if (ret == 0 && rxrpc_call_is_complete(call))		ret = call->error; 	_leave(" = %d", ret);	return ret;} /* * Return true if there's sufficient Tx queue space. */static bool rxrpc_check_tx_space(struct rxrpc_call *call, rxrpc_seq_t *_tx_win){	rxrpc_seq_t tx_bottom = READ_ONCE(call->tx_bottom); 	if (_tx_win)		*_tx_win = tx_bottom;	return call->send_top - tx_bottom < 256;} /* * Wait for space to appear in the Tx queue or a signal to occur. */static int rxrpc_wait_for_tx_window_intr(struct rxrpc_sock *rx,					 struct rxrpc_call *call,					 long *timeo){	for (;;) {		set_current_state(TASK_INTERRUPTIBLE);		if (rxrpc_check_tx_space(call, NULL))			return 0; 		if (rxrpc_call_is_complete(call))			return call->error; 		if (signal_pending(current))			return sock_intr_errno(*timeo); 		trace_rxrpc_txqueue(call, rxrpc_txqueue_wait);		*timeo = schedule_timeout(*timeo);	}} /* * Wait for space to appear in the Tx queue uninterruptibly, but with * a timeout of 2*RTT if no progress was made and a signal occurred. */static int rxrpc_wait_for_tx_window_waitall(struct rxrpc_sock *rx,					    struct rxrpc_call *call){	rxrpc_seq_t tx_start, tx_win;	signed long rtt, timeout; 	rtt = READ_ONCE(call->srtt_us) >> 3;	rtt = usecs_to_jiffies(rtt) * 2;	if (rtt < 2)		rtt = 2; 	timeout = rtt;	tx_start = READ_ONCE(call->tx_bottom); 	for (;;) {		set_current_state(TASK_UNINTERRUPTIBLE); 		if (rxrpc_check_tx_space(call, &tx_win))			return 0; 		if (rxrpc_call_is_complete(call))			return call->error; 		if (timeout == 0 &&		    tx_win == tx_start && signal_pending(current))			return -EINTR; 		if (tx_win != tx_start) {			timeout = rtt;			tx_start = tx_win;		} 		trace_rxrpc_txqueue(call, rxrpc_txqueue_wait);		timeout = schedule_timeout(timeout);	}} /* * Wait for space to appear in the Tx queue uninterruptibly. */static int rxrpc_wait_for_tx_window_nonintr(struct rxrpc_sock *rx,					    struct rxrpc_call *call,					    long *timeo){	for (;;) {		set_current_state(TASK_UNINTERRUPTIBLE);		if (rxrpc_check_tx_space(call, NULL))			return 0; 		if (rxrpc_call_is_complete(call))			return call->error; 		trace_rxrpc_txqueue(call, rxrpc_txqueue_wait);		*timeo = schedule_timeout(*timeo);	}} /* * wait for space to appear in the transmit/ACK window * - caller holds the socket locked */static int rxrpc_wait_for_tx_window(struct rxrpc_sock *rx,				    struct rxrpc_call *call,				    long *timeo,				    bool waitall){	DECLARE_WAITQUEUE(myself, current);	int ret; 	_enter(",{%u,%u,%u}",	       call->tx_bottom, call->tx_top, call->tx_winsize); 	add_wait_queue(&call->waitq, &myself); 	switch (call->interruptibility) {	case RXRPC_INTERRUPTIBLE:		if (waitall)			ret = rxrpc_wait_for_tx_window_waitall(rx, call);		else			ret = rxrpc_wait_for_tx_window_intr(rx, call, timeo);		break;	case RXRPC_PREINTERRUPTIBLE:	case RXRPC_UNINTERRUPTIBLE:	default:		ret = rxrpc_wait_for_tx_window_nonintr(rx, call, timeo);		break;	} 	remove_wait_queue(&call->waitq, &myself);	set_current_state(TASK_RUNNING);	_leave(" = %d", ret);	return ret;} /* * Notify the owner of the call that the transmit phase is ended and the last * packet has been queued. */static void rxrpc_notify_end_tx(struct rxrpc_sock *rx, struct rxrpc_call *call,				rxrpc_notify_end_tx_t notify_end_tx){	if (notify_end_tx)		notify_end_tx(&rx->sk, call, call->user_call_ID);} /* * Queue a DATA packet for transmission, set the resend timeout and send * the packet immediately.  Returns the error from rxrpc_send_data_packet() * in case the caller wants to do something with it. */static void rxrpc_queue_packet(struct rxrpc_sock *rx, struct rxrpc_call *call,			       struct rxrpc_txbuf *txb,			       rxrpc_notify_end_tx_t notify_end_tx){	struct rxrpc_txqueue *sq = call->send_queue;	rxrpc_seq_t seq = txb->seq;	bool poke, last = txb->flags & RXRPC_LAST_PACKET;	int ix = seq & RXRPC_TXQ_MASK;	rxrpc_inc_stat(call->rxnet, stat_tx_data); 	ASSERTCMP(txb->seq, ==, call->send_top + 1); 	if (last)		trace_rxrpc_txqueue(call, rxrpc_txqueue_queue_last);	else		trace_rxrpc_txqueue(call, rxrpc_txqueue_queue); 	if (WARN_ON_ONCE(sq->bufs[ix]))		trace_rxrpc_tq(call, sq, seq, rxrpc_tq_queue_dup);	else		trace_rxrpc_tq(call, sq, seq, rxrpc_tq_queue); 	/* Add the packet to the call's output buffer */	poke = (READ_ONCE(call->tx_bottom) == call->send_top);	sq->bufs[ix] = txb;	/* Order send_top after the queue->next pointer and txb content. */	smp_store_release(&call->send_top, seq);	if (last) {		set_bit(RXRPC_CALL_TX_NO_MORE, &call->flags);		rxrpc_notify_end_tx(rx, call, notify_end_tx);		call->send_queue = NULL;	} 	if (poke)		rxrpc_poke_call(call, rxrpc_call_poke_start);} /* * Allocate a new txqueue unit and add it to the transmission queue. */static int rxrpc_alloc_txqueue(struct sock *sk, struct rxrpc_call *call){	struct rxrpc_txqueue *tq; 	tq = kzalloc_obj(*tq, sk->sk_allocation);	if (!tq)		return -ENOMEM; 	tq->xmit_ts_base = KTIME_MIN;	for (int i = 0; i < RXRPC_NR_TXQUEUE; i++)		tq->segment_xmit_ts[i] = UINT_MAX; 	if (call->send_queue) {		tq->qbase = call->send_top + 1;		call->send_queue->next = tq;		call->send_queue = tq;	} else if (WARN_ON(call->tx_queue)) {		kfree(tq);		return -ENOMEM;	} else {		/* We start at seq 1, so pretend seq 0 is hard-acked. */		tq->nr_reported_acks = 1;		tq->segment_acked = 1UL;		tq->qbase = 0;		call->tx_qbase = 0;		call->send_queue = tq;		call->tx_qtail = tq;		call->tx_queue = tq;	} 	trace_rxrpc_tq(call, tq, call->send_top, rxrpc_tq_alloc);	return 0;} /* * send data through a socket * - must be called in process context * - The caller holds the call user access mutex, but not the socket lock. */static int rxrpc_send_data(struct rxrpc_sock *rx,			   struct rxrpc_call *call,			   struct msghdr *msg, size_t len,			   rxrpc_notify_end_tx_t notify_end_tx,			   bool *_dropped_lock){	struct rxrpc_txbuf *txb;	struct sock *sk = &rx->sk;	enum rxrpc_call_state state;	long timeo;	bool more = msg->msg_flags & MSG_MORE;	int ret, copied = 0; 	if (test_bit(RXRPC_CALL_TX_NO_MORE, &call->flags)) {		trace_rxrpc_abort(call->debug_id, rxrpc_sendmsg_late_send,				  call->cid, call->call_id, call->rx_consumed,				  0, -EPROTO);		return -EPROTO;	} 	timeo = sock_sndtimeo(sk, msg->msg_flags & MSG_DONTWAIT); 	ret = rxrpc_wait_to_be_connected(call, &timeo);	if (ret < 0)		return ret; 	if (call->conn->state == RXRPC_CONN_CLIENT_UNSECURED) {		ret = rxrpc_init_client_conn_security(call->conn);		if (ret < 0)			return ret;	} 	/* this should be in poll */	sk_clear_bit(SOCKWQ_ASYNC_NOSPACE, sk); reload:	txb = call->tx_pending;	call->tx_pending = NULL;	if (txb)		rxrpc_see_txbuf(txb, rxrpc_txbuf_see_send_more); 	ret = -EPIPE;	if (sk->sk_shutdown & SEND_SHUTDOWN)		goto maybe_error;	state = rxrpc_call_state(call);	ret = -ESHUTDOWN;	if (state >= RXRPC_CALL_COMPLETE)		goto maybe_error;	ret = -EPROTO;	if (state != RXRPC_CALL_CLIENT_PRE_SEND &&	    state != RXRPC_CALL_CLIENT_SEND_REQUEST &&	    state != RXRPC_CALL_SERVER_ACK_REQUEST &&	    state != RXRPC_CALL_SERVER_SEND_REPLY) {		/* Request phase complete for this client call */		trace_rxrpc_abort(call->debug_id, rxrpc_sendmsg_late_send,				  call->cid, call->call_id, call->rx_consumed,				  0, -EPROTO);		goto maybe_error;	} 	ret = -EMSGSIZE;	if (call->tx_total_len != -1) {		if (len - copied > call->tx_total_len)			goto maybe_error;		if (!more && len - copied != call->tx_total_len)			goto maybe_error;	} 	do {		if (!txb) {			size_t remain; 			_debug("alloc"); 			if (!rxrpc_check_tx_space(call, NULL))				goto wait_for_space; 			/* See if we need to begin/extend the Tx queue. */			if (!call->send_queue || !((call->send_top + 1) & RXRPC_TXQ_MASK)) {				ret = rxrpc_alloc_txqueue(sk, call);				if (ret < 0)					goto maybe_error;			} 			/* Work out the maximum size of a packet.  Assume that			 * the security header is going to be in the padded			 * region (enc blocksize), but the trailer is not.			 */			remain = more ? INT_MAX : msg_data_left(msg);			txb = call->conn->security->alloc_txbuf(call, remain, sk->sk_allocation);			if (!txb) {				ret = -ENOMEM;				goto maybe_error;			}		} 		_debug("append"); 		/* append next segment of data to the current buffer */		if (msg_data_left(msg) > 0) {			size_t copy = umin(txb->space, msg_data_left(msg)); 			_debug("add %zu", copy);			if (!copy_from_iter_full(txb->data + txb->offset,						 copy, &msg->msg_iter))				goto efault;			_debug("added");			txb->space -= copy;			txb->len += copy;			txb->offset += copy;			copied += copy;			if (call->tx_total_len != -1)				call->tx_total_len -= copy;		} 		/* check for the far side aborting the call or a network error		 * occurring */		if (rxrpc_call_is_complete(call))			goto call_terminated; 		/* add the packet to the send queue if it's now full */		if (!txb->space ||		    (msg_data_left(msg) == 0 && !more)) {			if (msg_data_left(msg) == 0 && !more)				txb->flags |= RXRPC_LAST_PACKET; 			ret = call->security->secure_packet(call, txb);			if (ret < 0)				goto out;			rxrpc_queue_packet(rx, call, txb, notify_end_tx);			txb = NULL;		}	} while (msg_data_left(msg) > 0); success:	ret = copied;	if (rxrpc_call_is_complete(call) &&	    call->error < 0)		ret = call->error;out:	call->tx_pending = txb;	_leave(" = %d", ret);	return ret; call_terminated:	rxrpc_put_txbuf(txb, rxrpc_txbuf_put_send_aborted);	_leave(" = %d", call->error);	return call->error; maybe_error:	if (copied)		goto success;	goto out; efault:	ret = -EFAULT;	goto out; wait_for_space:	ret = -EAGAIN;	if (msg->msg_flags & MSG_DONTWAIT)		goto maybe_error;	mutex_unlock(&call->user_mutex);	*_dropped_lock = true;	ret = rxrpc_wait_for_tx_window(rx, call, &timeo,				       msg->msg_flags & MSG_WAITALL);	if (ret < 0)		goto maybe_error;	if (call->interruptibility == RXRPC_INTERRUPTIBLE) {		if (mutex_lock_interruptible(&call->user_mutex) < 0) {			ret = sock_intr_errno(timeo);			goto maybe_error;		}	} else {		mutex_lock(&call->user_mutex);	}	*_dropped_lock = false;	goto reload;} /*

Lines 1–500 of 877.