net/rxrpc/ar-input.c

   1 /* RxRPC packet reception
   2  *
   3  * Copyright (C) 2007 Red Hat, Inc. All Rights Reserved.
   4  * Written by David Howells (dhowells@redhat.com)
   5  *
   6  * This program is free software; you can redistribute it and/or
   7  * modify it under the terms of the GNU General Public License
   8  * as published by the Free Software Foundation; either version
   9  * 2 of the License, or (at your option) any later version.
  10  */
  11
  12 #include <linux/module.h>
  13 #include <linux/net.h>
  14 #include <linux/skbuff.h>
  15 #include <linux/errqueue.h>
  16 #include <linux/udp.h>
  17 #include <linux/in.h>
  18 #include <linux/in6.h>
  19 #include <linux/icmp.h>
  20 #include <linux/gfp.h>
  21 #include <net/sock.h>
  22 #include <net/af_rxrpc.h>
  23 #include <net/ip.h>
  24 #include <net/udp.h>
  25 #include <net/net_namespace.h>
  26 #include "ar-internal.h"
  27
  28 unsigned long rxrpc_ack_timeout = 1;
  29
  30 const char *rxrpc_pkts[] = {
  31         "?00",
  32         "DATA", "ACK", "BUSY", "ABORT", "ACKALL", "CHALL", "RESP", "DEBUG",
  33         "?09", "?10", "?11", "?12", "?13", "?14", "?15"
  34 };
  35
  36 /*
  37  * queue a packet for recvmsg to pass to userspace
  38  * - the caller must hold a lock on call->lock
  39  * - must not be called with interrupts disabled (sk_filter() disables BH's)
  40  * - eats the packet whether successful or not
  41  * - there must be just one reference to the packet, which the caller passes to
  42  *   this function
  43  */
  44 int rxrpc_queue_rcv_skb(struct rxrpc_call *call, struct sk_buff *skb,
  45                         bool force, bool terminal)
  46 {
  47         struct rxrpc_skb_priv *sp;
  48         struct rxrpc_sock *rx = call->socket;
  49         struct sock *sk;
  50         int skb_len, ret;
  51
  52         _enter(",,%d,%d", force, terminal);
  53
  54         ASSERT(!irqs_disabled());
  55
  56         sp = rxrpc_skb(skb);
  57         ASSERTCMP(sp->call, ==, call);
  58
  59         /* if we've already posted the terminal message for a call, then we
  60          * don't post any more */
  61         if (test_bit(RXRPC_CALL_TERMINAL_MSG, &call->flags)) {
  62                 _debug("already terminated");
  63                 ASSERTCMP(call->state, >=, RXRPC_CALL_COMPLETE);
  64                 skb->destructor = NULL;
  65                 sp->call = NULL;
  66                 rxrpc_put_call(call);
  67                 rxrpc_free_skb(skb);
  68                 return 0;
  69         }
  70
  71         sk = &rx->sk;
  72
  73         if (!force) {
  74                 /* cast skb->rcvbuf to unsigned...  It's pointless, but
  75                  * reduces number of warnings when compiling with -W
  76                  * --ANK */
  77 //              ret = -ENOBUFS;
  78 //              if (atomic_read(&sk->sk_rmem_alloc) + skb->truesize >=
  79 //                  (unsigned) sk->sk_rcvbuf)
  80 //                      goto out;
  81
  82                 ret = sk_filter(sk, skb);
  83                 if (ret < 0)
  84                         goto out;
  85         }
  86
  87         spin_lock_bh(&sk->sk_receive_queue.lock);
  88         if (!test_bit(RXRPC_CALL_TERMINAL_MSG, &call->flags) &&
  89             !test_bit(RXRPC_CALL_RELEASED, &call->flags) &&
  90             call->socket->sk.sk_state != RXRPC_CLOSE) {
  91                 skb->destructor = rxrpc_packet_destructor;
  92                 skb->dev = NULL;
  93                 skb->sk = sk;
  94                 atomic_add(skb->truesize, &sk->sk_rmem_alloc);
  95
  96                 if (terminal) {
  97                         _debug("<<<< TERMINAL MESSAGE >>>>");
  98                         set_bit(RXRPC_CALL_TERMINAL_MSG, &call->flags);
  99                 }
 100
 101                 /* allow interception by a kernel service */
 102                 if (rx->interceptor) {
 103                         rx->interceptor(sk, call->user_call_ID, skb);
 104                         spin_unlock_bh(&sk->sk_receive_queue.lock);
 105                 } else {
 106
 107                         /* Cache the SKB length before we tack it onto the
 108                          * receive queue.  Once it is added it no longer
 109                          * belongs to us and may be freed by other threads of
 110                          * control pulling packets from the queue */
 111                         skb_len = skb->len;
 112
 113                         _net("post skb %p", skb);
 114                         __skb_queue_tail(&sk->sk_receive_queue, skb);
 115                         spin_unlock_bh(&sk->sk_receive_queue.lock);
 116
 117                         if (!sock_flag(sk, SOCK_DEAD))
 118                                 sk->sk_data_ready(sk, skb_len);
 119                 }
 120                 skb = NULL;
 121         } else {
 122                 spin_unlock_bh(&sk->sk_receive_queue.lock);
 123         }
 124         ret = 0;
 125
 126 out:
 127         /* release the socket buffer */
 128         if (skb) {
 129                 skb->destructor = NULL;
 130                 sp->call = NULL;
 131                 rxrpc_put_call(call);
 132                 rxrpc_free_skb(skb);
 133         }
 134
 135         _leave(" = %d", ret);
 136         return ret;
 137 }
 138
 139 /*
 140  * process a DATA packet, posting the packet to the appropriate queue
 141  * - eats the packet if successful
 142  */
 143 static int rxrpc_fast_process_data(struct rxrpc_call *call,
 144                                    struct sk_buff *skb, u32 seq)
 145 {
 146         struct rxrpc_skb_priv *sp;
 147         bool terminal;
 148         int ret, ackbit, ack;
 149
 150         _enter("{%u,%u},,{%u}", call->rx_data_post, call->rx_first_oos, seq);
 151
 152         sp = rxrpc_skb(skb);
 153         ASSERTCMP(sp->call, ==, NULL);
 154
 155         spin_lock(&call->lock);
 156
 157         if (call->state > RXRPC_CALL_COMPLETE)
 158                 goto discard;
 159
 160         ASSERTCMP(call->rx_data_expect, >=, call->rx_data_post);
 161         ASSERTCMP(call->rx_data_post, >=, call->rx_data_recv);
 162         ASSERTCMP(call->rx_data_recv, >=, call->rx_data_eaten);
 163
 164         if (seq < call->rx_data_post) {
 165                 _debug("dup #%u [-%u]", seq, call->rx_data_post);
 166                 ack = RXRPC_ACK_DUPLICATE;
 167                 ret = -ENOBUFS;
 168                 goto discard_and_ack;
 169         }
 170
 171         /* we may already have the packet in the out of sequence queue */
 172         ackbit = seq - (call->rx_data_eaten + 1);
 173         ASSERTCMP(ackbit, >=, 0);
 174         if (__test_and_set_bit(ackbit, call->ackr_window)) {
 175                 _debug("dup oos #%u [%u,%u]",
 176                        seq, call->rx_data_eaten, call->rx_data_post);
 177                 ack = RXRPC_ACK_DUPLICATE;
 178                 goto discard_and_ack;
 179         }
 180
 181         if (seq >= call->ackr_win_top) {
 182                 _debug("exceed #%u [%u]", seq, call->ackr_win_top);
 183                 __clear_bit(ackbit, call->ackr_window);
 184                 ack = RXRPC_ACK_EXCEEDS_WINDOW;
 185                 goto discard_and_ack;
 186         }
 187
 188         if (seq == call->rx_data_expect) {
 189                 clear_bit(RXRPC_CALL_EXPECT_OOS, &call->flags);
 190                 call->rx_data_expect++;
 191         } else if (seq > call->rx_data_expect) {
 192                 _debug("oos #%u [%u]", seq, call->rx_data_expect);
 193                 call->rx_data_expect = seq + 1;
 194                 if (test_and_set_bit(RXRPC_CALL_EXPECT_OOS, &call->flags)) {
 195                         ack = RXRPC_ACK_OUT_OF_SEQUENCE;
 196                         goto enqueue_and_ack;
 197                 }
 198                 goto enqueue_packet;
 199         }
 200
 201         if (seq != call->rx_data_post) {
 202                 _debug("ahead #%u [%u]", seq, call->rx_data_post);
 203                 goto enqueue_packet;
 204         }
 205
 206         if (test_bit(RXRPC_CALL_RCVD_LAST, &call->flags))
 207                 goto protocol_error;
 208
 209         /* if the packet need security things doing to it, then it goes down
 210          * the slow path */
 211         if (call->conn->security)
 212                 goto enqueue_packet;
 213
 214         sp->call = call;
 215         rxrpc_get_call(call);
 216         terminal = ((sp->hdr.flags & RXRPC_LAST_PACKET) &&
 217                     !(sp->hdr.flags & RXRPC_CLIENT_INITIATED));
 218         ret = rxrpc_queue_rcv_skb(call, skb, false, terminal);
 219         if (ret < 0) {
 220                 if (ret == -ENOMEM || ret == -ENOBUFS) {
 221                         __clear_bit(ackbit, call->ackr_window);
 222                         ack = RXRPC_ACK_NOSPACE;
 223                         goto discard_and_ack;
 224                 }
 225                 goto out;
 226         }
 227
 228         skb = NULL;
 229
 230         _debug("post #%u", seq);
 231         ASSERTCMP(call->rx_data_post, ==, seq);
 232         call->rx_data_post++;
 233
 234         if (sp->hdr.flags & RXRPC_LAST_PACKET)
 235                 set_bit(RXRPC_CALL_RCVD_LAST, &call->flags);
 236
 237         /* if we've reached an out of sequence packet then we need to drain
 238          * that queue into the socket Rx queue now */
 239         if (call->rx_data_post == call->rx_first_oos) {
 240                 _debug("drain rx oos now");
 241                 read_lock(&call->state_lock);
 242                 if (call->state < RXRPC_CALL_COMPLETE &&
 243                     !test_and_set_bit(RXRPC_CALL_DRAIN_RX_OOS, &call->events))
 244                         rxrpc_queue_call(call);
 245                 read_unlock(&call->state_lock);
 246         }
 247
 248         spin_unlock(&call->lock);
 249         atomic_inc(&call->ackr_not_idle);
 250         rxrpc_propose_ACK(call, RXRPC_ACK_DELAY, sp->hdr.serial, false);
 251         _leave(" = 0 [posted]");
 252         return 0;
 253
 254 protocol_error:
 255         ret = -EBADMSG;
 256 out:
 257         spin_unlock(&call->lock);
 258         _leave(" = %d", ret);
 259         return ret;
 260
 261 discard_and_ack:
 262         _debug("discard and ACK packet %p", skb);
 263         __rxrpc_propose_ACK(call, ack, sp->hdr.serial, true);
 264 discard:
 265         spin_unlock(&call->lock);
 266         rxrpc_free_skb(skb);
 267         _leave(" = 0 [discarded]");
 268         return 0;
 269
 270 enqueue_and_ack:
 271         __rxrpc_propose_ACK(call, ack, sp->hdr.serial, true);
 272 enqueue_packet:
 273         _net("defer skb %p", skb);
 274         spin_unlock(&call->lock);
 275         skb_queue_tail(&call->rx_queue, skb);
 276         atomic_inc(&call->ackr_not_idle);
 277         read_lock(&call->state_lock);
 278         if (call->state < RXRPC_CALL_DEAD)
 279                 rxrpc_queue_call(call);
 280         read_unlock(&call->state_lock);
 281         _leave(" = 0 [queued]");
 282         return 0;
 283 }
 284
 285 /*
 286  * assume an implicit ACKALL of the transmission phase of a client socket upon
 287  * reception of the first reply packet
 288  */
 289 static void rxrpc_assume_implicit_ackall(struct rxrpc_call *call, u32 serial)
 290 {
 291         write_lock_bh(&call->state_lock);
 292
 293         switch (call->state) {
 294         case RXRPC_CALL_CLIENT_AWAIT_REPLY:
 295                 call->state = RXRPC_CALL_CLIENT_RECV_REPLY;
 296                 call->acks_latest = serial;
 297
 298                 _debug("implicit ACKALL %%%u", call->acks_latest);
 299                 set_bit(RXRPC_CALL_RCVD_ACKALL, &call->events);
 300                 write_unlock_bh(&call->state_lock);
 301
 302                 if (try_to_del_timer_sync(&call->resend_timer) >= 0) {
 303                         clear_bit(RXRPC_CALL_RESEND_TIMER, &call->events);
 304                         clear_bit(RXRPC_CALL_RESEND, &call->events);
 305                         clear_bit(RXRPC_CALL_RUN_RTIMER, &call->flags);
 306                 }
 307                 break;
 308
 309         default:
 310                 write_unlock_bh(&call->state_lock);
 311                 break;
 312         }
 313 }
 314
 315 /*
 316  * post an incoming packet to the nominated call to deal with
 317  * - must get rid of the sk_buff, either by freeing it or by queuing it
 318  */
 319 void rxrpc_fast_process_packet(struct rxrpc_call *call, struct sk_buff *skb)
 320 {
 321         struct rxrpc_skb_priv *sp = rxrpc_skb(skb);
 322         __be32 _abort_code;
 323         u32 serial, hi_serial, seq, abort_code;
 324
 325         _enter("%p,%p", call, skb);
 326
 327         ASSERT(!irqs_disabled());
 328
 329 #if 0 // INJECT RX ERROR
 330         if (sp->hdr.type == RXRPC_PACKET_TYPE_DATA) {
 331                 static int skip = 0;
 332                 if (++skip == 3) {
 333                         printk("DROPPED 3RD PACKET!!!!!!!!!!!!!\n");
 334                         skip = 0;
 335                         goto free_packet;
 336                 }
 337         }
 338 #endif
 339
 340         /* track the latest serial number on this connection for ACK packet
 341          * information */
 342         serial = ntohl(sp->hdr.serial);
 343         hi_serial = atomic_read(&call->conn->hi_serial);
 344         while (serial > hi_serial)
 345                 hi_serial = atomic_cmpxchg(&call->conn->hi_serial, hi_serial,
 346                                            serial);
 347
 348         /* request ACK generation for any ACK or DATA packet that requests
 349          * it */
 350         if (sp->hdr.flags & RXRPC_REQUEST_ACK) {
 351                 _proto("ACK Requested on %%%u", serial);
 352                 rxrpc_propose_ACK(call, RXRPC_ACK_REQUESTED, sp->hdr.serial,
 353                                   !(sp->hdr.flags & RXRPC_MORE_PACKETS));
 354         }
 355
 356         switch (sp->hdr.type) {
 357         case RXRPC_PACKET_TYPE_ABORT:
 358                 _debug("abort");
 359
 360                 if (skb_copy_bits(skb, 0, &_abort_code,
 361                                   sizeof(_abort_code)) < 0)
 362                         goto protocol_error;
 363
 364                 abort_code = ntohl(_abort_code);
 365                 _proto("Rx ABORT %%%u { %x }", serial, abort_code);
 366
 367                 write_lock_bh(&call->state_lock);
 368                 if (call->state < RXRPC_CALL_COMPLETE) {
 369                         call->state = RXRPC_CALL_REMOTELY_ABORTED;
 370                         call->abort_code = abort_code;
 371                         set_bit(RXRPC_CALL_RCVD_ABORT, &call->events);
 372                         rxrpc_queue_call(call);
 373                 }
 374                 goto free_packet_unlock;
 375
 376         case RXRPC_PACKET_TYPE_BUSY:
 377                 _proto("Rx BUSY %%%u", serial);
 378
 379                 if (call->conn->out_clientflag)
 380                         goto protocol_error;
 381
 382                 write_lock_bh(&call->state_lock);
 383                 switch (call->state) {
 384                 case RXRPC_CALL_CLIENT_SEND_REQUEST:
 385                         call->state = RXRPC_CALL_SERVER_BUSY;
 386                         set_bit(RXRPC_CALL_RCVD_BUSY, &call->events);
 387                         rxrpc_queue_call(call);
 388                 case RXRPC_CALL_SERVER_BUSY:
 389                         goto free_packet_unlock;
 390                 default:
 391                         goto protocol_error_locked;
 392                 }
 393
 394         default:
 395                 _proto("Rx %s %%%u", rxrpc_pkts[sp->hdr.type], serial);
 396                 goto protocol_error;
 397
 398         case RXRPC_PACKET_TYPE_DATA:
 399                 seq = ntohl(sp->hdr.seq);
 400
 401                 _proto("Rx DATA %%%u { #%u }", serial, seq);
 402
 403                 if (seq == 0)
 404                         goto protocol_error;
 405
 406                 call->ackr_prev_seq = sp->hdr.seq;
 407
 408                 /* received data implicitly ACKs all of the request packets we
 409                  * sent when we're acting as a client */
 410                 if (call->state == RXRPC_CALL_CLIENT_AWAIT_REPLY)
 411                         rxrpc_assume_implicit_ackall(call, serial);
 412
 413                 switch (rxrpc_fast_process_data(call, skb, seq)) {
 414                 case 0:
 415                         skb = NULL;
 416                         goto done;
 417
 418                 default:
 419                         BUG();
 420
 421                         /* data packet received beyond the last packet */
 422                 case -EBADMSG:
 423                         goto protocol_error;
 424                 }
 425
 426         case RXRPC_PACKET_TYPE_ACKALL:
 427         case RXRPC_PACKET_TYPE_ACK:
 428                 /* ACK processing is done in process context */
 429                 read_lock_bh(&call->state_lock);
 430                 if (call->state < RXRPC_CALL_DEAD) {
 431                         skb_queue_tail(&call->rx_queue, skb);
 432                         rxrpc_queue_call(call);
 433                         skb = NULL;
 434                 }
 435                 read_unlock_bh(&call->state_lock);
 436                 goto free_packet;
 437         }
 438
 439 protocol_error:
 440         _debug("protocol error");
 441         write_lock_bh(&call->state_lock);
 442 protocol_error_locked:
 443         if (call->state <= RXRPC_CALL_COMPLETE) {
 444                 call->state = RXRPC_CALL_LOCALLY_ABORTED;
 445                 call->abort_code = RX_PROTOCOL_ERROR;
 446                 set_bit(RXRPC_CALL_ABORT, &call->events);
 447                 rxrpc_queue_call(call);
 448         }
 449 free_packet_unlock:
 450         write_unlock_bh(&call->state_lock);
 451 free_packet:
 452         rxrpc_free_skb(skb);
 453 done:
 454         _leave("");
 455 }
 456
 457 /*
 458  * split up a jumbo data packet
 459  */
 460 static void rxrpc_process_jumbo_packet(struct rxrpc_call *call,
 461                                        struct sk_buff *jumbo)
 462 {
 463         struct rxrpc_jumbo_header jhdr;
 464         struct rxrpc_skb_priv *sp;
 465         struct sk_buff *part;
 466
 467         _enter(",{%u,%u}", jumbo->data_len, jumbo->len);
 468
 469         sp = rxrpc_skb(jumbo);
 470
 471         do {
 472                 sp->hdr.flags &= ~RXRPC_JUMBO_PACKET;
 473
 474                 /* make a clone to represent the first subpacket in what's left
 475                  * of the jumbo packet */
 476                 part = skb_clone(jumbo, GFP_ATOMIC);
 477                 if (!part) {
 478                         /* simply ditch the tail in the event of ENOMEM */
 479                         pskb_trim(jumbo, RXRPC_JUMBO_DATALEN);
 480                         break;
 481                 }
 482                 rxrpc_new_skb(part);
 483
 484                 pskb_trim(part, RXRPC_JUMBO_DATALEN);
 485
 486                 if (!pskb_pull(jumbo, RXRPC_JUMBO_DATALEN))
 487                         goto protocol_error;
 488
 489                 if (skb_copy_bits(jumbo, 0, &jhdr, sizeof(jhdr)) < 0)
 490                         goto protocol_error;
 491                 if (!pskb_pull(jumbo, sizeof(jhdr)))
 492                         BUG();
 493
 494                 sp->hdr.seq     = htonl(ntohl(sp->hdr.seq) + 1);
 495                 sp->hdr.serial  = htonl(ntohl(sp->hdr.serial) + 1);
 496                 sp->hdr.flags   = jhdr.flags;
 497                 sp->hdr._rsvd   = jhdr._rsvd;
 498
 499                 _proto("Rx DATA Jumbo %%%u", ntohl(sp->hdr.serial) - 1);
 500
 501                 rxrpc_fast_process_packet(call, part);
 502                 part = NULL;
 503
 504         } while (sp->hdr.flags & RXRPC_JUMBO_PACKET);
 505
 506         rxrpc_fast_process_packet(call, jumbo);
 507         _leave("");
 508         return;
 509
 510 protocol_error:
 511         _debug("protocol error");
 512         rxrpc_free_skb(part);
 513         rxrpc_free_skb(jumbo);
 514         write_lock_bh(&call->state_lock);
 515         if (call->state <= RXRPC_CALL_COMPLETE) {
 516                 call->state = RXRPC_CALL_LOCALLY_ABORTED;
 517                 call->abort_code = RX_PROTOCOL_ERROR;
 518                 set_bit(RXRPC_CALL_ABORT, &call->events);
 519                 rxrpc_queue_call(call);
 520         }
 521         write_unlock_bh(&call->state_lock);
 522         _leave("");
 523 }
 524
 525 /*
 526  * post an incoming packet to the appropriate call/socket to deal with
 527  * - must get rid of the sk_buff, either by freeing it or by queuing it
 528  */
 529 static void rxrpc_post_packet_to_call(struct rxrpc_connection *conn,
 530                                       struct sk_buff *skb)
 531 {
 532         struct rxrpc_skb_priv *sp;
 533         struct rxrpc_call *call;
 534         struct rb_node *p;
 535         __be32 call_id;
 536
 537         _enter("%p,%p", conn, skb);
 538
 539         read_lock_bh(&conn->lock);
 540
 541         sp = rxrpc_skb(skb);
 542
 543         /* look at extant calls by channel number first */
 544         call = conn->channels[ntohl(sp->hdr.cid) & RXRPC_CHANNELMASK];
 545         if (!call || call->call_id != sp->hdr.callNumber)
 546                 goto call_not_extant;
 547
 548         _debug("extant call [%d]", call->state);
 549         ASSERTCMP(call->conn, ==, conn);
 550
 551         read_lock(&call->state_lock);
 552         switch (call->state) {
 553         case RXRPC_CALL_LOCALLY_ABORTED:
 554                 if (!test_and_set_bit(RXRPC_CALL_ABORT, &call->events))
 555                         rxrpc_queue_call(call);
 556         case RXRPC_CALL_REMOTELY_ABORTED:
 557         case RXRPC_CALL_NETWORK_ERROR:
 558         case RXRPC_CALL_DEAD:
 559                 goto free_unlock;
 560         default:
 561                 break;
 562         }
 563
 564         read_unlock(&call->state_lock);
 565         rxrpc_get_call(call);
 566         read_unlock_bh(&conn->lock);
 567
 568         if (sp->hdr.type == RXRPC_PACKET_TYPE_DATA &&
 569             sp->hdr.flags & RXRPC_JUMBO_PACKET)
 570                 rxrpc_process_jumbo_packet(call, skb);
 571         else
 572                 rxrpc_fast_process_packet(call, skb);
 573
 574         rxrpc_put_call(call);
 575         goto done;
 576
 577 call_not_extant:
 578         /* search the completed calls in case what we're dealing with is
 579          * there */
 580         _debug("call not extant");
 581
 582         call_id = sp->hdr.callNumber;
 583         p = conn->calls.rb_node;
 584         while (p) {
 585                 call = rb_entry(p, struct rxrpc_call, conn_node);
 586
 587                 if (call_id < call->call_id)
 588                         p = p->rb_left;
 589                 else if (call_id > call->call_id)
 590                         p = p->rb_right;
 591                 else
 592                         goto found_completed_call;
 593         }
 594
 595 dead_call:
 596         /* it's a either a really old call that we no longer remember or its a
 597          * new incoming call */
 598         read_unlock_bh(&conn->lock);
 599
 600         if (sp->hdr.flags & RXRPC_CLIENT_INITIATED &&
 601             sp->hdr.seq == cpu_to_be32(1)) {
 602                 _debug("incoming call");
 603                 skb_queue_tail(&conn->trans->local->accept_queue, skb);
 604                 rxrpc_queue_work(&conn->trans->local->acceptor);
 605                 goto done;
 606         }
 607
 608         _debug("dead call");
 609         skb->priority = RX_CALL_DEAD;
 610         rxrpc_reject_packet(conn->trans->local, skb);
 611         goto done;
 612
 613         /* resend last packet of a completed call
 614          * - client calls may have been aborted or ACK'd
 615          * - server calls may have been aborted
 616          */
 617 found_completed_call:
 618         _debug("completed call");
 619
 620         if (atomic_read(&call->usage) == 0)
 621                 goto dead_call;
 622
 623         /* synchronise any state changes */
 624         read_lock(&call->state_lock);
 625         ASSERTIFCMP(call->state != RXRPC_CALL_CLIENT_FINAL_ACK,
 626                     call->state, >=, RXRPC_CALL_COMPLETE);
 627
 628         if (call->state == RXRPC_CALL_LOCALLY_ABORTED ||
 629             call->state == RXRPC_CALL_REMOTELY_ABORTED ||
 630             call->state == RXRPC_CALL_DEAD) {
 631                 read_unlock(&call->state_lock);
 632                 goto dead_call;
 633         }
 634
 635         if (call->conn->in_clientflag) {
 636                 read_unlock(&call->state_lock);
 637                 goto dead_call; /* complete server call */
 638         }
 639
 640         _debug("final ack again");
 641         rxrpc_get_call(call);
 642         set_bit(RXRPC_CALL_ACK_FINAL, &call->events);
 643         rxrpc_queue_call(call);
 644
 645 free_unlock:
 646         read_unlock(&call->state_lock);
 647         read_unlock_bh(&conn->lock);
 648         rxrpc_free_skb(skb);
 649 done:
 650         _leave("");
 651 }
 652
 653 /*
 654  * post connection-level events to the connection
 655  * - this includes challenges, responses and some aborts
 656  */
 657 static void rxrpc_post_packet_to_conn(struct rxrpc_connection *conn,
 658                                       struct sk_buff *skb)
 659 {
 660         _enter("%p,%p", conn, skb);
 661
 662         atomic_inc(&conn->usage);
 663         skb_queue_tail(&conn->rx_queue, skb);
 664         rxrpc_queue_conn(conn);
 665 }
 666
 667 /*
 668  * handle data received on the local endpoint
 669  * - may be called in interrupt context
 670  */
 671 void rxrpc_data_ready(struct sock *sk, int count)
 672 {
 673         struct rxrpc_connection *conn;
 674         struct rxrpc_transport *trans;
 675         struct rxrpc_skb_priv *sp;
 676         struct rxrpc_local *local;
 677         struct rxrpc_peer *peer;
 678         struct sk_buff *skb;
 679         int ret;
 680
 681         _enter("%p, %d", sk, count);
 682
 683         ASSERT(!irqs_disabled());
 684
 685         read_lock_bh(&rxrpc_local_lock);
 686         local = sk->sk_user_data;
 687         if (local && atomic_read(&local->usage) > 0)
 688                 rxrpc_get_local(local);
 689         else
 690                 local = NULL;
 691         read_unlock_bh(&rxrpc_local_lock);
 692         if (!local) {
 693                 _leave(" [local dead]");
 694                 return;
 695         }
 696
 697         skb = skb_recv_datagram(sk, 0, 1, &ret);
 698         if (!skb) {
 699                 rxrpc_put_local(local);
 700                 if (ret == -EAGAIN)
 701                         return;
 702                 _debug("UDP socket error %d", ret);
 703                 return;
 704         }
 705
 706         rxrpc_new_skb(skb);
 707
 708         _net("recv skb %p", skb);
 709
 710         /* we'll probably need to checksum it (didn't call sock_recvmsg) */
 711         if (skb_checksum_complete(skb)) {
 712                 rxrpc_free_skb(skb);
 713                 rxrpc_put_local(local);
 714                 UDP_INC_STATS_BH(&init_net, UDP_MIB_INERRORS, 0);
 715                 _leave(" [CSUM failed]");
 716                 return;
 717         }
 718
 719         UDP_INC_STATS_BH(&init_net, UDP_MIB_INDATAGRAMS, 0);
 720
 721         /* the socket buffer we have is owned by UDP, with UDP's data all over
 722          * it, but we really want our own */
 723         skb_orphan(skb);
 724         sp = rxrpc_skb(skb);
 725         memset(sp, 0, sizeof(*sp));
 726
 727         _net("Rx UDP packet from %08x:%04hu",
 728              ntohl(ip_hdr(skb)->saddr), ntohs(udp_hdr(skb)->source));
 729
 730         /* dig out the RxRPC connection details */
 731         if (skb_copy_bits(skb, sizeof(struct udphdr), &sp->hdr,
 732                           sizeof(sp->hdr)) < 0)
 733                 goto bad_message;
 734         if (!pskb_pull(skb, sizeof(struct udphdr) + sizeof(sp->hdr)))
 735                 BUG();
 736
 737         _net("Rx RxRPC %s ep=%x call=%x:%x",
 738              sp->hdr.flags & RXRPC_CLIENT_INITIATED ? "ToServer" : "ToClient",
 739              ntohl(sp->hdr.epoch),
 740              ntohl(sp->hdr.cid),
 741              ntohl(sp->hdr.callNumber));
 742
 743         if (sp->hdr.type == 0 || sp->hdr.type >= RXRPC_N_PACKET_TYPES) {
 744                 _proto("Rx Bad Packet Type %u", sp->hdr.type);
 745                 goto bad_message;
 746         }
 747
 748         if (sp->hdr.type == RXRPC_PACKET_TYPE_DATA &&
 749             (sp->hdr.callNumber == 0 || sp->hdr.seq == 0))
 750                 goto bad_message;
 751
 752         peer = rxrpc_find_peer(local, ip_hdr(skb)->saddr, udp_hdr(skb)->source);
 753         if (IS_ERR(peer))
 754                 goto cant_route_call;
 755
 756         trans = rxrpc_find_transport(local, peer);
 757         rxrpc_put_peer(peer);
 758         if (!trans)
 759                 goto cant_route_call;
 760
 761         conn = rxrpc_find_connection(trans, &sp->hdr);
 762         rxrpc_put_transport(trans);
 763         if (!conn)
 764                 goto cant_route_call;
 765
 766         _debug("CONN %p {%d}", conn, conn->debug_id);
 767
 768         if (sp->hdr.callNumber == 0)
 769                 rxrpc_post_packet_to_conn(conn, skb);
 770         else
 771                 rxrpc_post_packet_to_call(conn, skb);
 772         rxrpc_put_connection(conn);
 773         rxrpc_put_local(local);
 774         return;
 775
 776 cant_route_call:
 777         _debug("can't route call");
 778         if (sp->hdr.flags & RXRPC_CLIENT_INITIATED &&
 779             sp->hdr.type == RXRPC_PACKET_TYPE_DATA) {
 780                 if (sp->hdr.seq == cpu_to_be32(1)) {
 781                         _debug("first packet");
 782                         skb_queue_tail(&local->accept_queue, skb);
 783                         rxrpc_queue_work(&local->acceptor);
 784                         rxrpc_put_local(local);
 785                         _leave(" [incoming]");
 786                         return;
 787                 }
 788                 skb->priority = RX_INVALID_OPERATION;
 789         } else {
 790                 skb->priority = RX_CALL_DEAD;
 791         }
 792
 793         _debug("reject");
 794         rxrpc_reject_packet(local, skb);
 795         rxrpc_put_local(local);
 796         _leave(" [no call]");
 797         return;
 798
 799 bad_message:
 800         skb->priority = RX_PROTOCOL_ERROR;
 801         rxrpc_reject_packet(local, skb);
 802         rxrpc_put_local(local);
 803         _leave(" [badmsg]");
 804 }