net/rxrpc/recvmsg.c

   1 // SPDX-License-Identifier: GPL-2.0-or-later
   2 /* RxRPC recvmsg() implementation
   3  *
   4  * Copyright (C) 2007 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 <linux/net.h>
  11 #include <linux/skbuff.h>
  12 #include <linux/export.h>
  13 #include <linux/sched/signal.h>
  14
  15 #include <net/sock.h>
  16 #include <net/af_rxrpc.h>
  17 #include "ar-internal.h"
  18
  19 /*
  20  * Post a call for attention by the socket or kernel service.  Further
  21  * notifications are suppressed by putting recvmsg_link on a dummy queue.
  22  */
  23 void rxrpc_notify_socket(struct rxrpc_call *call)
  24 {
  25         struct rxrpc_sock *rx;
  26         struct sock *sk;
  27
  28         _enter("%d", call->debug_id);
  29
  30         if (!list_empty(&call->recvmsg_link))
  31                 return;
  32
  33         rcu_read_lock();
  34
  35         rx = rcu_dereference(call->socket);
  36         sk = &rx->sk;
  37         if (rx && sk->sk_state < RXRPC_CLOSE) {
  38                 if (call->notify_rx) {
  39                         spin_lock_bh(&call->notify_lock);
  40                         call->notify_rx(sk, call, call->user_call_ID);
  41                         spin_unlock_bh(&call->notify_lock);
  42                 } else {
  43                         write_lock_bh(&rx->recvmsg_lock);
  44                         if (list_empty(&call->recvmsg_link)) {
  45                                 rxrpc_get_call(call, rxrpc_call_got);
  46                                 list_add_tail(&call->recvmsg_link, &rx->recvmsg_q);
  47                         }
  48                         write_unlock_bh(&rx->recvmsg_lock);
  49
  50                         if (!sock_flag(sk, SOCK_DEAD)) {
  51                                 _debug("call %ps", sk->sk_data_ready);
  52                                 sk->sk_data_ready(sk);
  53                         }
  54                 }
  55         }
  56
  57         rcu_read_unlock();
  58         _leave("");
  59 }
  60
  61 /*
  62  * Transition a call to the complete state.
  63  */
  64 bool __rxrpc_set_call_completion(struct rxrpc_call *call,
  65                                  enum rxrpc_call_completion compl,
  66                                  u32 abort_code,
  67                                  int error)
  68 {
  69         if (call->state < RXRPC_CALL_COMPLETE) {
  70                 call->abort_code = abort_code;
  71                 call->error = error;
  72                 call->completion = compl;
  73                 call->state = RXRPC_CALL_COMPLETE;
  74                 trace_rxrpc_call_complete(call);
  75                 wake_up(&call->waitq);
  76                 rxrpc_notify_socket(call);
  77                 return true;
  78         }
  79         return false;
  80 }
  81
  82 bool rxrpc_set_call_completion(struct rxrpc_call *call,
  83                                enum rxrpc_call_completion compl,
  84                                u32 abort_code,
  85                                int error)
  86 {
  87         bool ret = false;
  88
  89         if (call->state < RXRPC_CALL_COMPLETE) {
  90                 write_lock_bh(&call->state_lock);
  91                 ret = __rxrpc_set_call_completion(call, compl, abort_code, error);
  92                 write_unlock_bh(&call->state_lock);
  93         }
  94         return ret;
  95 }
  96
  97 /*
  98  * Record that a call successfully completed.
  99  */
 100 bool __rxrpc_call_completed(struct rxrpc_call *call)
 101 {
 102         return __rxrpc_set_call_completion(call, RXRPC_CALL_SUCCEEDED, 0, 0);
 103 }
 104
 105 bool rxrpc_call_completed(struct rxrpc_call *call)
 106 {
 107         bool ret = false;
 108
 109         if (call->state < RXRPC_CALL_COMPLETE) {
 110                 write_lock_bh(&call->state_lock);
 111                 ret = __rxrpc_call_completed(call);
 112                 write_unlock_bh(&call->state_lock);
 113         }
 114         return ret;
 115 }
 116
 117 /*
 118  * Record that a call is locally aborted.
 119  */
 120 bool __rxrpc_abort_call(const char *why, struct rxrpc_call *call,
 121                         rxrpc_seq_t seq, u32 abort_code, int error)
 122 {
 123         trace_rxrpc_abort(call->debug_id, why, call->cid, call->call_id, seq,
 124                           abort_code, error);
 125         return __rxrpc_set_call_completion(call, RXRPC_CALL_LOCALLY_ABORTED,
 126                                            abort_code, error);
 127 }
 128
 129 bool rxrpc_abort_call(const char *why, struct rxrpc_call *call,
 130                       rxrpc_seq_t seq, u32 abort_code, int error)
 131 {
 132         bool ret;
 133
 134         write_lock_bh(&call->state_lock);
 135         ret = __rxrpc_abort_call(why, call, seq, abort_code, error);
 136         write_unlock_bh(&call->state_lock);
 137         return ret;
 138 }
 139
 140 /*
 141  * Pass a call terminating message to userspace.
 142  */
 143 static int rxrpc_recvmsg_term(struct rxrpc_call *call, struct msghdr *msg)
 144 {
 145         u32 tmp = 0;
 146         int ret;
 147
 148         switch (call->completion) {
 149         case RXRPC_CALL_SUCCEEDED:
 150                 ret = 0;
 151                 if (rxrpc_is_service_call(call))
 152                         ret = put_cmsg(msg, SOL_RXRPC, RXRPC_ACK, 0, &tmp);
 153                 break;
 154         case RXRPC_CALL_REMOTELY_ABORTED:
 155                 tmp = call->abort_code;
 156                 ret = put_cmsg(msg, SOL_RXRPC, RXRPC_ABORT, 4, &tmp);
 157                 break;
 158         case RXRPC_CALL_LOCALLY_ABORTED:
 159                 tmp = call->abort_code;
 160                 ret = put_cmsg(msg, SOL_RXRPC, RXRPC_ABORT, 4, &tmp);
 161                 break;
 162         case RXRPC_CALL_NETWORK_ERROR:
 163                 tmp = -call->error;
 164                 ret = put_cmsg(msg, SOL_RXRPC, RXRPC_NET_ERROR, 4, &tmp);
 165                 break;
 166         case RXRPC_CALL_LOCAL_ERROR:
 167                 tmp = -call->error;
 168                 ret = put_cmsg(msg, SOL_RXRPC, RXRPC_LOCAL_ERROR, 4, &tmp);
 169                 break;
 170         default:
 171                 pr_err("Invalid terminal call state %u\n", call->state);
 172                 BUG();
 173                 break;
 174         }
 175
 176         trace_rxrpc_recvdata(call, rxrpc_recvmsg_terminal, call->rx_hard_ack,
 177                              call->rx_pkt_offset, call->rx_pkt_len, ret);
 178         return ret;
 179 }
 180
 181 /*
 182  * End the packet reception phase.
 183  */
 184 static void rxrpc_end_rx_phase(struct rxrpc_call *call, rxrpc_serial_t serial)
 185 {
 186         _enter("%d,%s", call->debug_id, rxrpc_call_states[call->state]);
 187
 188         trace_rxrpc_receive(call, rxrpc_receive_end, 0, call->rx_top);
 189         ASSERTCMP(call->rx_hard_ack, ==, call->rx_top);
 190
 191         if (call->state == RXRPC_CALL_CLIENT_RECV_REPLY)
 192                 rxrpc_propose_delay_ACK(call, serial, rxrpc_propose_ack_terminal_ack);
 193
 194         write_lock_bh(&call->state_lock);
 195
 196         switch (call->state) {
 197         case RXRPC_CALL_CLIENT_RECV_REPLY:
 198                 __rxrpc_call_completed(call);
 199                 write_unlock_bh(&call->state_lock);
 200                 break;
 201
 202         case RXRPC_CALL_SERVER_RECV_REQUEST:
 203                 call->state = RXRPC_CALL_SERVER_ACK_REQUEST;
 204                 call->expect_req_by = jiffies + MAX_JIFFY_OFFSET;
 205                 write_unlock_bh(&call->state_lock);
 206                 rxrpc_propose_delay_ACK(call, serial,
 207                                         rxrpc_propose_ack_processing_op);
 208                 break;
 209         default:
 210                 write_unlock_bh(&call->state_lock);
 211                 break;
 212         }
 213 }
 214
 215 /*
 216  * Discard a packet we've used up and advance the Rx window by one.
 217  */
 218 static void rxrpc_rotate_rx_window(struct rxrpc_call *call)
 219 {
 220         struct rxrpc_skb_priv *sp;
 221         struct sk_buff *skb;
 222         rxrpc_serial_t serial;
 223         rxrpc_seq_t hard_ack, top;
 224         bool last = false;
 225         int ix;
 226
 227         _enter("%d", call->debug_id);
 228
 229         hard_ack = call->rx_hard_ack;
 230         top = smp_load_acquire(&call->rx_top);
 231         ASSERT(before(hard_ack, top));
 232
 233         hard_ack++;
 234         ix = hard_ack & RXRPC_RXTX_BUFF_MASK;
 235         skb = call->rxtx_buffer[ix];
 236         rxrpc_see_skb(skb, rxrpc_skb_rotated);
 237         sp = rxrpc_skb(skb);
 238
 239         serial = sp->hdr.serial;
 240
 241         if (sp->hdr.flags & RXRPC_LAST_PACKET)
 242                 last = true;
 243
 244         call->rxtx_buffer[ix] = NULL;
 245         call->rxtx_annotations[ix] = 0;
 246         /* Barrier against rxrpc_input_data(). */
 247         smp_store_release(&call->rx_hard_ack, hard_ack);
 248
 249         rxrpc_free_skb(skb, rxrpc_skb_freed);
 250
 251         trace_rxrpc_receive(call, rxrpc_receive_rotate, serial, hard_ack);
 252         if (last) {
 253                 rxrpc_end_rx_phase(call, serial);
 254         } else {
 255                 /* Check to see if there's an ACK that needs sending. */
 256                 if (atomic_inc_return(&call->ackr_nr_consumed) > 2 &&
 257                     !test_and_set_bit(RXRPC_CALL_IDLE_ACK_PENDING, &call->flags)) {
 258                         rxrpc_send_ACK(call, RXRPC_ACK_IDLE, serial,
 259                                        rxrpc_propose_ack_rotate_rx);
 260                         rxrpc_transmit_ack_packets(call->peer->local);
 261                 }
 262         }
 263 }
 264
 265 /*
 266  * Decrypt and verify a DATA packet.
 267  */
 268 static int rxrpc_verify_data(struct rxrpc_call *call, struct sk_buff *skb)
 269 {
 270         struct rxrpc_skb_priv *sp = rxrpc_skb(skb);
 271
 272         if (sp->flags & RXRPC_RX_VERIFIED)
 273                 return 0;
 274         return call->security->verify_packet(call, skb);
 275 }
 276
 277 /*
 278  * Deliver messages to a call.  This keeps processing packets until the buffer
 279  * is filled and we find either more DATA (returns 0) or the end of the DATA
 280  * (returns 1).  If more packets are required, it returns -EAGAIN.
 281  */
 282 static int rxrpc_recvmsg_data(struct socket *sock, struct rxrpc_call *call,
 283                               struct msghdr *msg, struct iov_iter *iter,
 284                               size_t len, int flags, size_t *_offset)
 285 {
 286         struct rxrpc_skb_priv *sp;
 287         struct sk_buff *skb;
 288         rxrpc_serial_t serial;
 289         rxrpc_seq_t hard_ack, top, seq;
 290         size_t remain;
 291         unsigned int rx_pkt_offset, rx_pkt_len;
 292         int ix, copy, ret = -EAGAIN, ret2;
 293
 294         rx_pkt_offset = call->rx_pkt_offset;
 295         rx_pkt_len = call->rx_pkt_len;
 296
 297         if (call->state >= RXRPC_CALL_SERVER_ACK_REQUEST) {
 298                 seq = call->rx_hard_ack;
 299                 ret = 1;
 300                 goto done;
 301         }
 302
 303         /* Barriers against rxrpc_input_data(). */
 304         hard_ack = call->rx_hard_ack;
 305         seq = hard_ack + 1;
 306
 307         while (top = smp_load_acquire(&call->rx_top),
 308                before_eq(seq, top)
 309                ) {
 310                 ix = seq & RXRPC_RXTX_BUFF_MASK;
 311                 skb = call->rxtx_buffer[ix];
 312                 if (!skb) {
 313                         trace_rxrpc_recvdata(call, rxrpc_recvmsg_hole, seq,
 314                                              rx_pkt_offset, rx_pkt_len, 0);
 315                         rxrpc_transmit_ack_packets(call->peer->local);
 316                         break;
 317                 }
 318                 smp_rmb();
 319                 rxrpc_see_skb(skb, rxrpc_skb_seen);
 320                 sp = rxrpc_skb(skb);
 321
 322                 if (!(flags & MSG_PEEK)) {
 323                         serial = sp->hdr.serial;
 324                         trace_rxrpc_receive(call, rxrpc_receive_front,
 325                                             serial, seq);
 326                 }
 327
 328                 if (msg)
 329                         sock_recv_timestamp(msg, sock->sk, skb);
 330
 331                 if (rx_pkt_offset == 0) {
 332                         ret2 = rxrpc_verify_data(call, skb);
 333                         rx_pkt_offset = sp->offset;
 334                         rx_pkt_len = sp->len;
 335                         trace_rxrpc_recvdata(call, rxrpc_recvmsg_next, seq,
 336                                              rx_pkt_offset, rx_pkt_len, ret2);
 337                         if (ret2 < 0) {
 338                                 ret = ret2;
 339                                 goto out;
 340                         }
 341                 } else {
 342                         trace_rxrpc_recvdata(call, rxrpc_recvmsg_cont, seq,
 343                                              rx_pkt_offset, rx_pkt_len, 0);
 344                 }
 345
 346                 /* We have to handle short, empty and used-up DATA packets. */
 347                 remain = len - *_offset;
 348                 copy = rx_pkt_len;
 349                 if (copy > remain)
 350                         copy = remain;
 351                 if (copy > 0) {
 352                         ret2 = skb_copy_datagram_iter(skb, rx_pkt_offset, iter,
 353                                                       copy);
 354                         if (ret2 < 0) {
 355                                 ret = ret2;
 356                                 goto out;
 357                         }
 358
 359                         /* handle piecemeal consumption of data packets */
 360                         rx_pkt_offset += copy;
 361                         rx_pkt_len -= copy;
 362                         *_offset += copy;
 363                 }
 364
 365                 if (rx_pkt_len > 0) {
 366                         trace_rxrpc_recvdata(call, rxrpc_recvmsg_full, seq,
 367                                              rx_pkt_offset, rx_pkt_len, 0);
 368                         ASSERTCMP(*_offset, ==, len);
 369                         ret = 0;
 370                         break;
 371                 }
 372
 373                 /* The whole packet has been transferred. */
 374                 if (sp->hdr.flags & RXRPC_LAST_PACKET)
 375                         ret = 1;
 376                 rx_pkt_offset = 0;
 377                 rx_pkt_len = 0;
 378
 379                 if (!(flags & MSG_PEEK))
 380                         rxrpc_rotate_rx_window(call);
 381
 382                 seq++;
 383         }
 384
 385 out:
 386         if (!(flags & MSG_PEEK)) {
 387                 call->rx_pkt_offset = rx_pkt_offset;
 388                 call->rx_pkt_len = rx_pkt_len;
 389         }
 390 done:
 391         trace_rxrpc_recvdata(call, rxrpc_recvmsg_data_return, seq,
 392                              rx_pkt_offset, rx_pkt_len, ret);
 393         if (ret == -EAGAIN)
 394                 set_bit(RXRPC_CALL_RX_UNDERRUN, &call->flags);
 395         return ret;
 396 }
 397
 398 /*
 399  * Receive a message from an RxRPC socket
 400  * - we need to be careful about two or more threads calling recvmsg
 401  *   simultaneously
 402  */
 403 int rxrpc_recvmsg(struct socket *sock, struct msghdr *msg, size_t len,
 404                   int flags)
 405 {
 406         struct rxrpc_call *call;
 407         struct rxrpc_sock *rx = rxrpc_sk(sock->sk);
 408         struct list_head *l;
 409         size_t copied = 0;
 410         long timeo;
 411         int ret;
 412
 413         DEFINE_WAIT(wait);
 414
 415         trace_rxrpc_recvmsg(NULL, rxrpc_recvmsg_enter, 0);
 416
 417         if (flags & (MSG_OOB | MSG_TRUNC))
 418                 return -EOPNOTSUPP;
 419
 420         timeo = sock_rcvtimeo(&rx->sk, flags & MSG_DONTWAIT);
 421
 422 try_again:
 423         lock_sock(&rx->sk);
 424
 425         /* Return immediately if a client socket has no outstanding calls */
 426         if (RB_EMPTY_ROOT(&rx->calls) &&
 427             list_empty(&rx->recvmsg_q) &&
 428             rx->sk.sk_state != RXRPC_SERVER_LISTENING) {
 429                 release_sock(&rx->sk);
 430                 return -EAGAIN;
 431         }
 432
 433         if (list_empty(&rx->recvmsg_q)) {
 434                 ret = -EWOULDBLOCK;
 435                 if (timeo == 0) {
 436                         call = NULL;
 437                         goto error_no_call;
 438                 }
 439
 440                 release_sock(&rx->sk);
 441
 442                 /* Wait for something to happen */
 443                 prepare_to_wait_exclusive(sk_sleep(&rx->sk), &wait,
 444                                           TASK_INTERRUPTIBLE);
 445                 ret = sock_error(&rx->sk);
 446                 if (ret)
 447                         goto wait_error;
 448
 449                 if (list_empty(&rx->recvmsg_q)) {
 450                         if (signal_pending(current))
 451                                 goto wait_interrupted;
 452                         trace_rxrpc_recvmsg(NULL, rxrpc_recvmsg_wait, 0);
 453                         timeo = schedule_timeout(timeo);
 454                 }
 455                 finish_wait(sk_sleep(&rx->sk), &wait);
 456                 goto try_again;
 457         }
 458
 459         /* Find the next call and dequeue it if we're not just peeking.  If we
 460          * do dequeue it, that comes with a ref that we will need to release.
 461          */
 462         write_lock_bh(&rx->recvmsg_lock);
 463         l = rx->recvmsg_q.next;
 464         call = list_entry(l, struct rxrpc_call, recvmsg_link);
 465         if (!(flags & MSG_PEEK))
 466                 list_del_init(&call->recvmsg_link);
 467         else
 468                 rxrpc_get_call(call, rxrpc_call_got);
 469         write_unlock_bh(&rx->recvmsg_lock);
 470
 471         trace_rxrpc_recvmsg(call, rxrpc_recvmsg_dequeue, 0);
 472
 473         /* We're going to drop the socket lock, so we need to lock the call
 474          * against interference by sendmsg.
 475          */
 476         if (!mutex_trylock(&call->user_mutex)) {
 477                 ret = -EWOULDBLOCK;
 478                 if (flags & MSG_DONTWAIT)
 479                         goto error_requeue_call;
 480                 ret = -ERESTARTSYS;
 481                 if (mutex_lock_interruptible(&call->user_mutex) < 0)
 482                         goto error_requeue_call;
 483         }
 484
 485         release_sock(&rx->sk);
 486
 487         if (test_bit(RXRPC_CALL_RELEASED, &call->flags))
 488                 BUG();
 489
 490         if (test_bit(RXRPC_CALL_HAS_USERID, &call->flags)) {
 491                 if (flags & MSG_CMSG_COMPAT) {
 492                         unsigned int id32 = call->user_call_ID;
 493
 494                         ret = put_cmsg(msg, SOL_RXRPC, RXRPC_USER_CALL_ID,
 495                                        sizeof(unsigned int), &id32);
 496                 } else {
 497                         unsigned long idl = call->user_call_ID;
 498
 499                         ret = put_cmsg(msg, SOL_RXRPC, RXRPC_USER_CALL_ID,
 500                                        sizeof(unsigned long), &idl);
 501                 }
 502                 if (ret < 0)
 503                         goto error_unlock_call;
 504         }
 505
 506         if (msg->msg_name && call->peer) {
 507                 struct sockaddr_rxrpc *srx = msg->msg_name;
 508                 size_t len = sizeof(call->peer->srx);
 509
 510                 memcpy(msg->msg_name, &call->peer->srx, len);
 511                 srx->srx_service = call->service_id;
 512                 msg->msg_namelen = len;
 513         }
 514
 515         switch (READ_ONCE(call->state)) {
 516         case RXRPC_CALL_CLIENT_RECV_REPLY:
 517         case RXRPC_CALL_SERVER_RECV_REQUEST:
 518         case RXRPC_CALL_SERVER_ACK_REQUEST:
 519                 ret = rxrpc_recvmsg_data(sock, call, msg, &msg->msg_iter, len,
 520                                          flags, &copied);
 521                 if (ret == -EAGAIN)
 522                         ret = 0;
 523
 524                 rxrpc_transmit_ack_packets(call->peer->local);
 525                 if (after(call->rx_top, call->rx_hard_ack) &&
 526                     call->rxtx_buffer[(call->rx_hard_ack + 1) & RXRPC_RXTX_BUFF_MASK])
 527                         rxrpc_notify_socket(call);
 528                 break;
 529         default:
 530                 ret = 0;
 531                 break;
 532         }
 533
 534         if (ret < 0)
 535                 goto error_unlock_call;
 536
 537         if (call->state == RXRPC_CALL_COMPLETE) {
 538                 ret = rxrpc_recvmsg_term(call, msg);
 539                 if (ret < 0)
 540                         goto error_unlock_call;
 541                 if (!(flags & MSG_PEEK))
 542                         rxrpc_release_call(rx, call);
 543                 msg->msg_flags |= MSG_EOR;
 544                 ret = 1;
 545         }
 546
 547         if (ret == 0)
 548                 msg->msg_flags |= MSG_MORE;
 549         else
 550                 msg->msg_flags &= ~MSG_MORE;
 551         ret = copied;
 552
 553 error_unlock_call:
 554         mutex_unlock(&call->user_mutex);
 555         rxrpc_put_call(call, rxrpc_call_put);
 556         trace_rxrpc_recvmsg(call, rxrpc_recvmsg_return, ret);
 557         return ret;
 558
 559 error_requeue_call:
 560         if (!(flags & MSG_PEEK)) {
 561                 write_lock_bh(&rx->recvmsg_lock);
 562                 list_add(&call->recvmsg_link, &rx->recvmsg_q);
 563                 write_unlock_bh(&rx->recvmsg_lock);
 564                 trace_rxrpc_recvmsg(call, rxrpc_recvmsg_requeue, 0);
 565         } else {
 566                 rxrpc_put_call(call, rxrpc_call_put);
 567         }
 568 error_no_call:
 569         release_sock(&rx->sk);
 570 error_trace:
 571         trace_rxrpc_recvmsg(call, rxrpc_recvmsg_return, ret);
 572         return ret;
 573
 574 wait_interrupted:
 575         ret = sock_intr_errno(timeo);
 576 wait_error:
 577         finish_wait(sk_sleep(&rx->sk), &wait);
 578         call = NULL;
 579         goto error_trace;
 580 }
 581
 582 /**
 583  * rxrpc_kernel_recv_data - Allow a kernel service to receive data/info
 584  * @sock: The socket that the call exists on
 585  * @call: The call to send data through
 586  * @iter: The buffer to receive into
 587  * @_len: The amount of data we want to receive (decreased on return)
 588  * @want_more: True if more data is expected to be read
 589  * @_abort: Where the abort code is stored if -ECONNABORTED is returned
 590  * @_service: Where to store the actual service ID (may be upgraded)
 591  *
 592  * Allow a kernel service to receive data and pick up information about the
 593  * state of a call.  Returns 0 if got what was asked for and there's more
 594  * available, 1 if we got what was asked for and we're at the end of the data
 595  * and -EAGAIN if we need more data.
 596  *
 597  * Note that we may return -EAGAIN to drain empty packets at the end of the
 598  * data, even if we've already copied over the requested data.
 599  *
 600  * *_abort should also be initialised to 0.
 601  */
 602 int rxrpc_kernel_recv_data(struct socket *sock, struct rxrpc_call *call,
 603                            struct iov_iter *iter, size_t *_len,
 604                            bool want_more, u32 *_abort, u16 *_service)
 605 {
 606         size_t offset = 0;
 607         int ret;
 608
 609         _enter("{%d,%s},%zu,%d",
 610                call->debug_id, rxrpc_call_states[call->state],
 611                *_len, want_more);
 612
 613         ASSERTCMP(call->state, !=, RXRPC_CALL_SERVER_SECURING);
 614
 615         mutex_lock(&call->user_mutex);
 616
 617         switch (READ_ONCE(call->state)) {
 618         case RXRPC_CALL_CLIENT_RECV_REPLY:
 619         case RXRPC_CALL_SERVER_RECV_REQUEST:
 620         case RXRPC_CALL_SERVER_ACK_REQUEST:
 621                 ret = rxrpc_recvmsg_data(sock, call, NULL, iter,
 622                                          *_len, 0, &offset);
 623                 *_len -= offset;
 624                 if (ret < 0)
 625                         goto out;
 626
 627                 /* We can only reach here with a partially full buffer if we
 628                  * have reached the end of the data.  We must otherwise have a
 629                  * full buffer or have been given -EAGAIN.
 630                  */
 631                 if (ret == 1) {
 632                         if (iov_iter_count(iter) > 0)
 633                                 goto short_data;
 634                         if (!want_more)
 635                                 goto read_phase_complete;
 636                         ret = 0;
 637                         goto out;
 638                 }
 639
 640                 if (!want_more)
 641                         goto excess_data;
 642                 goto out;
 643
 644         case RXRPC_CALL_COMPLETE:
 645                 goto call_complete;
 646
 647         default:
 648                 ret = -EINPROGRESS;
 649                 goto out;
 650         }
 651
 652 read_phase_complete:
 653         ret = 1;
 654 out:
 655         rxrpc_transmit_ack_packets(call->peer->local);
 656         if (_service)
 657                 *_service = call->service_id;
 658         mutex_unlock(&call->user_mutex);
 659         _leave(" = %d [%zu,%d]", ret, iov_iter_count(iter), *_abort);
 660         return ret;
 661
 662 short_data:
 663         trace_rxrpc_rx_eproto(call, 0, tracepoint_string("short_data"));
 664         ret = -EBADMSG;
 665         goto out;
 666 excess_data:
 667         trace_rxrpc_rx_eproto(call, 0, tracepoint_string("excess_data"));
 668         ret = -EMSGSIZE;
 669         goto out;
 670 call_complete:
 671         *_abort = call->abort_code;
 672         ret = call->error;
 673         if (call->completion == RXRPC_CALL_SUCCEEDED) {
 674                 ret = 1;
 675                 if (iov_iter_count(iter) > 0)
 676                         ret = -ECONNRESET;
 677         }
 678         goto out;
 679 }
 680 EXPORT_SYMBOL(rxrpc_kernel_recv_data);