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