ar-input.c

linux 内核源代码

		if (seq == 0)
			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)
			rxrpc_assume_implicit_ackall(call, serial);

		switch (rxrpc_fast_process_data(call, skb, seq)) {
		case 0:
			skb = NULL;
			goto done;

		default:
			BUG();

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

	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);
			rxrpc_queue_call(call);
			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;
		set_bit(RXRPC_CALL_ABORT, &call->events);
		rxrpc_queue_call(call);
	}
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();

		sp->hdr.seq	= htonl(ntohl(sp->hdr.seq) + 1);
		sp->hdr.serial	= htonl(ntohl(sp->hdr.serial) + 1);
		sp->hdr.flags	= jhdr.flags;
		sp->hdr._rsvd	= jhdr._rsvd;

		_proto("Rx DATA Jumbo %%%u", ntohl(sp->hdr.serial) - 1);

		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;
		set_bit(RXRPC_CALL_ABORT, &call->events);
		rxrpc_queue_call(call);
	}
	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
 */
static void rxrpc_post_packet_to_call(struct rxrpc_connection *conn,
				      struct sk_buff *skb)
{
	struct rxrpc_skb_priv *sp;
	struct rxrpc_call *call;
	struct rb_node *p;
	__be32 call_id;

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

	read_lock_bh(&conn->lock);

	sp = rxrpc_skb(skb);

	/* look at extant calls by channel number first */
	call = conn->channels[ntohl(sp->hdr.cid) & RXRPC_CHANNELMASK];
	if (!call || call->call_id != sp->hdr.callNumber)
		goto call_not_extant;

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

	read_lock(&call->state_lock);
	switch (call->state) {
	case RXRPC_CALL_LOCALLY_ABORTED:
		if (!test_and_set_bit(RXRPC_CALL_ABORT, &call->events))
			rxrpc_queue_call(call);
	case RXRPC_CALL_REMOTELY_ABORTED:
	case RXRPC_CALL_NETWORK_ERROR:
	case RXRPC_CALL_DEAD:
		goto free_unlock;
	default:
		break;
	}

	read_unlock(&call->state_lock);
	rxrpc_get_call(call);
	read_unlock_bh(&conn->lock);

	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;

call_not_extant:
	/* search the completed calls in case what we're dealing with is
	 * there */
	_debug("call not extant");

	call_id = sp->hdr.callNumber;
	p = conn->calls.rb_node;
	while (p) {
		call = rb_entry(p, struct rxrpc_call, conn_node);

		if (call_id < call->call_id)
			p = p->rb_left;
		else if (call_id > call->call_id)
			p = p->rb_right;
		else
			goto found_completed_call;
	}

dead_call:
	/* it's a either a really old call that we no longer remember or its a
	 * new incoming call */
	read_unlock_bh(&conn->lock);

	if (sp->hdr.flags & RXRPC_CLIENT_INITIATED &&
	    sp->hdr.seq == __constant_cpu_to_be32(1)) {
		_debug("incoming call");
		skb_queue_tail(&conn->trans->local->accept_queue, skb);
		rxrpc_queue_work(&conn->trans->local->acceptor);
		goto done;
	}

	_debug("dead call");
	skb->priority = RX_CALL_DEAD;
	rxrpc_reject_packet(conn->trans->local, skb);
	goto done;

	/* resend last packet of a completed call
	 * - client calls may have been aborted or ACK'd
	 * - server calls may have been aborted
	 */
found_completed_call:
	_debug("completed call");

	if (atomic_read(&call->usage) == 0)
		goto dead_call;

	/* synchronise any state changes */
	read_lock(&call->state_lock);
	ASSERTIFCMP(call->state != RXRPC_CALL_CLIENT_FINAL_ACK,
		    call->state, >=, RXRPC_CALL_COMPLETE);

	if (call->state == RXRPC_CALL_LOCALLY_ABORTED ||
	    call->state == RXRPC_CALL_REMOTELY_ABORTED ||
	    call->state == RXRPC_CALL_DEAD) {
		read_unlock(&call->state_lock);
		goto dead_call;
	}

	if (call->conn->in_clientflag) {
		read_unlock(&call->state_lock);
		goto dead_call; /* complete server call */
	}

	_debug("final ack again");
	rxrpc_get_call(call);
	set_bit(RXRPC_CALL_ACK_FINAL, &call->events);
	rxrpc_queue_call(call);

free_unlock:
	read_unlock(&call->state_lock);
	read_unlock_bh(&conn->lock);
	rxrpc_free_skb(skb);
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);
	rxrpc_queue_conn(conn);
}

/*
 * handle data received on the local endpoint
 * - may be called in interrupt context
 */
void rxrpc_data_ready(struct sock *sk, int count)
{
	struct rxrpc_connection *conn;
	struct rxrpc_transport *trans;
	struct rxrpc_skb_priv *sp;
	struct rxrpc_local *local;
	struct rxrpc_peer *peer;
	struct sk_buff *skb;
	int ret;

	_enter("%p, %d", sk, count);

	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);
		_leave(" [CSUM failed]");
		return;
	}

	/* the socket buffer we have is owned by UDP, with UDP's data all over
	 * it, but we really want our own */
	skb_orphan(skb);
	sp = rxrpc_skb(skb);
	memset(sp, 0, sizeof(*sp));

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

	/* dig out the RxRPC connection details */
	if (skb_copy_bits(skb, sizeof(struct udphdr), &sp->hdr,
			  sizeof(sp->hdr)) < 0)
		goto bad_message;
	if (!pskb_pull(skb, sizeof(struct udphdr) + sizeof(sp->hdr)))
		BUG();

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

	if (sp->hdr.type == 0 || sp->hdr.type >= RXRPC_N_PACKET_TYPES) {
		_proto("Rx Bad Packet Type %u", sp->hdr.type);
		goto bad_message;
	}

	if (sp->hdr.type == RXRPC_PACKET_TYPE_DATA &&
	    (sp->hdr.callNumber == 0 || sp->hdr.seq == 0))
		goto bad_message;

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

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

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

	_debug("CONN %p {%d}", conn, conn->debug_id);

	if (sp->hdr.callNumber == 0)
		rxrpc_post_packet_to_conn(conn, skb);
	else
		rxrpc_post_packet_to_call(conn, skb);
	rxrpc_put_connection(conn);
	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) {
		if (sp->hdr.seq == __constant_cpu_to_be32(1)) {
			_debug("first packet");
			skb_queue_tail(&local->accept_queue, skb);
			rxrpc_queue_work(&local->acceptor);
			rxrpc_put_local(local);
			_leave(" [incoming]");
			return;
		}
		skb->priority = RX_INVALID_OPERATION;
	} else {
		skb->priority = RX_CALL_DEAD;
	}

	_debug("reject");
	rxrpc_reject_packet(local, skb);
	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]");
}