ar-input.c 19.9 KB
Newer Older
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19
/* RxRPC packet reception
 *
 * Copyright (C) 2007 Red Hat, Inc. All Rights Reserved.
 * Written by David Howells (dhowells@redhat.com)
 *
 * This program is free software; you can redistribute it and/or
 * modify it under the terms of the GNU General Public License
 * as published by the Free Software Foundation; either version
 * 2 of the License, or (at your option) any later version.
 */

#include <linux/module.h>
#include <linux/net.h>
#include <linux/skbuff.h>
#include <linux/errqueue.h>
#include <linux/udp.h>
#include <linux/in.h>
#include <linux/in6.h>
#include <linux/icmp.h>
20
#include <linux/gfp.h>
21 22 23
#include <net/sock.h>
#include <net/af_rxrpc.h>
#include <net/ip.h>
24
#include <net/udp.h>
25
#include <net/net_namespace.h>
26 27 28 29 30
#include "ar-internal.h"

const char *rxrpc_pkts[] = {
	"?00",
	"DATA", "ACK", "BUSY", "ABORT", "ACKALL", "CHALL", "RESP", "DEBUG",
31
	"?09", "?10", "?11", "?12", "VERSION", "?14", "?15"
32 33 34 35 36 37 38 39 40 41 42 43 44 45
};

/*
 * queue a packet for recvmsg to pass to userspace
 * - the caller must hold a lock on call->lock
 * - must not be called with interrupts disabled (sk_filter() disables BH's)
 * - eats the packet whether successful or not
 * - there must be just one reference to the packet, which the caller passes to
 *   this function
 */
int rxrpc_queue_rcv_skb(struct rxrpc_call *call, struct sk_buff *skb,
			bool force, bool terminal)
{
	struct rxrpc_skb_priv *sp;
46
	struct rxrpc_sock *rx = call->socket;
47
	struct sock *sk;
48
	int ret;
49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68

	_enter(",,%d,%d", force, terminal);

	ASSERT(!irqs_disabled());

	sp = rxrpc_skb(skb);
	ASSERTCMP(sp->call, ==, call);

	/* if we've already posted the terminal message for a call, then we
	 * don't post any more */
	if (test_bit(RXRPC_CALL_TERMINAL_MSG, &call->flags)) {
		_debug("already terminated");
		ASSERTCMP(call->state, >=, RXRPC_CALL_COMPLETE);
		skb->destructor = NULL;
		sp->call = NULL;
		rxrpc_put_call(call);
		rxrpc_free_skb(skb);
		return 0;
	}

69
	sk = &rx->sk;
70 71 72 73 74 75 76

	if (!force) {
		/* cast skb->rcvbuf to unsigned...  It's pointless, but
		 * reduces number of warnings when compiling with -W
		 * --ANK */
//		ret = -ENOBUFS;
//		if (atomic_read(&sk->sk_rmem_alloc) + skb->truesize >=
77
//		    (unsigned int) sk->sk_rcvbuf)
78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98
//			goto out;

		ret = sk_filter(sk, skb);
		if (ret < 0)
			goto out;
	}

	spin_lock_bh(&sk->sk_receive_queue.lock);
	if (!test_bit(RXRPC_CALL_TERMINAL_MSG, &call->flags) &&
	    !test_bit(RXRPC_CALL_RELEASED, &call->flags) &&
	    call->socket->sk.sk_state != RXRPC_CLOSE) {
		skb->destructor = rxrpc_packet_destructor;
		skb->dev = NULL;
		skb->sk = sk;
		atomic_add(skb->truesize, &sk->sk_rmem_alloc);

		if (terminal) {
			_debug("<<<< TERMINAL MESSAGE >>>>");
			set_bit(RXRPC_CALL_TERMINAL_MSG, &call->flags);
		}

99 100 101 102 103 104 105 106 107 108
		/* allow interception by a kernel service */
		if (rx->interceptor) {
			rx->interceptor(sk, call->user_call_ID, skb);
			spin_unlock_bh(&sk->sk_receive_queue.lock);
		} else {
			_net("post skb %p", skb);
			__skb_queue_tail(&sk->sk_receive_queue, skb);
			spin_unlock_bh(&sk->sk_receive_queue.lock);

			if (!sock_flag(sk, SOCK_DEAD))
109
				sk->sk_data_ready(sk);
110
		}
111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164
		skb = NULL;
	} else {
		spin_unlock_bh(&sk->sk_receive_queue.lock);
	}
	ret = 0;

out:
	/* release the socket buffer */
	if (skb) {
		skb->destructor = NULL;
		sp->call = NULL;
		rxrpc_put_call(call);
		rxrpc_free_skb(skb);
	}

	_leave(" = %d", ret);
	return ret;
}

/*
 * process a DATA packet, posting the packet to the appropriate queue
 * - eats the packet if successful
 */
static int rxrpc_fast_process_data(struct rxrpc_call *call,
				   struct sk_buff *skb, u32 seq)
{
	struct rxrpc_skb_priv *sp;
	bool terminal;
	int ret, ackbit, ack;

	_enter("{%u,%u},,{%u}", call->rx_data_post, call->rx_first_oos, seq);

	sp = rxrpc_skb(skb);
	ASSERTCMP(sp->call, ==, NULL);

	spin_lock(&call->lock);

	if (call->state > RXRPC_CALL_COMPLETE)
		goto discard;

	ASSERTCMP(call->rx_data_expect, >=, call->rx_data_post);
	ASSERTCMP(call->rx_data_post, >=, call->rx_data_recv);
	ASSERTCMP(call->rx_data_recv, >=, call->rx_data_eaten);

	if (seq < call->rx_data_post) {
		_debug("dup #%u [-%u]", seq, call->rx_data_post);
		ack = RXRPC_ACK_DUPLICATE;
		ret = -ENOBUFS;
		goto discard_and_ack;
	}

	/* we may already have the packet in the out of sequence queue */
	ackbit = seq - (call->rx_data_eaten + 1);
	ASSERTCMP(ackbit, >=, 0);
165
	if (__test_and_set_bit(ackbit, call->ackr_window)) {
166 167 168 169 170 171 172 173
		_debug("dup oos #%u [%u,%u]",
		       seq, call->rx_data_eaten, call->rx_data_post);
		ack = RXRPC_ACK_DUPLICATE;
		goto discard_and_ack;
	}

	if (seq >= call->ackr_win_top) {
		_debug("exceed #%u [%u]", seq, call->ackr_win_top);
174
		__clear_bit(ackbit, call->ackr_window);
175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 209 210 211
		ack = RXRPC_ACK_EXCEEDS_WINDOW;
		goto discard_and_ack;
	}

	if (seq == call->rx_data_expect) {
		clear_bit(RXRPC_CALL_EXPECT_OOS, &call->flags);
		call->rx_data_expect++;
	} else if (seq > call->rx_data_expect) {
		_debug("oos #%u [%u]", seq, call->rx_data_expect);
		call->rx_data_expect = seq + 1;
		if (test_and_set_bit(RXRPC_CALL_EXPECT_OOS, &call->flags)) {
			ack = RXRPC_ACK_OUT_OF_SEQUENCE;
			goto enqueue_and_ack;
		}
		goto enqueue_packet;
	}

	if (seq != call->rx_data_post) {
		_debug("ahead #%u [%u]", seq, call->rx_data_post);
		goto enqueue_packet;
	}

	if (test_bit(RXRPC_CALL_RCVD_LAST, &call->flags))
		goto protocol_error;

	/* if the packet need security things doing to it, then it goes down
	 * the slow path */
	if (call->conn->security)
		goto enqueue_packet;

	sp->call = call;
	rxrpc_get_call(call);
	terminal = ((sp->hdr.flags & RXRPC_LAST_PACKET) &&
		    !(sp->hdr.flags & RXRPC_CLIENT_INITIATED));
	ret = rxrpc_queue_rcv_skb(call, skb, false, terminal);
	if (ret < 0) {
		if (ret == -ENOMEM || ret == -ENOBUFS) {
212
			__clear_bit(ackbit, call->ackr_window);
213 214 215 216 217 218 219 220 221 222 223 224 225 226 227 228 229 230 231 232 233
			ack = RXRPC_ACK_NOSPACE;
			goto discard_and_ack;
		}
		goto out;
	}

	skb = NULL;

	_debug("post #%u", seq);
	ASSERTCMP(call->rx_data_post, ==, seq);
	call->rx_data_post++;

	if (sp->hdr.flags & RXRPC_LAST_PACKET)
		set_bit(RXRPC_CALL_RCVD_LAST, &call->flags);

	/* if we've reached an out of sequence packet then we need to drain
	 * that queue into the socket Rx queue now */
	if (call->rx_data_post == call->rx_first_oos) {
		_debug("drain rx oos now");
		read_lock(&call->state_lock);
		if (call->state < RXRPC_CALL_COMPLETE &&
234
		    !test_and_set_bit(RXRPC_CALL_EV_DRAIN_RX_OOS, &call->events))
235
			rxrpc_queue_call(call);
236 237 238 239 240 241 242 243 244 245 246 247 248 249 250 251 252 253 254 255 256 257 258 259 260 261 262 263 264 265 266 267 268 269
		read_unlock(&call->state_lock);
	}

	spin_unlock(&call->lock);
	atomic_inc(&call->ackr_not_idle);
	rxrpc_propose_ACK(call, RXRPC_ACK_DELAY, sp->hdr.serial, false);
	_leave(" = 0 [posted]");
	return 0;

protocol_error:
	ret = -EBADMSG;
out:
	spin_unlock(&call->lock);
	_leave(" = %d", ret);
	return ret;

discard_and_ack:
	_debug("discard and ACK packet %p", skb);
	__rxrpc_propose_ACK(call, ack, sp->hdr.serial, true);
discard:
	spin_unlock(&call->lock);
	rxrpc_free_skb(skb);
	_leave(" = 0 [discarded]");
	return 0;

enqueue_and_ack:
	__rxrpc_propose_ACK(call, ack, sp->hdr.serial, true);
enqueue_packet:
	_net("defer skb %p", skb);
	spin_unlock(&call->lock);
	skb_queue_tail(&call->rx_queue, skb);
	atomic_inc(&call->ackr_not_idle);
	read_lock(&call->state_lock);
	if (call->state < RXRPC_CALL_DEAD)
270
		rxrpc_queue_call(call);
271 272 273 274 275 276 277 278 279 280 281 282 283 284 285 286 287 288 289
	read_unlock(&call->state_lock);
	_leave(" = 0 [queued]");
	return 0;
}

/*
 * assume an implicit ACKALL of the transmission phase of a client socket upon
 * reception of the first reply packet
 */
static void rxrpc_assume_implicit_ackall(struct rxrpc_call *call, u32 serial)
{
	write_lock_bh(&call->state_lock);

	switch (call->state) {
	case RXRPC_CALL_CLIENT_AWAIT_REPLY:
		call->state = RXRPC_CALL_CLIENT_RECV_REPLY;
		call->acks_latest = serial;

		_debug("implicit ACKALL %%%u", call->acks_latest);
290
		set_bit(RXRPC_CALL_EV_RCVD_ACKALL, &call->events);
291 292 293
		write_unlock_bh(&call->state_lock);

		if (try_to_del_timer_sync(&call->resend_timer) >= 0) {
294 295
			clear_bit(RXRPC_CALL_EV_RESEND_TIMER, &call->events);
			clear_bit(RXRPC_CALL_EV_RESEND, &call->events);
296 297 298 299 300 301 302 303 304 305 306 307 308 309 310 311 312
			clear_bit(RXRPC_CALL_RUN_RTIMER, &call->flags);
		}
		break;

	default:
		write_unlock_bh(&call->state_lock);
		break;
	}
}

/*
 * post an incoming packet to the nominated call to deal with
 * - must get rid of the sk_buff, either by freeing it or by queuing it
 */
void rxrpc_fast_process_packet(struct rxrpc_call *call, struct sk_buff *skb)
{
	struct rxrpc_skb_priv *sp = rxrpc_skb(skb);
313 314
	__be32 wtmp;
	u32 hi_serial, abort_code;
315 316 317 318 319 320 321 322 323 324 325 326 327 328 329 330 331 332 333

	_enter("%p,%p", call, skb);

	ASSERT(!irqs_disabled());

#if 0 // INJECT RX ERROR
	if (sp->hdr.type == RXRPC_PACKET_TYPE_DATA) {
		static int skip = 0;
		if (++skip == 3) {
			printk("DROPPED 3RD PACKET!!!!!!!!!!!!!\n");
			skip = 0;
			goto free_packet;
		}
	}
#endif

	/* track the latest serial number on this connection for ACK packet
	 * information */
	hi_serial = atomic_read(&call->conn->hi_serial);
334
	while (sp->hdr.serial > hi_serial)
335
		hi_serial = atomic_cmpxchg(&call->conn->hi_serial, hi_serial,
336
					   sp->hdr.serial);
337 338 339 340

	/* request ACK generation for any ACK or DATA packet that requests
	 * it */
	if (sp->hdr.flags & RXRPC_REQUEST_ACK) {
341
		_proto("ACK Requested on %%%u", sp->hdr.serial);
342
		rxrpc_propose_ACK(call, RXRPC_ACK_REQUESTED, sp->hdr.serial, false);
343 344 345 346 347 348
	}

	switch (sp->hdr.type) {
	case RXRPC_PACKET_TYPE_ABORT:
		_debug("abort");

349
		if (skb_copy_bits(skb, 0, &wtmp, sizeof(wtmp)) < 0)
350 351
			goto protocol_error;

352 353
		abort_code = ntohl(wtmp);
		_proto("Rx ABORT %%%u { %x }", sp->hdr.serial, abort_code);
354 355 356 357 358

		write_lock_bh(&call->state_lock);
		if (call->state < RXRPC_CALL_COMPLETE) {
			call->state = RXRPC_CALL_REMOTELY_ABORTED;
			call->abort_code = abort_code;
359
			set_bit(RXRPC_CALL_EV_RCVD_ABORT, &call->events);
360
			rxrpc_queue_call(call);
361 362 363 364
		}
		goto free_packet_unlock;

	case RXRPC_PACKET_TYPE_BUSY:
365
		_proto("Rx BUSY %%%u", sp->hdr.serial);
366 367 368 369 370 371 372 373

		if (call->conn->out_clientflag)
			goto protocol_error;

		write_lock_bh(&call->state_lock);
		switch (call->state) {
		case RXRPC_CALL_CLIENT_SEND_REQUEST:
			call->state = RXRPC_CALL_SERVER_BUSY;
374
			set_bit(RXRPC_CALL_EV_RCVD_BUSY, &call->events);
375
			rxrpc_queue_call(call);
376 377 378 379 380 381 382
		case RXRPC_CALL_SERVER_BUSY:
			goto free_packet_unlock;
		default:
			goto protocol_error_locked;
		}

	default:
383
		_proto("Rx %s %%%u", rxrpc_pkts[sp->hdr.type], sp->hdr.serial);
384 385 386
		goto protocol_error;

	case RXRPC_PACKET_TYPE_DATA:
387
		_proto("Rx DATA %%%u { #%u }", sp->hdr.serial, sp->hdr.seq);
388

389
		if (sp->hdr.seq == 0)
390 391 392 393 394 395 396
			goto protocol_error;

		call->ackr_prev_seq = sp->hdr.seq;

		/* received data implicitly ACKs all of the request packets we
		 * sent when we're acting as a client */
		if (call->state == RXRPC_CALL_CLIENT_AWAIT_REPLY)
397
			rxrpc_assume_implicit_ackall(call, sp->hdr.serial);
398

399
		switch (rxrpc_fast_process_data(call, skb, sp->hdr.seq)) {
400 401 402 403 404 405 406 407 408 409 410 411
		case 0:
			skb = NULL;
			goto done;

		default:
			BUG();

			/* data packet received beyond the last packet */
		case -EBADMSG:
			goto protocol_error;
		}

412
	case RXRPC_PACKET_TYPE_ACKALL:
413 414 415 416 417
	case RXRPC_PACKET_TYPE_ACK:
		/* ACK processing is done in process context */
		read_lock_bh(&call->state_lock);
		if (call->state < RXRPC_CALL_DEAD) {
			skb_queue_tail(&call->rx_queue, skb);
418
			rxrpc_queue_call(call);
419 420 421 422 423 424 425 426 427 428 429 430 431
			skb = NULL;
		}
		read_unlock_bh(&call->state_lock);
		goto free_packet;
	}

protocol_error:
	_debug("protocol error");
	write_lock_bh(&call->state_lock);
protocol_error_locked:
	if (call->state <= RXRPC_CALL_COMPLETE) {
		call->state = RXRPC_CALL_LOCALLY_ABORTED;
		call->abort_code = RX_PROTOCOL_ERROR;
432
		set_bit(RXRPC_CALL_EV_ABORT, &call->events);
433
		rxrpc_queue_call(call);
434 435 436 437 438 439 440 441 442 443 444 445 446 447 448 449 450 451 452 453 454 455 456 457 458 459 460 461 462 463 464 465 466 467 468 469 470 471 472 473 474 475 476 477 478 479
	}
free_packet_unlock:
	write_unlock_bh(&call->state_lock);
free_packet:
	rxrpc_free_skb(skb);
done:
	_leave("");
}

/*
 * split up a jumbo data packet
 */
static void rxrpc_process_jumbo_packet(struct rxrpc_call *call,
				       struct sk_buff *jumbo)
{
	struct rxrpc_jumbo_header jhdr;
	struct rxrpc_skb_priv *sp;
	struct sk_buff *part;

	_enter(",{%u,%u}", jumbo->data_len, jumbo->len);

	sp = rxrpc_skb(jumbo);

	do {
		sp->hdr.flags &= ~RXRPC_JUMBO_PACKET;

		/* make a clone to represent the first subpacket in what's left
		 * of the jumbo packet */
		part = skb_clone(jumbo, GFP_ATOMIC);
		if (!part) {
			/* simply ditch the tail in the event of ENOMEM */
			pskb_trim(jumbo, RXRPC_JUMBO_DATALEN);
			break;
		}
		rxrpc_new_skb(part);

		pskb_trim(part, RXRPC_JUMBO_DATALEN);

		if (!pskb_pull(jumbo, RXRPC_JUMBO_DATALEN))
			goto protocol_error;

		if (skb_copy_bits(jumbo, 0, &jhdr, sizeof(jhdr)) < 0)
			goto protocol_error;
		if (!pskb_pull(jumbo, sizeof(jhdr)))
			BUG();

480 481
		sp->hdr.seq	+= 1;
		sp->hdr.serial	+= 1;
482 483 484
		sp->hdr.flags	= jhdr.flags;
		sp->hdr._rsvd	= jhdr._rsvd;

485
		_proto("Rx DATA Jumbo %%%u", sp->hdr.serial - 1);
486 487 488 489 490 491 492 493 494 495 496 497 498 499 500 501 502 503

		rxrpc_fast_process_packet(call, part);
		part = NULL;

	} while (sp->hdr.flags & RXRPC_JUMBO_PACKET);

	rxrpc_fast_process_packet(call, jumbo);
	_leave("");
	return;

protocol_error:
	_debug("protocol error");
	rxrpc_free_skb(part);
	rxrpc_free_skb(jumbo);
	write_lock_bh(&call->state_lock);
	if (call->state <= RXRPC_CALL_COMPLETE) {
		call->state = RXRPC_CALL_LOCALLY_ABORTED;
		call->abort_code = RX_PROTOCOL_ERROR;
504
		set_bit(RXRPC_CALL_EV_ABORT, &call->events);
505
		rxrpc_queue_call(call);
506 507 508 509 510 511 512 513 514
	}
	write_unlock_bh(&call->state_lock);
	_leave("");
}

/*
 * post an incoming packet to the appropriate call/socket to deal with
 * - must get rid of the sk_buff, either by freeing it or by queuing it
 */
515
static void rxrpc_post_packet_to_call(struct rxrpc_call *call,
516 517 518 519
				      struct sk_buff *skb)
{
	struct rxrpc_skb_priv *sp;

520
	_enter("%p,%p", call, skb);
521 522 523 524 525 526 527 528

	sp = rxrpc_skb(skb);

	_debug("extant call [%d]", call->state);

	read_lock(&call->state_lock);
	switch (call->state) {
	case RXRPC_CALL_LOCALLY_ABORTED:
529
		if (!test_and_set_bit(RXRPC_CALL_EV_ABORT, &call->events)) {
530
			rxrpc_queue_call(call);
531 532
			goto free_unlock;
		}
533 534 535
	case RXRPC_CALL_REMOTELY_ABORTED:
	case RXRPC_CALL_NETWORK_ERROR:
	case RXRPC_CALL_DEAD:
536 537 538 539 540 541 542 543 544
		goto dead_call;
	case RXRPC_CALL_COMPLETE:
	case RXRPC_CALL_CLIENT_FINAL_ACK:
		/* complete server call */
		if (call->conn->in_clientflag)
			goto dead_call;
		/* resend last packet of a completed call */
		_debug("final ack again");
		rxrpc_get_call(call);
545
		set_bit(RXRPC_CALL_EV_ACK_FINAL, &call->events);
546
		rxrpc_queue_call(call);
547 548 549 550 551 552 553 554 555 556 557 558 559 560 561 562 563 564
		goto free_unlock;
	default:
		break;
	}

	read_unlock(&call->state_lock);
	rxrpc_get_call(call);

	if (sp->hdr.type == RXRPC_PACKET_TYPE_DATA &&
	    sp->hdr.flags & RXRPC_JUMBO_PACKET)
		rxrpc_process_jumbo_packet(call, skb);
	else
		rxrpc_fast_process_packet(call, skb);

	rxrpc_put_call(call);
	goto done;

dead_call:
565 566
	if (sp->hdr.type != RXRPC_PACKET_TYPE_ABORT) {
		skb->priority = RX_CALL_DEAD;
567 568
		rxrpc_reject_packet(call->conn->trans->local, skb);
		goto unlock;
569 570 571
	}
free_unlock:
	rxrpc_free_skb(skb);
572 573
unlock:
	read_unlock(&call->state_lock);
574 575 576 577 578 579 580 581 582 583 584 585 586 587 588
done:
	_leave("");
}

/*
 * post connection-level events to the connection
 * - this includes challenges, responses and some aborts
 */
static void rxrpc_post_packet_to_conn(struct rxrpc_connection *conn,
				      struct sk_buff *skb)
{
	_enter("%p,%p", conn, skb);

	atomic_inc(&conn->usage);
	skb_queue_tail(&conn->rx_queue, skb);
589
	rxrpc_queue_conn(conn);
590 591
}

592 593 594 595 596 597 598 599 600 601 602 603 604 605
/*
 * post endpoint-level events to the local endpoint
 * - this includes debug and version messages
 */
static void rxrpc_post_packet_to_local(struct rxrpc_local *local,
				       struct sk_buff *skb)
{
	_enter("%p,%p", local, skb);

	atomic_inc(&local->usage);
	skb_queue_tail(&local->event_queue, skb);
	rxrpc_queue_work(&local->event_processor);
}

606 607 608 609 610 611 612 613 614 615 616 617 618 619 620 621 622 623 624 625 626 627 628 629 630 631 632 633 634
/*
 * Extract the wire header from a packet and translate the byte order.
 */
static noinline
int rxrpc_extract_header(struct rxrpc_skb_priv *sp, struct sk_buff *skb)
{
	struct rxrpc_wire_header whdr;

	/* dig out the RxRPC connection details */
	if (skb_copy_bits(skb, sizeof(struct udphdr), &whdr, sizeof(whdr)) < 0)
		return -EBADMSG;
	if (!pskb_pull(skb, sizeof(struct udphdr) + sizeof(whdr)))
		BUG();

	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);
	return 0;
}

635 636 637 638 639 640 641 642 643 644 645 646 647 648 649 650 651 652 653 654 655 656 657 658 659 660 661 662
static struct rxrpc_connection *rxrpc_conn_from_local(struct rxrpc_local *local,
					       struct sk_buff *skb,
					       struct rxrpc_skb_priv *sp)
{
	struct rxrpc_peer *peer;
	struct rxrpc_transport *trans;
	struct rxrpc_connection *conn;

	peer = rxrpc_find_peer(local, ip_hdr(skb)->saddr,
				udp_hdr(skb)->source);
	if (IS_ERR(peer))
		goto cant_find_conn;

	trans = rxrpc_find_transport(local, peer);
	rxrpc_put_peer(peer);
	if (!trans)
		goto cant_find_conn;

	conn = rxrpc_find_connection(trans, &sp->hdr);
	rxrpc_put_transport(trans);
	if (!conn)
		goto cant_find_conn;

	return conn;
cant_find_conn:
	return NULL;
}

663 664 665 666
/*
 * handle data received on the local endpoint
 * - may be called in interrupt context
 */
667
void rxrpc_data_ready(struct sock *sk)
668 669 670 671 672 673
{
	struct rxrpc_skb_priv *sp;
	struct rxrpc_local *local;
	struct sk_buff *skb;
	int ret;

674
	_enter("%p", sk);
675 676 677 678 679 680 681 682 683 684 685 686 687 688 689 690 691 692 693 694 695 696 697 698 699 700 701 702 703 704 705 706

	ASSERT(!irqs_disabled());

	read_lock_bh(&rxrpc_local_lock);
	local = sk->sk_user_data;
	if (local && atomic_read(&local->usage) > 0)
		rxrpc_get_local(local);
	else
		local = NULL;
	read_unlock_bh(&rxrpc_local_lock);
	if (!local) {
		_leave(" [local dead]");
		return;
	}

	skb = skb_recv_datagram(sk, 0, 1, &ret);
	if (!skb) {
		rxrpc_put_local(local);
		if (ret == -EAGAIN)
			return;
		_debug("UDP socket error %d", ret);
		return;
	}

	rxrpc_new_skb(skb);

	_net("recv skb %p", skb);

	/* we'll probably need to checksum it (didn't call sock_recvmsg) */
	if (skb_checksum_complete(skb)) {
		rxrpc_free_skb(skb);
		rxrpc_put_local(local);
707
		UDP_INC_STATS_BH(&init_net, UDP_MIB_INERRORS, 0);
708 709 710 711
		_leave(" [CSUM failed]");
		return;
	}

712
	UDP_INC_STATS_BH(&init_net, UDP_MIB_INDATAGRAMS, 0);
713

714 715 716
	/* The socket buffer we have is owned by UDP, with UDP's data all over
	 * it, but we really want our own data there.
	 */
717 718 719 720 721 722 723
	skb_orphan(skb);
	sp = rxrpc_skb(skb);

	_net("Rx UDP packet from %08x:%04hu",
	     ntohl(ip_hdr(skb)->saddr), ntohs(udp_hdr(skb)->source));

	/* dig out the RxRPC connection details */
724
	if (rxrpc_extract_header(sp, skb) < 0)
725 726 727 728
		goto bad_message;

	_net("Rx RxRPC %s ep=%x call=%x:%x",
	     sp->hdr.flags & RXRPC_CLIENT_INITIATED ? "ToServer" : "ToClient",
729
	     sp->hdr.epoch, sp->hdr.cid, sp->hdr.callNumber);
730

731 732
	if (sp->hdr.type >= RXRPC_N_PACKET_TYPES ||
	    !((RXRPC_SUPPORTED_PACKET_TYPES >> sp->hdr.type) & 1)) {
733 734 735 736
		_proto("Rx Bad Packet Type %u", sp->hdr.type);
		goto bad_message;
	}

737 738 739 740 741
	if (sp->hdr.type == RXRPC_PACKET_TYPE_VERSION) {
		rxrpc_post_packet_to_local(local, skb);
		goto out;
	}
	
742 743 744 745
	if (sp->hdr.type == RXRPC_PACKET_TYPE_DATA &&
	    (sp->hdr.callNumber == 0 || sp->hdr.seq == 0))
		goto bad_message;

746 747 748 749 750
	if (sp->hdr.callNumber == 0) {
		/* This is a connection-level packet. These should be
		 * fairly rare, so the extra overhead of looking them up the
		 * old-fashioned way doesn't really hurt */
		struct rxrpc_connection *conn;
751

752 753 754
		conn = rxrpc_conn_from_local(local, skb, sp);
		if (!conn)
			goto cant_route_call;
755

756
		_debug("CONN %p {%d}", conn, conn->debug_id);
757
		rxrpc_post_packet_to_conn(conn, skb);
758 759 760
		rxrpc_put_connection(conn);
	} else {
		struct rxrpc_call *call;
761 762 763

		call = rxrpc_find_call_hash(&sp->hdr, local,
					    AF_INET, &ip_hdr(skb)->saddr);
764 765 766 767 768
		if (call)
			rxrpc_post_packet_to_call(call, skb);
		else
			goto cant_route_call;
	}
769 770

out:
771 772 773 774 775 776 777
	rxrpc_put_local(local);
	return;

cant_route_call:
	_debug("can't route call");
	if (sp->hdr.flags & RXRPC_CLIENT_INITIATED &&
	    sp->hdr.type == RXRPC_PACKET_TYPE_DATA) {
778
		if (sp->hdr.seq == 1) {
779 780
			_debug("first packet");
			skb_queue_tail(&local->accept_queue, skb);
781
			rxrpc_queue_work(&local->acceptor);
782 783 784 785 786 787 788 789 790
			rxrpc_put_local(local);
			_leave(" [incoming]");
			return;
		}
		skb->priority = RX_INVALID_OPERATION;
	} else {
		skb->priority = RX_CALL_DEAD;
	}

791 792 793 794
	if (sp->hdr.type != RXRPC_PACKET_TYPE_ABORT) {
		_debug("reject type %d",sp->hdr.type);
		rxrpc_reject_packet(local, skb);
	}
795 796 797 798 799 800 801 802 803 804
	rxrpc_put_local(local);
	_leave(" [no call]");
	return;

bad_message:
	skb->priority = RX_PROTOCOL_ERROR;
	rxrpc_reject_packet(local, skb);
	rxrpc_put_local(local);
	_leave(" [badmsg]");
}