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 unsigned long flags; 28 29 _enter("%d", call->debug_id); 30 31 if (test_bit(RXRPC_CALL_RELEASED, &call->flags)) { 32 rxrpc_see_call(call, rxrpc_call_see_notify_released); 33 return; 34 } 35 36 rcu_read_lock(); 37 38 rx = rcu_dereference(call->socket); 39 sk = &rx->sk; 40 if (rx && sk->sk_state < RXRPC_CLOSE) { 41 if (call->notify_rx) { 42 spin_lock_irqsave(&call->notify_lock, flags); 43 call->notify_rx(sk, call, call->user_call_ID); 44 spin_unlock_irqrestore(&call->notify_lock, flags); 45 } else { 46 spin_lock_irqsave(&rx->recvmsg_lock, flags); 47 if (list_empty(&call->recvmsg_link)) { 48 rxrpc_get_call(call, rxrpc_call_get_notify_socket); 49 list_add_tail(&call->recvmsg_link, &rx->recvmsg_q); 50 } 51 spin_unlock_irqrestore(&rx->recvmsg_lock, flags); 52 53 if (!sock_flag(sk, SOCK_DEAD)) { 54 _debug("call %ps", sk->sk_data_ready); 55 sk->sk_data_ready(sk); 56 } 57 } 58 } 59 60 rcu_read_unlock(); 61 _leave(""); 62 } 63 64 /* 65 * Pass a call terminating message to userspace. 66 */ 67 static int rxrpc_recvmsg_term(struct rxrpc_call *call, struct msghdr *msg) 68 { 69 u32 tmp = 0; 70 int ret; 71 72 switch (call->completion) { 73 case RXRPC_CALL_SUCCEEDED: 74 ret = 0; 75 if (rxrpc_is_service_call(call)) 76 ret = put_cmsg(msg, SOL_RXRPC, RXRPC_ACK, 0, &tmp); 77 break; 78 case RXRPC_CALL_REMOTELY_ABORTED: 79 tmp = call->abort_code; 80 ret = put_cmsg(msg, SOL_RXRPC, RXRPC_ABORT, 4, &tmp); 81 break; 82 case RXRPC_CALL_LOCALLY_ABORTED: 83 tmp = call->abort_code; 84 ret = put_cmsg(msg, SOL_RXRPC, RXRPC_ABORT, 4, &tmp); 85 break; 86 case RXRPC_CALL_NETWORK_ERROR: 87 tmp = -call->error; 88 ret = put_cmsg(msg, SOL_RXRPC, RXRPC_NET_ERROR, 4, &tmp); 89 break; 90 case RXRPC_CALL_LOCAL_ERROR: 91 tmp = -call->error; 92 ret = put_cmsg(msg, SOL_RXRPC, RXRPC_LOCAL_ERROR, 4, &tmp); 93 break; 94 default: 95 pr_err("Invalid terminal call state %u\n", call->completion); 96 BUG(); 97 break; 98 } 99 100 trace_rxrpc_recvdata(call, rxrpc_recvmsg_terminal, 101 call->ackr_window - 1, 102 call->rx_pkt_offset, call->rx_pkt_len, ret); 103 return ret; 104 } 105 106 /* 107 * Discard a packet we've used up and advance the Rx window by one. 108 */ 109 static void rxrpc_rotate_rx_window(struct rxrpc_call *call) 110 { 111 struct rxrpc_skb_priv *sp; 112 struct sk_buff *skb; 113 rxrpc_serial_t serial; 114 rxrpc_seq_t old_consumed = call->rx_consumed, tseq; 115 bool last; 116 int acked; 117 118 _enter("%d", call->debug_id); 119 120 skb = skb_dequeue(&call->recvmsg_queue); 121 rxrpc_see_skb(skb, rxrpc_skb_see_rotate); 122 123 sp = rxrpc_skb(skb); 124 tseq = sp->hdr.seq; 125 serial = sp->hdr.serial; 126 last = sp->hdr.flags & RXRPC_LAST_PACKET; 127 128 /* Barrier against rxrpc_input_data(). */ 129 if (after(tseq, call->rx_consumed)) 130 smp_store_release(&call->rx_consumed, tseq); 131 132 rxrpc_free_skb(skb, rxrpc_skb_put_rotate); 133 134 trace_rxrpc_receive(call, last ? rxrpc_receive_rotate_last : rxrpc_receive_rotate, 135 serial, call->rx_consumed); 136 137 if (last) 138 set_bit(RXRPC_CALL_RECVMSG_READ_ALL, &call->flags); 139 140 /* Check to see if there's an ACK that needs sending. */ 141 acked = atomic_add_return(call->rx_consumed - old_consumed, 142 &call->ackr_nr_consumed); 143 if (acked > 8 && 144 !test_and_set_bit(RXRPC_CALL_RX_IS_IDLE, &call->flags)) 145 rxrpc_poke_call(call, rxrpc_call_poke_idle); 146 } 147 148 /* 149 * Decrypt and verify a DATA packet. The content of the packet is pulled out 150 * into a flat buffer rather than decrypting in place in the skbuff. This also 151 * has the advantage of aligning the buffer correctly for the crypto routines. 152 * 153 * We keep track of the sequence number of the packet currently decrypted into 154 * the buffer in ->rx_dec_seq. If MSG_PEEK is used and steps onto a new 155 * packet, subsequent recvmsg() calls will have to go back and re-decrypt the 156 * current packet. 157 */ 158 static int rxrpc_verify_data(struct rxrpc_call *call, struct sk_buff *skb) 159 { 160 struct rxrpc_skb_priv *sp = rxrpc_skb(skb); 161 int ret; 162 163 if (sp->len > call->rx_dec_bsize || !call->rx_dec_buffer) { 164 /* Make sure we can hold a 1412-byte jumbo subpacket and make 165 * sure that the buffer size is aligned to a crypto blocksize. 166 */ 167 size_t size = clamp(round_up(sp->len, 32), 2048, 65535); 168 void *buffer = krealloc(call->rx_dec_buffer, size, GFP_NOFS); 169 170 if (!buffer) 171 return -ENOMEM; 172 call->rx_dec_buffer = buffer; 173 call->rx_dec_bsize = size; 174 } 175 176 ret = -EFAULT; 177 if (skb_copy_bits(skb, sp->offset, call->rx_dec_buffer, sp->len) < 0) 178 goto err; 179 180 call->rx_dec_offset = 0; 181 call->rx_dec_len = sp->len; 182 call->rx_dec_seq = sp->hdr.seq; 183 ret = call->security->verify_packet(call, skb); 184 if (ret < 0) 185 goto err; 186 return 0; 187 188 err: 189 kfree(call->rx_dec_buffer); 190 call->rx_dec_buffer = NULL; 191 call->rx_dec_bsize = 0; 192 call->rx_dec_offset = 0; 193 call->rx_dec_len = 0; 194 return ret; 195 } 196 197 /* 198 * Transcribe a call's user ID to a control message. 199 */ 200 static int rxrpc_recvmsg_user_id(struct rxrpc_call *call, struct msghdr *msg, 201 int flags) 202 { 203 if (!test_bit(RXRPC_CALL_HAS_USERID, &call->flags)) 204 return 0; 205 206 if (flags & MSG_CMSG_COMPAT) { 207 unsigned int id32 = call->user_call_ID; 208 209 return put_cmsg(msg, SOL_RXRPC, RXRPC_USER_CALL_ID, 210 sizeof(unsigned int), &id32); 211 } else { 212 unsigned long idl = call->user_call_ID; 213 214 return put_cmsg(msg, SOL_RXRPC, RXRPC_USER_CALL_ID, 215 sizeof(unsigned long), &idl); 216 } 217 } 218 219 /* 220 * Deal with a CHALLENGE packet. 221 */ 222 static int rxrpc_recvmsg_challenge(struct socket *sock, struct msghdr *msg, 223 struct sk_buff *challenge, unsigned int flags) 224 { 225 struct rxrpc_skb_priv *sp = rxrpc_skb(challenge); 226 struct rxrpc_connection *conn = sp->chall.conn; 227 228 return conn->security->challenge_to_recvmsg(conn, challenge, msg); 229 } 230 231 /* 232 * Process OOB packets. Called with the socket locked. 233 */ 234 static int rxrpc_recvmsg_oob(struct socket *sock, struct msghdr *msg, 235 unsigned int flags) 236 { 237 struct rxrpc_sock *rx = rxrpc_sk(sock->sk); 238 struct sk_buff *skb; 239 bool need_response = false; 240 int ret; 241 242 skb = skb_peek(&rx->recvmsg_oobq); 243 if (!skb) 244 return -EAGAIN; 245 rxrpc_see_skb(skb, rxrpc_skb_see_recvmsg); 246 247 ret = put_cmsg(msg, SOL_RXRPC, RXRPC_OOB_ID, sizeof(u64), 248 &skb->skb_mstamp_ns); 249 if (ret < 0) 250 return ret; 251 252 switch ((enum rxrpc_oob_type)skb->mark) { 253 case RXRPC_OOB_CHALLENGE: 254 need_response = true; 255 ret = rxrpc_recvmsg_challenge(sock, msg, skb, flags); 256 break; 257 default: 258 WARN_ONCE(1, "recvmsg() can't process unknown OOB type %u\n", 259 skb->mark); 260 ret = -EIO; 261 break; 262 } 263 264 if (!(flags & MSG_PEEK)) { 265 skb_unlink(skb, &rx->recvmsg_oobq); 266 if (need_response) 267 rxrpc_add_pending_oob(rx, skb); 268 else 269 rxrpc_free_skb(skb, rxrpc_skb_put_oob); 270 } 271 return ret; 272 } 273 274 /* 275 * Deliver messages to a call. This keeps processing packets until the buffer 276 * is filled and we find either more DATA (returns 0) or the end of the DATA 277 * (returns 1). If more packets are required, it returns -EAGAIN and if the 278 * call has failed it returns -EIO. 279 */ 280 static int rxrpc_recvmsg_data(struct socket *sock, struct rxrpc_call *call, 281 struct msghdr *msg, struct iov_iter *iter, 282 size_t len, int flags, size_t *_offset) 283 { 284 struct rxrpc_skb_priv *sp; 285 struct rxrpc_sock *rx = rxrpc_sk(sock->sk); 286 struct sk_buff *skb; 287 rxrpc_seq_t seq = 0; 288 size_t remain; 289 unsigned int rx_pkt_offset, rx_pkt_len; 290 int copy, ret = -EAGAIN, ret2; 291 292 rx_pkt_offset = call->rx_pkt_offset; 293 rx_pkt_len = call->rx_pkt_len; 294 295 if (rxrpc_call_has_failed(call)) { 296 seq = call->ackr_window - 1; 297 ret = -EIO; 298 goto done; 299 } 300 301 if (test_bit(RXRPC_CALL_RECVMSG_READ_ALL, &call->flags)) { 302 seq = call->ackr_window - 1; 303 ret = 1; 304 goto done; 305 } 306 307 /* No one else can be removing stuff from the queue, so we shouldn't 308 * need the Rx lock to walk it. 309 */ 310 skb = skb_peek(&call->recvmsg_queue); 311 while (skb) { 312 rxrpc_see_skb(skb, rxrpc_skb_see_recvmsg); 313 sp = rxrpc_skb(skb); 314 seq = sp->hdr.seq; 315 316 if (!(flags & MSG_PEEK)) 317 trace_rxrpc_receive(call, rxrpc_receive_front, 318 sp->hdr.serial, seq); 319 320 if (msg) 321 sock_recv_timestamp(msg, sock->sk, skb); 322 323 if (call->rx_dec_seq != sp->hdr.seq || 324 !call->rx_dec_buffer) { 325 ret2 = rxrpc_verify_data(call, skb); 326 trace_rxrpc_recvdata(call, rxrpc_recvmsg_next, seq, 327 call->rx_dec_offset, 328 call->rx_dec_len, ret2); 329 if (ret2 < 0) { 330 ret = ret2; 331 goto out; 332 } 333 } 334 335 if (rx_pkt_offset == USHRT_MAX) { 336 rx_pkt_offset = call->rx_dec_offset; 337 rx_pkt_len = call->rx_dec_len; 338 } else { 339 trace_rxrpc_recvdata(call, rxrpc_recvmsg_cont, seq, 340 rx_pkt_offset, rx_pkt_len, 0); 341 } 342 343 /* We have to handle short, empty and used-up DATA packets. */ 344 remain = len - *_offset; 345 copy = rx_pkt_len; 346 if (copy > remain) 347 copy = remain; 348 if (copy > 0) { 349 ret2 = copy_to_iter(call->rx_dec_buffer + rx_pkt_offset, 350 copy, iter); 351 if (ret2 != copy) { 352 ret = -EFAULT; 353 goto out; 354 } 355 356 /* handle piecemeal consumption of data packets */ 357 rx_pkt_offset += copy; 358 rx_pkt_len -= copy; 359 *_offset += copy; 360 } 361 362 if (rx_pkt_len > 0) { 363 trace_rxrpc_recvdata(call, rxrpc_recvmsg_full, seq, 364 rx_pkt_offset, rx_pkt_len, 0); 365 ASSERTCMP(*_offset, ==, len); 366 ret = 0; 367 break; 368 } 369 370 /* The whole packet has been transferred. */ 371 if (sp->hdr.flags & RXRPC_LAST_PACKET) 372 ret = 1; 373 rx_pkt_offset = USHRT_MAX; 374 rx_pkt_len = 0; 375 376 skb = skb_peek_next(skb, &call->recvmsg_queue); 377 378 if (!(flags & MSG_PEEK)) 379 rxrpc_rotate_rx_window(call); 380 381 if (!rx->app_ops && 382 !skb_queue_empty_lockless(&rx->recvmsg_oobq)) { 383 trace_rxrpc_recvdata(call, rxrpc_recvmsg_oobq, seq, 384 rx_pkt_offset, rx_pkt_len, ret); 385 break; 386 } 387 } 388 389 out: 390 if (!(flags & MSG_PEEK)) { 391 call->rx_pkt_offset = rx_pkt_offset; 392 call->rx_pkt_len = rx_pkt_len; 393 } 394 395 done: 396 trace_rxrpc_recvdata(call, rxrpc_recvmsg_data_return, seq, 397 rx_pkt_offset, rx_pkt_len, ret); 398 if (ret == -EAGAIN) 399 set_bit(RXRPC_CALL_RX_IS_IDLE, &call->flags); 400 return ret; 401 } 402 403 /* 404 * Receive a message from an RxRPC socket 405 * - we need to be careful about two or more threads calling recvmsg 406 * simultaneously 407 */ 408 int rxrpc_recvmsg(struct socket *sock, struct msghdr *msg, size_t len, 409 int flags) 410 { 411 struct rxrpc_call *call; 412 struct rxrpc_sock *rx = rxrpc_sk(sock->sk); 413 struct list_head *l; 414 unsigned int call_debug_id = 0; 415 size_t copied = 0; 416 long timeo; 417 int ret; 418 419 DEFINE_WAIT(wait); 420 421 trace_rxrpc_recvmsg(0, rxrpc_recvmsg_enter, 0); 422 423 if (flags & (MSG_OOB | MSG_TRUNC)) 424 return -EOPNOTSUPP; 425 426 timeo = sock_rcvtimeo(&rx->sk, flags & MSG_DONTWAIT); 427 428 try_again: 429 lock_sock(&rx->sk); 430 431 /* Return immediately if a client socket has no outstanding calls */ 432 if (RB_EMPTY_ROOT(&rx->calls) && 433 list_empty(&rx->recvmsg_q) && 434 skb_queue_empty_lockless(&rx->recvmsg_oobq) && 435 rx->sk.sk_state != RXRPC_SERVER_LISTENING) { 436 release_sock(&rx->sk); 437 return -EAGAIN; 438 } 439 440 if (list_empty(&rx->recvmsg_q) && 441 skb_queue_empty_lockless(&rx->recvmsg_oobq)) { 442 ret = -EWOULDBLOCK; 443 if (timeo == 0) { 444 call = NULL; 445 goto error_no_call; 446 } 447 448 release_sock(&rx->sk); 449 450 /* Wait for something to happen */ 451 prepare_to_wait_exclusive(sk_sleep(&rx->sk), &wait, 452 TASK_INTERRUPTIBLE); 453 ret = sock_error(&rx->sk); 454 if (ret) 455 goto wait_error; 456 457 if (list_empty(&rx->recvmsg_q) && 458 skb_queue_empty_lockless(&rx->recvmsg_oobq)) { 459 if (signal_pending(current)) 460 goto wait_interrupted; 461 trace_rxrpc_recvmsg(0, rxrpc_recvmsg_wait, 0); 462 timeo = schedule_timeout(timeo); 463 } 464 finish_wait(sk_sleep(&rx->sk), &wait); 465 goto try_again; 466 } 467 468 /* Deal with OOB messages before we consider getting normal data. */ 469 if (!skb_queue_empty_lockless(&rx->recvmsg_oobq)) { 470 ret = rxrpc_recvmsg_oob(sock, msg, flags); 471 release_sock(&rx->sk); 472 if (ret == -EAGAIN) 473 goto try_again; 474 goto error_trace; 475 } 476 477 /* Find the next call and dequeue it if we're not just peeking. If we 478 * do dequeue it, that comes with a ref that we will need to release. 479 * We also want to weed out calls that got requeued whilst we were 480 * shovelling data out. 481 */ 482 spin_lock_irq(&rx->recvmsg_lock); 483 l = rx->recvmsg_q.next; 484 call = list_entry(l, struct rxrpc_call, recvmsg_link); 485 486 if (!rxrpc_call_is_complete(call) && 487 skb_queue_empty(&call->recvmsg_queue) && 488 skb_queue_empty(&rx->recvmsg_oobq)) { 489 list_del_init(&call->recvmsg_link); 490 spin_unlock_irq(&rx->recvmsg_lock); 491 release_sock(&rx->sk); 492 trace_rxrpc_recvmsg(call->debug_id, rxrpc_recvmsg_unqueue, 0); 493 rxrpc_put_call(call, rxrpc_call_put_recvmsg); 494 goto try_again; 495 } 496 497 rxrpc_see_call(call, rxrpc_call_see_recvmsg); 498 if (test_bit(RXRPC_CALL_RELEASED, &call->flags)) { 499 rxrpc_see_call(call, rxrpc_call_see_already_released); 500 list_del_init(&call->recvmsg_link); 501 spin_unlock_irq(&rx->recvmsg_lock); 502 release_sock(&rx->sk); 503 trace_rxrpc_recvmsg(call->debug_id, rxrpc_recvmsg_unqueue, 0); 504 rxrpc_put_call(call, rxrpc_call_put_recvmsg); 505 goto try_again; 506 } 507 if (!(flags & MSG_PEEK)) 508 list_del_init(&call->recvmsg_link); 509 else 510 rxrpc_get_call(call, rxrpc_call_get_recvmsg); 511 spin_unlock_irq(&rx->recvmsg_lock); 512 513 call_debug_id = call->debug_id; 514 trace_rxrpc_recvmsg(call_debug_id, rxrpc_recvmsg_dequeue, 0); 515 516 /* We're going to drop the socket lock, so we need to lock the call 517 * against interference by sendmsg. 518 */ 519 if (!mutex_trylock(&call->user_mutex)) { 520 ret = -EWOULDBLOCK; 521 if (flags & MSG_DONTWAIT) 522 goto error_requeue_call; 523 ret = -ERESTARTSYS; 524 if (mutex_lock_interruptible(&call->user_mutex) < 0) 525 goto error_requeue_call; 526 } 527 528 release_sock(&rx->sk); 529 530 if (test_bit(RXRPC_CALL_RELEASED, &call->flags)) { 531 rxrpc_see_call(call, rxrpc_call_see_already_released); 532 mutex_unlock(&call->user_mutex); 533 rxrpc_put_call(call, rxrpc_call_put_recvmsg); 534 goto try_again; 535 } 536 537 ret = rxrpc_recvmsg_user_id(call, msg, flags); 538 if (ret < 0) 539 goto error_unlock_call; 540 541 if (msg->msg_name && call->peer) { 542 size_t len = sizeof(call->dest_srx); 543 544 memcpy(msg->msg_name, &call->dest_srx, len); 545 msg->msg_namelen = len; 546 } 547 548 ret = rxrpc_recvmsg_data(sock, call, msg, &msg->msg_iter, len, 549 flags, &copied); 550 if (ret == -EAGAIN) 551 ret = 0; 552 if (ret == -EIO) 553 goto call_failed; 554 if (ret < 0) 555 goto error_unlock_call; 556 557 if (rxrpc_call_is_complete(call) && 558 skb_queue_empty(&call->recvmsg_queue)) 559 goto call_complete; 560 if (rxrpc_call_has_failed(call)) 561 goto call_failed; 562 563 if (!(flags & MSG_PEEK) && 564 !skb_queue_empty(&call->recvmsg_queue)) 565 rxrpc_notify_socket(call); 566 goto not_yet_complete; 567 568 call_failed: 569 rxrpc_purge_queue(&call->recvmsg_queue); 570 call_complete: 571 ret = rxrpc_recvmsg_term(call, msg); 572 if (ret < 0) 573 goto error_unlock_call; 574 if (!(flags & MSG_PEEK)) 575 rxrpc_release_call(rx, call); 576 msg->msg_flags |= MSG_EOR; 577 ret = 1; 578 579 not_yet_complete: 580 if (ret == 0) 581 msg->msg_flags |= MSG_MORE; 582 else 583 msg->msg_flags &= ~MSG_MORE; 584 ret = copied; 585 586 error_unlock_call: 587 mutex_unlock(&call->user_mutex); 588 rxrpc_put_call(call, rxrpc_call_put_recvmsg); 589 trace_rxrpc_recvmsg(call_debug_id, rxrpc_recvmsg_return, ret); 590 return ret; 591 592 error_requeue_call: 593 if (!(flags & MSG_PEEK)) { 594 spin_lock_irq(&rx->recvmsg_lock); 595 if (list_empty(&call->recvmsg_link)) { 596 list_add(&call->recvmsg_link, &rx->recvmsg_q); 597 rxrpc_see_call(call, rxrpc_call_see_recvmsg_requeue); 598 spin_unlock_irq(&rx->recvmsg_lock); 599 } else if (list_is_first(&call->recvmsg_link, &rx->recvmsg_q)) { 600 spin_unlock_irq(&rx->recvmsg_lock); 601 rxrpc_put_call(call, rxrpc_call_see_recvmsg_requeue_first); 602 } else { 603 list_move(&call->recvmsg_link, &rx->recvmsg_q); 604 spin_unlock_irq(&rx->recvmsg_lock); 605 rxrpc_put_call(call, rxrpc_call_see_recvmsg_requeue_move); 606 } 607 trace_rxrpc_recvmsg(call_debug_id, rxrpc_recvmsg_requeue, 0); 608 } else { 609 rxrpc_put_call(call, rxrpc_call_put_recvmsg_peek_nowait); 610 } 611 error_no_call: 612 release_sock(&rx->sk); 613 error_trace: 614 trace_rxrpc_recvmsg(call_debug_id, rxrpc_recvmsg_return, ret); 615 return ret; 616 617 wait_interrupted: 618 ret = sock_intr_errno(timeo); 619 wait_error: 620 finish_wait(sk_sleep(&rx->sk), &wait); 621 call = NULL; 622 goto error_trace; 623 } 624 625 /** 626 * rxrpc_kernel_recv_data - Allow a kernel service to receive data/info 627 * @sock: The socket that the call exists on 628 * @call: The call to send data through 629 * @iter: The buffer to receive into 630 * @_len: The amount of data we want to receive (decreased on return) 631 * @want_more: True if more data is expected to be read 632 * @_abort: Where the abort code is stored if -ECONNABORTED is returned 633 * @_service: Where to store the actual service ID (may be upgraded) 634 * 635 * Allow a kernel service to receive data and pick up information about the 636 * state of a call. Note that *@_abort should also be initialised to %0. 637 * 638 * Note that we may return %-EAGAIN to drain empty packets at the end 639 * of the data, even if we've already copied over the requested data. 640 * 641 * Return: %0 if got what was asked for and there's more available, %1 if we 642 * got what was asked for and we're at the end of the call, %2 if a service 643 * call received all of the request but is still in progress and %-EAGAIN if we 644 * need more data. A variety of other errors can be returned if the call 645 * completed with failure. 646 */ 647 int rxrpc_kernel_recv_data(struct socket *sock, struct rxrpc_call *call, 648 struct iov_iter *iter, size_t *_len, 649 bool want_more, u32 *_abort, u16 *_service) 650 { 651 size_t offset = 0; 652 int ret; 653 654 _enter("{%d},%zu,%d", call->debug_id, *_len, want_more); 655 656 mutex_lock(&call->user_mutex); 657 658 ret = rxrpc_recvmsg_data(sock, call, NULL, iter, *_len, 0, &offset); 659 *_len -= offset; 660 if (ret == -EIO) 661 goto call_failed; 662 if (ret < 0) 663 goto out; 664 665 /* We can only reach here with a partially full buffer if we have 666 * reached the end of the data. We must otherwise have a full buffer 667 * or have been given -EAGAIN. 668 */ 669 if (ret == 1) { 670 if (iov_iter_count(iter) > 0) 671 goto short_data; 672 if (!want_more) 673 goto read_phase_complete; 674 ret = 0; 675 goto out; 676 } 677 678 if (!want_more) 679 goto excess_data; 680 goto out; 681 682 read_phase_complete: 683 ret = 1; 684 if (rxrpc_is_service_call(call)) { 685 if (rxrpc_call_is_complete(call)) 686 goto call_failed; 687 ret = 2; 688 } 689 out: 690 if (_service) 691 *_service = call->dest_srx.srx_service; 692 mutex_unlock(&call->user_mutex); 693 _leave(" = %d [%zu,%d]", ret, iov_iter_count(iter), *_abort); 694 return ret; 695 696 short_data: 697 trace_rxrpc_abort(call->debug_id, rxrpc_recvmsg_short_data, 698 call->cid, call->call_id, call->rx_consumed, 699 0, -EBADMSG); 700 ret = -EBADMSG; 701 goto out; 702 excess_data: 703 trace_rxrpc_abort(call->debug_id, rxrpc_recvmsg_excess_data, 704 call->cid, call->call_id, call->rx_consumed, 705 0, -EMSGSIZE); 706 ret = -EMSGSIZE; 707 goto out; 708 call_failed: 709 *_abort = call->abort_code; 710 ret = call->error; 711 if (call->completion == RXRPC_CALL_SUCCEEDED) { 712 ret = 1; 713 if (iov_iter_count(iter) > 0) 714 ret = -ECONNRESET; 715 } 716 goto out; 717 } 718 EXPORT_SYMBOL(rxrpc_kernel_recv_data); 719