[linux-block.git] / net / rxrpc / io_thread.c

// SPDX-License-Identifier: GPL-2.0-or-later
/* RxRPC packet reception
 *
 * Copyright (C) 2007, 2016, 2022 Red Hat, Inc. All Rights Reserved.
 * Written by David Howells (dhowells@redhat.com)
 */

#define pr_fmt(fmt) KBUILD_MODNAME ": " fmt

#include "ar-internal.h"

static int rxrpc_input_packet_on_conn(struct rxrpc_connection *conn,
				      struct sockaddr_rxrpc *peer_srx,
				      struct sk_buff *skb);

/*
 * handle data received on the local endpoint
 * - may be called in interrupt context
 *
 * [!] Note that as this is called from the encap_rcv hook, the socket is not
 * held locked by the caller and nothing prevents sk_user_data on the UDP from
 * being cleared in the middle of processing this function.
 *
 * Called with the RCU read lock held from the IP layer via UDP.
 */
int rxrpc_encap_rcv(struct sock *udp_sk, struct sk_buff *skb)
{
	struct sk_buff_head *rx_queue;
	struct rxrpc_local *local = rcu_dereference_sk_user_data(udp_sk);
	struct task_struct *io_thread;

	if (unlikely(!local)) {
		kfree_skb(skb);
		return 0;
	}
	io_thread = READ_ONCE(local->io_thread);
	if (!io_thread) {
		kfree_skb(skb);
		return 0;
	}
	if (skb->tstamp == 0)
		skb->tstamp = ktime_get_real();

	skb->mark = RXRPC_SKB_MARK_PACKET;
	rxrpc_new_skb(skb, rxrpc_skb_new_encap_rcv);
	rx_queue = &local->rx_queue;
#ifdef CONFIG_AF_RXRPC_INJECT_RX_DELAY
	if (rxrpc_inject_rx_delay ||
	    !skb_queue_empty(&local->rx_delay_queue)) {
		skb->tstamp = ktime_add_ms(skb->tstamp, rxrpc_inject_rx_delay);
		rx_queue = &local->rx_delay_queue;
	}
#endif

	skb_queue_tail(rx_queue, skb);
	wake_up_process(io_thread);
	return 0;
}

/*
 * Handle an error received on the local endpoint.
 */
void rxrpc_error_report(struct sock *sk)
{
	struct rxrpc_local *local;
	struct sk_buff *skb;

	rcu_read_lock();
	local = rcu_dereference_sk_user_data(sk);
	if (unlikely(!local)) {
		rcu_read_unlock();
		return;
	}

	while ((skb = skb_dequeue(&sk->sk_error_queue))) {
		skb->mark = RXRPC_SKB_MARK_ERROR;
		rxrpc_new_skb(skb, rxrpc_skb_new_error_report);
		skb_queue_tail(&local->rx_queue, skb);
	}

	rxrpc_wake_up_io_thread(local);
	rcu_read_unlock();
}

/*
 * Directly produce an abort from a packet.
 */
bool rxrpc_direct_abort(struct sk_buff *skb, enum rxrpc_abort_reason why,
			s32 abort_code, int err)
{
	struct rxrpc_skb_priv *sp = rxrpc_skb(skb);

	trace_rxrpc_abort(0, why, sp->hdr.cid, sp->hdr.callNumber, sp->hdr.seq,
			  abort_code, err);
	skb->mark = RXRPC_SKB_MARK_REJECT_ABORT;
	skb->priority = abort_code;
	return false;
}

static bool rxrpc_bad_message(struct sk_buff *skb, enum rxrpc_abort_reason why)
{
	return rxrpc_direct_abort(skb, why, RX_PROTOCOL_ERROR, -EBADMSG);
}

#define just_discard true

/*
 * Process event packets targeted at a local endpoint.
 */
static bool rxrpc_input_version(struct rxrpc_local *local, struct sk_buff *skb)
{
	struct rxrpc_skb_priv *sp = rxrpc_skb(skb);
	char v;

	_enter("");

	rxrpc_see_skb(skb, rxrpc_skb_see_version);
	if (skb_copy_bits(skb, sizeof(struct rxrpc_wire_header), &v, 1) >= 0) {
		if (v == 0)
			rxrpc_send_version_request(local, &sp->hdr, skb);
	}

	return true;
}

/*
 * Extract the wire header from a packet and translate the byte order.
 */
static bool rxrpc_extract_header(struct rxrpc_skb_priv *sp,
				 struct sk_buff *skb)
{
	struct rxrpc_wire_header whdr;
	struct rxrpc_ackpacket ack;

	/* dig out the RxRPC connection details */
	if (skb_copy_bits(skb, 0, &whdr, sizeof(whdr)) < 0)
		return rxrpc_bad_message(skb, rxrpc_badmsg_short_hdr);

	memset(sp, 0, sizeof(*sp));
	sp->hdr.epoch		= ntohl(whdr.epoch);
	sp->hdr.cid		= ntohl(whdr.cid);
	sp->hdr.callNumber	= ntohl(whdr.callNumber);
	sp->hdr.seq		= ntohl(whdr.seq);
	sp->hdr.serial		= ntohl(whdr.serial);
	sp->hdr.flags		= whdr.flags;
	sp->hdr.type		= whdr.type;
	sp->hdr.userStatus	= whdr.userStatus;
	sp->hdr.securityIndex	= whdr.securityIndex;
	sp->hdr._rsvd		= ntohs(whdr._rsvd);
	sp->hdr.serviceId	= ntohs(whdr.serviceId);

	if (sp->hdr.type == RXRPC_PACKET_TYPE_ACK) {
		if (skb_copy_bits(skb, sizeof(whdr), &ack, sizeof(ack)) < 0)
			return rxrpc_bad_message(skb, rxrpc_badmsg_short_ack);
		sp->ack.first_ack	= ntohl(ack.firstPacket);
		sp->ack.prev_ack	= ntohl(ack.previousPacket);
		sp->ack.acked_serial	= ntohl(ack.serial);
		sp->ack.reason		= ack.reason;
		sp->ack.nr_acks		= ack.nAcks;
	}
	return true;
}

/*
 * Extract the abort code from an ABORT packet and stash it in skb->priority.
 */
static bool rxrpc_extract_abort(struct sk_buff *skb)
{
	__be32 wtmp;

	if (skb_copy_bits(skb, sizeof(struct rxrpc_wire_header),
			  &wtmp, sizeof(wtmp)) < 0)
		return false;
	skb->priority = ntohl(wtmp);
	return true;
}

/*
 * Process packets received on the local endpoint
 */
static bool rxrpc_input_packet(struct rxrpc_local *local, struct sk_buff **_skb)
{
	struct rxrpc_connection *conn;
	struct sockaddr_rxrpc peer_srx;
	struct rxrpc_skb_priv *sp;
	struct rxrpc_peer *peer = NULL;
	struct sk_buff *skb = *_skb;
	bool ret = false;

	skb_pull(skb, sizeof(struct udphdr));

	sp = rxrpc_skb(skb);

	/* dig out the RxRPC connection details */
	if (!rxrpc_extract_header(sp, skb))
		return just_discard;

	if (IS_ENABLED(CONFIG_AF_RXRPC_INJECT_LOSS)) {
		static int lose;
		if ((lose++ & 7) == 7) {
			trace_rxrpc_rx_lose(sp);
			return just_discard;
		}
	}

	trace_rxrpc_rx_packet(sp);

	switch (sp->hdr.type) {
	case RXRPC_PACKET_TYPE_VERSION:
		if (rxrpc_to_client(sp))
			return just_discard;
		return rxrpc_input_version(local, skb);

	case RXRPC_PACKET_TYPE_BUSY:
		if (rxrpc_to_server(sp))
			return just_discard;
		fallthrough;
	case RXRPC_PACKET_TYPE_ACK:
	case RXRPC_PACKET_TYPE_ACKALL:
		if (sp->hdr.callNumber == 0)
			return rxrpc_bad_message(skb, rxrpc_badmsg_zero_call);
		break;
	case RXRPC_PACKET_TYPE_ABORT:
		if (!rxrpc_extract_abort(skb))
			return just_discard; /* Just discard if malformed */
		break;

	case RXRPC_PACKET_TYPE_DATA:
		if (sp->hdr.callNumber == 0)
			return rxrpc_bad_message(skb, rxrpc_badmsg_zero_call);
		if (sp->hdr.seq == 0)
			return rxrpc_bad_message(skb, rxrpc_badmsg_zero_seq);

		/* Unshare the packet so that it can be modified for in-place
		 * decryption.
		 */
		if (sp->hdr.securityIndex != 0) {
			skb = skb_unshare(skb, GFP_ATOMIC);
			if (!skb) {
				rxrpc_eaten_skb(*_skb, rxrpc_skb_eaten_by_unshare_nomem);
				*_skb = NULL;
				return just_discard;
			}

			if (skb != *_skb) {
				rxrpc_eaten_skb(*_skb, rxrpc_skb_eaten_by_unshare);
				*_skb = skb;
				rxrpc_new_skb(skb, rxrpc_skb_new_unshared);
				sp = rxrpc_skb(skb);
			}
		}
		break;

	case RXRPC_PACKET_TYPE_CHALLENGE:
		if (rxrpc_to_server(sp))
			return just_discard;
		break;
	case RXRPC_PACKET_TYPE_RESPONSE:
		if (rxrpc_to_client(sp))
			return just_discard;
		break;

		/* Packet types 9-11 should just be ignored. */
	case RXRPC_PACKET_TYPE_PARAMS:
	case RXRPC_PACKET_TYPE_10:
	case RXRPC_PACKET_TYPE_11:
		return just_discard;

	default:
		return rxrpc_bad_message(skb, rxrpc_badmsg_unsupported_packet);
	}

	if (sp->hdr.serviceId == 0)
		return rxrpc_bad_message(skb, rxrpc_badmsg_zero_service);

	if (WARN_ON_ONCE(rxrpc_extract_addr_from_skb(&peer_srx, skb) < 0))
		return just_discard; /* Unsupported address type. */

	if (peer_srx.transport.family != local->srx.transport.family &&
	    (peer_srx.transport.family == AF_INET &&
	     local->srx.transport.family != AF_INET6)) {
		pr_warn_ratelimited("AF_RXRPC: Protocol mismatch %u not %u\n",
				    peer_srx.transport.family,
				    local->srx.transport.family);
		return just_discard; /* Wrong address type. */
	}

	if (rxrpc_to_client(sp)) {
		rcu_read_lock();
		conn = rxrpc_find_client_connection_rcu(local, &peer_srx, skb);
		conn = rxrpc_get_connection_maybe(conn, rxrpc_conn_get_call_input);
		rcu_read_unlock();
		if (!conn)
			return rxrpc_protocol_error(skb, rxrpc_eproto_no_client_conn);

		ret = rxrpc_input_packet_on_conn(conn, &peer_srx, skb);
		rxrpc_put_connection(conn, rxrpc_conn_put_call_input);
		return ret;
	}

	/* We need to look up service connections by the full protocol
	 * parameter set.  We look up the peer first as an intermediate step
	 * and then the connection from the peer's tree.
	 */
	rcu_read_lock();

	peer = rxrpc_lookup_peer_rcu(local, &peer_srx);
	if (!peer) {
		rcu_read_unlock();
		return rxrpc_new_incoming_call(local, NULL, NULL, &peer_srx, skb);
	}

	conn = rxrpc_find_service_conn_rcu(peer, skb);
	conn = rxrpc_get_connection_maybe(conn, rxrpc_conn_get_call_input);
	if (conn) {
		rcu_read_unlock();
		ret = rxrpc_input_packet_on_conn(conn, &peer_srx, skb);
		rxrpc_put_connection(conn, rxrpc_conn_put_call_input);
		return ret;
	}

	peer = rxrpc_get_peer_maybe(peer, rxrpc_peer_get_input);
	rcu_read_unlock();

	ret = rxrpc_new_incoming_call(local, peer, NULL, &peer_srx, skb);
	rxrpc_put_peer(peer, rxrpc_peer_put_input);
	return ret;
}

/*
 * Deal with a packet that's associated with an extant connection.
 */
static int rxrpc_input_packet_on_conn(struct rxrpc_connection *conn,
				      struct sockaddr_rxrpc *peer_srx,
				      struct sk_buff *skb)
{
	struct rxrpc_skb_priv *sp = rxrpc_skb(skb);
	struct rxrpc_channel *chan;
	struct rxrpc_call *call = NULL;
	unsigned int channel;

	if (sp->hdr.securityIndex != conn->security_ix)
		return rxrpc_direct_abort(skb, rxrpc_eproto_wrong_security,
					  RXKADINCONSISTENCY, -EBADMSG);

	if (sp->hdr.serviceId != conn->service_id) {
		int old_id;

		if (!test_bit(RXRPC_CONN_PROBING_FOR_UPGRADE, &conn->flags))
			return rxrpc_protocol_error(skb, rxrpc_eproto_reupgrade);

		old_id = cmpxchg(&conn->service_id, conn->orig_service_id,
				 sp->hdr.serviceId);
		if (old_id != conn->orig_service_id &&
		    old_id != sp->hdr.serviceId)
			return rxrpc_protocol_error(skb, rxrpc_eproto_bad_upgrade);
	}

	if (after(sp->hdr.serial, conn->hi_serial))
		conn->hi_serial = sp->hdr.serial;

	/* It's a connection-level packet if the call number is 0. */
	if (sp->hdr.callNumber == 0)
		return rxrpc_input_conn_packet(conn, skb);

	/* Deal with path MTU discovery probing. */
	if (sp->hdr.type == RXRPC_PACKET_TYPE_ACK &&
	    conn->pmtud_probe &&
	    after_eq(sp->ack.acked_serial, conn->pmtud_probe))
		rxrpc_input_probe_for_pmtud(conn, sp->ack.acked_serial, false);

	/* Call-bound packets are routed by connection channel. */
	channel = sp->hdr.cid & RXRPC_CHANNELMASK;
	chan = &conn->channels[channel];

	/* Ignore really old calls */
	if (sp->hdr.callNumber < chan->last_call)
		return just_discard;

	if (sp->hdr.callNumber == chan->last_call) {
		if (chan->call ||
		    sp->hdr.type == RXRPC_PACKET_TYPE_ABORT)
			return just_discard;

		/* For the previous service call, if completed successfully, we
		 * discard all further packets.
		 */
		if (rxrpc_conn_is_service(conn) &&
		    chan->last_type == RXRPC_PACKET_TYPE_ACK)
			return just_discard;

		/* But otherwise we need to retransmit the final packet from
		 * data cached in the connection record.
		 */
		if (sp->hdr.type == RXRPC_PACKET_TYPE_DATA)
			trace_rxrpc_rx_data(chan->call_debug_id,
					    sp->hdr.seq,
					    sp->hdr.serial,
					    sp->hdr.flags);
		rxrpc_conn_retransmit_call(conn, skb, channel);
		return just_discard;
	}

	call = rxrpc_try_get_call(chan->call, rxrpc_call_get_input);

	if (sp->hdr.callNumber > chan->call_id) {
		if (rxrpc_to_client(sp)) {
			rxrpc_put_call(call, rxrpc_call_put_input);
			return rxrpc_protocol_error(skb,
						    rxrpc_eproto_unexpected_implicit_end);
		}

		if (call) {
			rxrpc_implicit_end_call(call, skb);
			rxrpc_put_call(call, rxrpc_call_put_input);
			call = NULL;
		}
	}

	if (!call) {
		if (rxrpc_to_client(sp))
			return rxrpc_protocol_error(skb, rxrpc_eproto_no_client_call);
		return rxrpc_new_incoming_call(conn->local, conn->peer, conn,
					       peer_srx, skb);
	}

	rxrpc_queue_rx_call_packet(call, skb);
	rxrpc_put_call(call, rxrpc_call_put_input);
	return true;
}

/*
 * I/O and event handling thread.
 */
int rxrpc_io_thread(void *data)
{
	struct rxrpc_connection *conn;
	struct sk_buff_head rx_queue;
	struct rxrpc_local *local = data;
	struct rxrpc_call *call;
	struct sk_buff *skb;
#ifdef CONFIG_AF_RXRPC_INJECT_RX_DELAY
	ktime_t now;
#endif
	bool should_stop;
	LIST_HEAD(conn_attend_q);
	LIST_HEAD(call_attend_q);

	complete(&local->io_thread_ready);

	skb_queue_head_init(&rx_queue);

	set_user_nice(current, MIN_NICE);

	for (;;) {
		rxrpc_inc_stat(local->rxnet, stat_io_loop);

		/* Inject a delay into packets if requested. */
#ifdef CONFIG_AF_RXRPC_INJECT_RX_DELAY
		now = ktime_get_real();
		while ((skb = skb_peek(&local->rx_delay_queue))) {
			if (ktime_before(now, skb->tstamp))
				break;
			skb = skb_dequeue(&local->rx_delay_queue);
			skb_queue_tail(&local->rx_queue, skb);
		}
#endif

		if (!skb_queue_empty(&local->rx_queue)) {
			spin_lock_irq(&local->rx_queue.lock);
			skb_queue_splice_tail_init(&local->rx_queue, &rx_queue);
			spin_unlock_irq(&local->rx_queue.lock);
			trace_rxrpc_iothread_rx(local, skb_queue_len(&rx_queue));
		}

		/* Distribute packets and errors. */
		while ((skb = __skb_dequeue(&rx_queue))) {
			struct rxrpc_skb_priv *sp = rxrpc_skb(skb);
			switch (skb->mark) {
			case RXRPC_SKB_MARK_PACKET:
				skb->priority = 0;
				if (!rxrpc_input_packet(local, &skb))
					rxrpc_reject_packet(local, skb);
				trace_rxrpc_rx_done(skb->mark, skb->priority);
				rxrpc_free_skb(skb, rxrpc_skb_put_input);
				break;
			case RXRPC_SKB_MARK_ERROR:
				rxrpc_input_error(local, skb);
				rxrpc_free_skb(skb, rxrpc_skb_put_error_report);
				break;
			case RXRPC_SKB_MARK_SERVICE_CONN_SECURED:
				rxrpc_input_conn_event(sp->poke_conn, skb);
				rxrpc_put_connection(sp->poke_conn, rxrpc_conn_put_poke);
				rxrpc_free_skb(skb, rxrpc_skb_put_conn_secured);
				break;
			default:
				WARN_ON_ONCE(1);
				rxrpc_free_skb(skb, rxrpc_skb_put_unknown);
				break;
			}
		}

		/* Deal with connections that want immediate attention. */
		if (!list_empty_careful(&local->conn_attend_q)) {
			spin_lock_irq(&local->lock);
			list_splice_tail_init(&local->conn_attend_q, &conn_attend_q);
			spin_unlock_irq(&local->lock);
		}

		while ((conn = list_first_entry_or_null(&conn_attend_q,
							struct rxrpc_connection,
							attend_link))) {
			spin_lock_irq(&local->lock);
			list_del_init(&conn->attend_link);
			spin_unlock_irq(&local->lock);
			rxrpc_input_conn_event(conn, NULL);
			rxrpc_put_connection(conn, rxrpc_conn_put_poke);
		}

		if (test_and_clear_bit(RXRPC_CLIENT_CONN_REAP_TIMER,
				       &local->client_conn_flags))
			rxrpc_discard_expired_client_conns(local);

		/* Deal with calls that want immediate attention. */
		spin_lock_irq(&local->lock);
		list_splice_tail_init(&local->call_attend_q, &call_attend_q);
		spin_unlock_irq(&local->lock);

		while ((call = list_first_entry_or_null(&call_attend_q,
							struct rxrpc_call,
							attend_link))) {
			spin_lock_irq(&local->lock);
			list_del_init(&call->attend_link);
			spin_unlock_irq(&local->lock);
			trace_rxrpc_call_poked(call);
			rxrpc_input_call_event(call);
			rxrpc_put_call(call, rxrpc_call_put_poke);
		}

		if (!list_empty(&local->new_client_calls))
			rxrpc_connect_client_calls(local);

		set_current_state(TASK_INTERRUPTIBLE);
		should_stop = kthread_should_stop();
		if (!skb_queue_empty(&local->rx_queue) ||
		    !list_empty(&local->call_attend_q) ||
		    !list_empty(&local->conn_attend_q) ||
		    !list_empty(&local->new_client_calls) ||
		    test_bit(RXRPC_CLIENT_CONN_REAP_TIMER,
			     &local->client_conn_flags)) {
			__set_current_state(TASK_RUNNING);
			continue;
		}

		if (should_stop)
			break;

#ifdef CONFIG_AF_RXRPC_INJECT_RX_DELAY
		skb = skb_peek(&local->rx_delay_queue);
		if (skb) {
			unsigned long timeout;
			ktime_t tstamp = skb->tstamp;
			ktime_t now = ktime_get_real();
			s64 delay_ns = ktime_to_ns(ktime_sub(tstamp, now));

			if (delay_ns <= 0) {
				__set_current_state(TASK_RUNNING);
				continue;
			}

			timeout = nsecs_to_jiffies(delay_ns);
			timeout = umax(timeout, 1);
			schedule_timeout(timeout);
			__set_current_state(TASK_RUNNING);
			continue;
		}
#endif

		schedule();
	}

	__set_current_state(TASK_RUNNING);
	rxrpc_see_local(local, rxrpc_local_stop);
	rxrpc_destroy_local(local);
	WRITE_ONCE(local->io_thread, NULL);
	rxrpc_see_local(local, rxrpc_local_stopped);
	return 0;
}
Commit	Line	Data
	1	// SPDX-License-Identifier: GPL-2.0-or-later
	2	/* RxRPC packet reception
	3	*
	4	* Copyright (C) 2007, 2016, 2022 Red Hat, Inc. All Rights Reserved.
	5	* Written by David Howells (dhowells@redhat.com)
	6	*/
	7
	8	#define pr_fmt(fmt) KBUILD_MODNAME ": " fmt
	9
	10	#include "ar-internal.h"
	11
	12	static int rxrpc_input_packet_on_conn(struct rxrpc_connection *conn,
	13	struct sockaddr_rxrpc *peer_srx,
	14	struct sk_buff *skb);
	15
	16	/*
	17	* handle data received on the local endpoint
	18	* - may be called in interrupt context
	19	*
	20	* [!] Note that as this is called from the encap_rcv hook, the socket is not
	21	* held locked by the caller and nothing prevents sk_user_data on the UDP from
	22	* being cleared in the middle of processing this function.
	23	*
	24	* Called with the RCU read lock held from the IP layer via UDP.
	25	*/
	26	int rxrpc_encap_rcv(struct sock udp_sk, struct sk_buff skb)
	27	{
	28	struct sk_buff_head *rx_queue;
	29	struct rxrpc_local *local = rcu_dereference_sk_user_data(udp_sk);
	30	struct task_struct *io_thread;
	31
	32	if (unlikely(!local)) {
	33	kfree_skb(skb);
	34	return 0;
	35	}
	36	io_thread = READ_ONCE(local->io_thread);
	37	if (!io_thread) {
	38	kfree_skb(skb);
	39	return 0;
	40	}
	41	if (skb->tstamp == 0)
	42	skb->tstamp = ktime_get_real();
	43
	44	skb->mark = RXRPC_SKB_MARK_PACKET;
	45	rxrpc_new_skb(skb, rxrpc_skb_new_encap_rcv);
	46	rx_queue = &local->rx_queue;
	47	#ifdef CONFIG_AF_RXRPC_INJECT_RX_DELAY
	48	if (rxrpc_inject_rx_delay \|\|
	49	!skb_queue_empty(&local->rx_delay_queue)) {
	50	skb->tstamp = ktime_add_ms(skb->tstamp, rxrpc_inject_rx_delay);
	51	rx_queue = &local->rx_delay_queue;
	52	}
	53	#endif
	54
	55	skb_queue_tail(rx_queue, skb);
	56	wake_up_process(io_thread);
	57	return 0;
	58	}
	59
	60	/*
	61	* Handle an error received on the local endpoint.
	62	*/
	63	void rxrpc_error_report(struct sock *sk)
	64	{
	65	struct rxrpc_local *local;
	66	struct sk_buff *skb;
	67
	68	rcu_read_lock();
	69	local = rcu_dereference_sk_user_data(sk);
	70	if (unlikely(!local)) {
	71	rcu_read_unlock();
	72	return;
	73	}
	74
	75	while ((skb = skb_dequeue(&sk->sk_error_queue))) {
	76	skb->mark = RXRPC_SKB_MARK_ERROR;
	77	rxrpc_new_skb(skb, rxrpc_skb_new_error_report);
	78	skb_queue_tail(&local->rx_queue, skb);
	79	}
	80
	81	rxrpc_wake_up_io_thread(local);
	82	rcu_read_unlock();
	83	}
	84
	85	/*
	86	* Directly produce an abort from a packet.
	87	*/
	88	bool rxrpc_direct_abort(struct sk_buff *skb, enum rxrpc_abort_reason why,
	89	s32 abort_code, int err)
	90	{
	91	struct rxrpc_skb_priv *sp = rxrpc_skb(skb);
	92
	93	trace_rxrpc_abort(0, why, sp->hdr.cid, sp->hdr.callNumber, sp->hdr.seq,
	94	abort_code, err);
	95	skb->mark = RXRPC_SKB_MARK_REJECT_ABORT;
	96	skb->priority = abort_code;
	97	return false;
	98	}
	99
	100	static bool rxrpc_bad_message(struct sk_buff *skb, enum rxrpc_abort_reason why)
	101	{
	102	return rxrpc_direct_abort(skb, why, RX_PROTOCOL_ERROR, -EBADMSG);
	103	}
	104
	105	#define just_discard true
	106
	107	/*
	108	* Process event packets targeted at a local endpoint.
	109	*/
	110	static bool rxrpc_input_version(struct rxrpc_local local, struct sk_buff skb)
	111	{
	112	struct rxrpc_skb_priv *sp = rxrpc_skb(skb);
	113	char v;
	114
	115	_enter("");
	116
	117	rxrpc_see_skb(skb, rxrpc_skb_see_version);
	118	if (skb_copy_bits(skb, sizeof(struct rxrpc_wire_header), &v, 1) >= 0) {
	119	if (v == 0)
	120	rxrpc_send_version_request(local, &sp->hdr, skb);
	121	}
	122
	123	return true;
	124	}
	125
	126	/*
	127	* Extract the wire header from a packet and translate the byte order.
	128	*/
	129	static bool rxrpc_extract_header(struct rxrpc_skb_priv *sp,
	130	struct sk_buff *skb)
	131	{
	132	struct rxrpc_wire_header whdr;
	133	struct rxrpc_ackpacket ack;
	134
	135	/* dig out the RxRPC connection details */
	136	if (skb_copy_bits(skb, 0, &whdr, sizeof(whdr)) < 0)
	137	return rxrpc_bad_message(skb, rxrpc_badmsg_short_hdr);
	138
	139	memset(sp, 0, sizeof(*sp));
	140	sp->hdr.epoch = ntohl(whdr.epoch);
	141	sp->hdr.cid = ntohl(whdr.cid);
	142	sp->hdr.callNumber = ntohl(whdr.callNumber);
	143	sp->hdr.seq = ntohl(whdr.seq);
	144	sp->hdr.serial = ntohl(whdr.serial);
	145	sp->hdr.flags = whdr.flags;
	146	sp->hdr.type = whdr.type;
	147	sp->hdr.userStatus = whdr.userStatus;
	148	sp->hdr.securityIndex = whdr.securityIndex;
	149	sp->hdr._rsvd = ntohs(whdr._rsvd);
	150	sp->hdr.serviceId = ntohs(whdr.serviceId);
	151
	152	if (sp->hdr.type == RXRPC_PACKET_TYPE_ACK) {
	153	if (skb_copy_bits(skb, sizeof(whdr), &ack, sizeof(ack)) < 0)
	154	return rxrpc_bad_message(skb, rxrpc_badmsg_short_ack);
	155	sp->ack.first_ack = ntohl(ack.firstPacket);
	156	sp->ack.prev_ack = ntohl(ack.previousPacket);
	157	sp->ack.acked_serial = ntohl(ack.serial);
	158	sp->ack.reason = ack.reason;
	159	sp->ack.nr_acks = ack.nAcks;
	160	}
	161	return true;
	162	}
	163
	164	/*
	165	* Extract the abort code from an ABORT packet and stash it in skb->priority.
	166	*/
	167	static bool rxrpc_extract_abort(struct sk_buff *skb)
	168	{
	169	__be32 wtmp;
	170
	171	if (skb_copy_bits(skb, sizeof(struct rxrpc_wire_header),
	172	&wtmp, sizeof(wtmp)) < 0)
	173	return false;
	174	skb->priority = ntohl(wtmp);
	175	return true;
	176	}
	177
	178	/*
	179	* Process packets received on the local endpoint
	180	*/
	181	static bool rxrpc_input_packet(struct rxrpc_local local, struct sk_buff *_skb)
	182	{
	183	struct rxrpc_connection *conn;
	184	struct sockaddr_rxrpc peer_srx;
	185	struct rxrpc_skb_priv *sp;
	186	struct rxrpc_peer *peer = NULL;
	187	struct sk_buff skb = _skb;
	188	bool ret = false;
	189
	190	skb_pull(skb, sizeof(struct udphdr));
	191
	192	sp = rxrpc_skb(skb);
	193
	194	/* dig out the RxRPC connection details */
	195	if (!rxrpc_extract_header(sp, skb))
	196	return just_discard;
	197
	198	if (IS_ENABLED(CONFIG_AF_RXRPC_INJECT_LOSS)) {
	199	static int lose;
	200	if ((lose++ & 7) == 7) {
	201	trace_rxrpc_rx_lose(sp);
	202	return just_discard;
	203	}
	204	}
	205
	206	trace_rxrpc_rx_packet(sp);
	207
	208	switch (sp->hdr.type) {
	209	case RXRPC_PACKET_TYPE_VERSION:
	210	if (rxrpc_to_client(sp))
	211	return just_discard;
	212	return rxrpc_input_version(local, skb);
	213
	214	case RXRPC_PACKET_TYPE_BUSY:
	215	if (rxrpc_to_server(sp))
	216	return just_discard;
	217	fallthrough;
	218	case RXRPC_PACKET_TYPE_ACK:
	219	case RXRPC_PACKET_TYPE_ACKALL:
	220	if (sp->hdr.callNumber == 0)
	221	return rxrpc_bad_message(skb, rxrpc_badmsg_zero_call);
	222	break;
	223	case RXRPC_PACKET_TYPE_ABORT:
	224	if (!rxrpc_extract_abort(skb))
	225	return just_discard; /* Just discard if malformed */
	226	break;
	227
	228	case RXRPC_PACKET_TYPE_DATA:
	229	if (sp->hdr.callNumber == 0)
	230	return rxrpc_bad_message(skb, rxrpc_badmsg_zero_call);
	231	if (sp->hdr.seq == 0)
	232	return rxrpc_bad_message(skb, rxrpc_badmsg_zero_seq);
	233
	234	/* Unshare the packet so that it can be modified for in-place
	235	* decryption.
	236	*/
	237	if (sp->hdr.securityIndex != 0) {
	238	skb = skb_unshare(skb, GFP_ATOMIC);
	239	if (!skb) {
	240	rxrpc_eaten_skb(*_skb, rxrpc_skb_eaten_by_unshare_nomem);
	241	*_skb = NULL;
	242	return just_discard;
	243	}
	244
	245	if (skb != *_skb) {
	246	rxrpc_eaten_skb(*_skb, rxrpc_skb_eaten_by_unshare);
	247	*_skb = skb;
	248	rxrpc_new_skb(skb, rxrpc_skb_new_unshared);
	249	sp = rxrpc_skb(skb);
	250	}
	251	}
	252	break;
	253
	254	case RXRPC_PACKET_TYPE_CHALLENGE:
	255	if (rxrpc_to_server(sp))
	256	return just_discard;
	257	break;
	258	case RXRPC_PACKET_TYPE_RESPONSE:
	259	if (rxrpc_to_client(sp))
	260	return just_discard;
	261	break;
	262
	263	/* Packet types 9-11 should just be ignored. */
	264	case RXRPC_PACKET_TYPE_PARAMS:
	265	case RXRPC_PACKET_TYPE_10:
	266	case RXRPC_PACKET_TYPE_11:
	267	return just_discard;
	268
	269	default:
	270	return rxrpc_bad_message(skb, rxrpc_badmsg_unsupported_packet);
	271	}
	272
	273	if (sp->hdr.serviceId == 0)
	274	return rxrpc_bad_message(skb, rxrpc_badmsg_zero_service);
	275
	276	if (WARN_ON_ONCE(rxrpc_extract_addr_from_skb(&peer_srx, skb) < 0))
	277	return just_discard; /* Unsupported address type. */
	278
	279	if (peer_srx.transport.family != local->srx.transport.family &&
	280	(peer_srx.transport.family == AF_INET &&
	281	local->srx.transport.family != AF_INET6)) {
	282	pr_warn_ratelimited("AF_RXRPC: Protocol mismatch %u not %u\n",
	283	peer_srx.transport.family,
	284	local->srx.transport.family);
	285	return just_discard; /* Wrong address type. */
	286	}
	287
	288	if (rxrpc_to_client(sp)) {
	289	rcu_read_lock();
	290	conn = rxrpc_find_client_connection_rcu(local, &peer_srx, skb);
	291	conn = rxrpc_get_connection_maybe(conn, rxrpc_conn_get_call_input);
	292	rcu_read_unlock();
	293	if (!conn)
	294	return rxrpc_protocol_error(skb, rxrpc_eproto_no_client_conn);
	295
	296	ret = rxrpc_input_packet_on_conn(conn, &peer_srx, skb);
	297	rxrpc_put_connection(conn, rxrpc_conn_put_call_input);
	298	return ret;
	299	}
	300
	301	/* We need to look up service connections by the full protocol
	302	* parameter set. We look up the peer first as an intermediate step
	303	* and then the connection from the peer's tree.
	304	*/
	305	rcu_read_lock();
	306
	307	peer = rxrpc_lookup_peer_rcu(local, &peer_srx);
	308	if (!peer) {
	309	rcu_read_unlock();
	310	return rxrpc_new_incoming_call(local, NULL, NULL, &peer_srx, skb);
	311	}
	312
	313	conn = rxrpc_find_service_conn_rcu(peer, skb);
	314	conn = rxrpc_get_connection_maybe(conn, rxrpc_conn_get_call_input);
	315	if (conn) {
	316	rcu_read_unlock();
	317	ret = rxrpc_input_packet_on_conn(conn, &peer_srx, skb);
	318	rxrpc_put_connection(conn, rxrpc_conn_put_call_input);
	319	return ret;
	320	}
	321
	322	peer = rxrpc_get_peer_maybe(peer, rxrpc_peer_get_input);
	323	rcu_read_unlock();
	324
	325	ret = rxrpc_new_incoming_call(local, peer, NULL, &peer_srx, skb);
	326	rxrpc_put_peer(peer, rxrpc_peer_put_input);
	327	return ret;
	328	}
	329
	330	/*
	331	* Deal with a packet that's associated with an extant connection.
	332	*/
	333	static int rxrpc_input_packet_on_conn(struct rxrpc_connection *conn,
	334	struct sockaddr_rxrpc *peer_srx,
	335	struct sk_buff *skb)
	336	{
	337	struct rxrpc_skb_priv *sp = rxrpc_skb(skb);
	338	struct rxrpc_channel *chan;
	339	struct rxrpc_call *call = NULL;
	340	unsigned int channel;
	341
	342	if (sp->hdr.securityIndex != conn->security_ix)
	343	return rxrpc_direct_abort(skb, rxrpc_eproto_wrong_security,
	344	RXKADINCONSISTENCY, -EBADMSG);
	345
	346	if (sp->hdr.serviceId != conn->service_id) {
	347	int old_id;
	348
	349	if (!test_bit(RXRPC_CONN_PROBING_FOR_UPGRADE, &conn->flags))
	350	return rxrpc_protocol_error(skb, rxrpc_eproto_reupgrade);
	351
	352	old_id = cmpxchg(&conn->service_id, conn->orig_service_id,
	353	sp->hdr.serviceId);
	354	if (old_id != conn->orig_service_id &&
	355	old_id != sp->hdr.serviceId)
	356	return rxrpc_protocol_error(skb, rxrpc_eproto_bad_upgrade);
	357	}
	358
	359	if (after(sp->hdr.serial, conn->hi_serial))
	360	conn->hi_serial = sp->hdr.serial;
	361
	362	/* It's a connection-level packet if the call number is 0. */
	363	if (sp->hdr.callNumber == 0)
	364	return rxrpc_input_conn_packet(conn, skb);
	365
	366	/* Deal with path MTU discovery probing. */
	367	if (sp->hdr.type == RXRPC_PACKET_TYPE_ACK &&
	368	conn->pmtud_probe &&
	369	after_eq(sp->ack.acked_serial, conn->pmtud_probe))
	370	rxrpc_input_probe_for_pmtud(conn, sp->ack.acked_serial, false);
	371
	372	/* Call-bound packets are routed by connection channel. */
	373	channel = sp->hdr.cid & RXRPC_CHANNELMASK;
	374	chan = &conn->channels[channel];
	375
	376	/* Ignore really old calls */
	377	if (sp->hdr.callNumber < chan->last_call)
	378	return just_discard;
	379
	380	if (sp->hdr.callNumber == chan->last_call) {
	381	if (chan->call \|\|
	382	sp->hdr.type == RXRPC_PACKET_TYPE_ABORT)
	383	return just_discard;
	384
	385	/* For the previous service call, if completed successfully, we
	386	* discard all further packets.
	387	*/
	388	if (rxrpc_conn_is_service(conn) &&
	389	chan->last_type == RXRPC_PACKET_TYPE_ACK)
	390	return just_discard;
	391
	392	/* But otherwise we need to retransmit the final packet from
	393	* data cached in the connection record.
	394	*/
	395	if (sp->hdr.type == RXRPC_PACKET_TYPE_DATA)
	396	trace_rxrpc_rx_data(chan->call_debug_id,
	397	sp->hdr.seq,
	398	sp->hdr.serial,
	399	sp->hdr.flags);
	400	rxrpc_conn_retransmit_call(conn, skb, channel);
	401	return just_discard;
	402	}
	403
	404	call = rxrpc_try_get_call(chan->call, rxrpc_call_get_input);
	405
	406	if (sp->hdr.callNumber > chan->call_id) {
	407	if (rxrpc_to_client(sp)) {
	408	rxrpc_put_call(call, rxrpc_call_put_input);
	409	return rxrpc_protocol_error(skb,
	410	rxrpc_eproto_unexpected_implicit_end);
	411	}
	412
	413	if (call) {
	414	rxrpc_implicit_end_call(call, skb);
	415	rxrpc_put_call(call, rxrpc_call_put_input);
	416	call = NULL;
	417	}
	418	}
	419
	420	if (!call) {
	421	if (rxrpc_to_client(sp))
	422	return rxrpc_protocol_error(skb, rxrpc_eproto_no_client_call);
	423	return rxrpc_new_incoming_call(conn->local, conn->peer, conn,
	424	peer_srx, skb);
	425	}
	426
	427	rxrpc_queue_rx_call_packet(call, skb);
	428	rxrpc_put_call(call, rxrpc_call_put_input);
	429	return true;
	430	}
	431
	432	/*
	433	* I/O and event handling thread.
	434	*/
	435	int rxrpc_io_thread(void *data)
	436	{
	437	struct rxrpc_connection *conn;
	438	struct sk_buff_head rx_queue;
	439	struct rxrpc_local *local = data;
	440	struct rxrpc_call *call;
	441	struct sk_buff *skb;
	442	#ifdef CONFIG_AF_RXRPC_INJECT_RX_DELAY
	443	ktime_t now;
	444	#endif
	445	bool should_stop;
	446	LIST_HEAD(conn_attend_q);
	447	LIST_HEAD(call_attend_q);
	448
	449	complete(&local->io_thread_ready);
	450
	451	skb_queue_head_init(&rx_queue);
	452
	453	set_user_nice(current, MIN_NICE);
	454
	455	for (;;) {
	456	rxrpc_inc_stat(local->rxnet, stat_io_loop);
	457
	458	/* Inject a delay into packets if requested. */
	459	#ifdef CONFIG_AF_RXRPC_INJECT_RX_DELAY
	460	now = ktime_get_real();
	461	while ((skb = skb_peek(&local->rx_delay_queue))) {
	462	if (ktime_before(now, skb->tstamp))
	463	break;
	464	skb = skb_dequeue(&local->rx_delay_queue);
	465	skb_queue_tail(&local->rx_queue, skb);
	466	}
	467	#endif
	468
	469	if (!skb_queue_empty(&local->rx_queue)) {
	470	spin_lock_irq(&local->rx_queue.lock);
	471	skb_queue_splice_tail_init(&local->rx_queue, &rx_queue);
	472	spin_unlock_irq(&local->rx_queue.lock);
	473	trace_rxrpc_iothread_rx(local, skb_queue_len(&rx_queue));
	474	}
	475
	476	/* Distribute packets and errors. */
	477	while ((skb = __skb_dequeue(&rx_queue))) {
	478	struct rxrpc_skb_priv *sp = rxrpc_skb(skb);
	479	switch (skb->mark) {
	480	case RXRPC_SKB_MARK_PACKET:
	481	skb->priority = 0;
	482	if (!rxrpc_input_packet(local, &skb))
	483	rxrpc_reject_packet(local, skb);
	484	trace_rxrpc_rx_done(skb->mark, skb->priority);
	485	rxrpc_free_skb(skb, rxrpc_skb_put_input);
	486	break;
	487	case RXRPC_SKB_MARK_ERROR:
	488	rxrpc_input_error(local, skb);
	489	rxrpc_free_skb(skb, rxrpc_skb_put_error_report);
	490	break;
	491	case RXRPC_SKB_MARK_SERVICE_CONN_SECURED:
	492	rxrpc_input_conn_event(sp->poke_conn, skb);
	493	rxrpc_put_connection(sp->poke_conn, rxrpc_conn_put_poke);
	494	rxrpc_free_skb(skb, rxrpc_skb_put_conn_secured);
	495	break;
	496	default:
	497	WARN_ON_ONCE(1);
	498	rxrpc_free_skb(skb, rxrpc_skb_put_unknown);
	499	break;
	500	}
	501	}
	502
	503	/* Deal with connections that want immediate attention. */
	504	if (!list_empty_careful(&local->conn_attend_q)) {
	505	spin_lock_irq(&local->lock);
	506	list_splice_tail_init(&local->conn_attend_q, &conn_attend_q);
	507	spin_unlock_irq(&local->lock);
	508	}
	509
	510	while ((conn = list_first_entry_or_null(&conn_attend_q,
	511	struct rxrpc_connection,
	512	attend_link))) {
	513	spin_lock_irq(&local->lock);
	514	list_del_init(&conn->attend_link);
	515	spin_unlock_irq(&local->lock);
	516	rxrpc_input_conn_event(conn, NULL);
	517	rxrpc_put_connection(conn, rxrpc_conn_put_poke);
	518	}
	519
	520	if (test_and_clear_bit(RXRPC_CLIENT_CONN_REAP_TIMER,
	521	&local->client_conn_flags))
	522	rxrpc_discard_expired_client_conns(local);
	523
	524	/* Deal with calls that want immediate attention. */
	525	spin_lock_irq(&local->lock);
	526	list_splice_tail_init(&local->call_attend_q, &call_attend_q);
	527	spin_unlock_irq(&local->lock);
	528
	529	while ((call = list_first_entry_or_null(&call_attend_q,
	530	struct rxrpc_call,
	531	attend_link))) {
	532	spin_lock_irq(&local->lock);
	533	list_del_init(&call->attend_link);
	534	spin_unlock_irq(&local->lock);
	535	trace_rxrpc_call_poked(call);
	536	rxrpc_input_call_event(call);
	537	rxrpc_put_call(call, rxrpc_call_put_poke);
	538	}
	539
	540	if (!list_empty(&local->new_client_calls))
	541	rxrpc_connect_client_calls(local);
	542
	543	set_current_state(TASK_INTERRUPTIBLE);
	544	should_stop = kthread_should_stop();
	545	if (!skb_queue_empty(&local->rx_queue) \|\|
	546	!list_empty(&local->call_attend_q) \|\|
	547	!list_empty(&local->conn_attend_q) \|\|
	548	!list_empty(&local->new_client_calls) \|\|
	549	test_bit(RXRPC_CLIENT_CONN_REAP_TIMER,
	550	&local->client_conn_flags)) {
	551	__set_current_state(TASK_RUNNING);
	552	continue;
	553	}
	554
	555	if (should_stop)
	556	break;
	557
	558	#ifdef CONFIG_AF_RXRPC_INJECT_RX_DELAY
	559	skb = skb_peek(&local->rx_delay_queue);
	560	if (skb) {
	561	unsigned long timeout;
	562	ktime_t tstamp = skb->tstamp;
	563	ktime_t now = ktime_get_real();
	564	s64 delay_ns = ktime_to_ns(ktime_sub(tstamp, now));
	565
	566	if (delay_ns <= 0) {
	567	__set_current_state(TASK_RUNNING);
	568	continue;
	569	}
	570
	571	timeout = nsecs_to_jiffies(delay_ns);
	572	timeout = umax(timeout, 1);
	573	schedule_timeout(timeout);
	574	__set_current_state(TASK_RUNNING);
	575	continue;
	576	}
	577	#endif
	578
	579	schedule();
	580	}
	581
	582	__set_current_state(TASK_RUNNING);
	583	rxrpc_see_local(local, rxrpc_local_stop);
	584	rxrpc_destroy_local(local);
	585	WRITE_ONCE(local->io_thread, NULL);
	586	rxrpc_see_local(local, rxrpc_local_stopped);
	587	return 0;
	588	}