1 // SPDX-License-Identifier: GPL-2.0-or-later 2 /* Maintain an RxRPC server socket to do AFS communications through 3 * 4 * Copyright (C) 2007 Red Hat, Inc. All Rights Reserved. 5 * Written by David Howells (dhowells@redhat.com) 6 */ 7 8 #include <linux/slab.h> 9 #include <linux/sched/signal.h> 10 11 #include <net/sock.h> 12 #include <net/af_rxrpc.h> 13 #include "internal.h" 14 #include "afs_cm.h" 15 #include "protocol_yfs.h" 16 #define RXRPC_TRACE_ONLY_DEFINE_ENUMS 17 #include <trace/events/rxrpc.h> 18 19 struct workqueue_struct *afs_async_calls; 20 21 static void afs_deferred_free_worker(struct work_struct *work); 22 static void afs_wake_up_call_waiter(struct sock *, struct rxrpc_call *, unsigned long); 23 static void afs_wake_up_async_call(struct sock *, struct rxrpc_call *, unsigned long); 24 static void afs_process_async_call(struct work_struct *); 25 static void afs_rx_new_call(struct sock *, struct rxrpc_call *, unsigned long); 26 static void afs_rx_discard_new_call(struct rxrpc_call *, unsigned long); 27 static int afs_deliver_cm_op_id(struct afs_call *); 28 29 /* asynchronous incoming call initial processing */ 30 static const struct afs_call_type afs_RXCMxxxx = { 31 .name = "CB.xxxx", 32 .deliver = afs_deliver_cm_op_id, 33 }; 34 35 /* 36 * open an RxRPC socket and bind it to be a server for callback notifications 37 * - the socket is left in blocking mode and non-blocking ops use MSG_DONTWAIT 38 */ 39 int afs_open_socket(struct afs_net *net) 40 { 41 struct sockaddr_rxrpc srx; 42 struct socket *socket; 43 int ret; 44 45 _enter(""); 46 47 ret = sock_create_kern(net->net, AF_RXRPC, SOCK_DGRAM, PF_INET6, &socket); 48 if (ret < 0) 49 goto error_1; 50 51 socket->sk->sk_allocation = GFP_NOFS; 52 53 /* bind the callback manager's address to make this a server socket */ 54 memset(&srx, 0, sizeof(srx)); 55 srx.srx_family = AF_RXRPC; 56 srx.srx_service = CM_SERVICE; 57 srx.transport_type = SOCK_DGRAM; 58 srx.transport_len = sizeof(srx.transport.sin6); 59 srx.transport.sin6.sin6_family = AF_INET6; 60 srx.transport.sin6.sin6_port = htons(AFS_CM_PORT); 61 62 ret = rxrpc_sock_set_min_security_level(socket->sk, 63 RXRPC_SECURITY_ENCRYPT); 64 if (ret < 0) 65 goto error_2; 66 67 ret = kernel_bind(socket, (struct sockaddr *) &srx, sizeof(srx)); 68 if (ret == -EADDRINUSE) { 69 srx.transport.sin6.sin6_port = 0; 70 ret = kernel_bind(socket, (struct sockaddr *) &srx, sizeof(srx)); 71 } 72 if (ret < 0) 73 goto error_2; 74 75 srx.srx_service = YFS_CM_SERVICE; 76 ret = kernel_bind(socket, (struct sockaddr *) &srx, sizeof(srx)); 77 if (ret < 0) 78 goto error_2; 79 80 /* Ideally, we'd turn on service upgrade here, but we can't because 81 * OpenAFS is buggy and leaks the userStatus field from packet to 82 * packet and between FS packets and CB packets - so if we try to do an 83 * upgrade on an FS packet, OpenAFS will leak that into the CB packet 84 * it sends back to us. 85 */ 86 87 rxrpc_kernel_new_call_notification(socket, afs_rx_new_call, 88 afs_rx_discard_new_call); 89 90 ret = kernel_listen(socket, INT_MAX); 91 if (ret < 0) 92 goto error_2; 93 94 net->socket = socket; 95 afs_charge_preallocation(&net->charge_preallocation_work); 96 _leave(" = 0"); 97 return 0; 98 99 error_2: 100 sock_release(socket); 101 error_1: 102 _leave(" = %d", ret); 103 return ret; 104 } 105 106 /* 107 * close the RxRPC socket AFS was using 108 */ 109 void afs_close_socket(struct afs_net *net) 110 { 111 _enter(""); 112 113 kernel_listen(net->socket, 0); 114 flush_workqueue(afs_async_calls); 115 116 if (net->spare_incoming_call) { 117 afs_put_call(net->spare_incoming_call); 118 net->spare_incoming_call = NULL; 119 } 120 121 _debug("outstanding %u", atomic_read(&net->nr_outstanding_calls)); 122 wait_var_event(&net->nr_outstanding_calls, 123 !atomic_read(&net->nr_outstanding_calls)); 124 _debug("no outstanding calls"); 125 126 kernel_sock_shutdown(net->socket, SHUT_RDWR); 127 flush_workqueue(afs_async_calls); 128 sock_release(net->socket); 129 130 _debug("dework"); 131 _leave(""); 132 } 133 134 /* 135 * Allocate a call. 136 */ 137 static struct afs_call *afs_alloc_call(struct afs_net *net, 138 const struct afs_call_type *type, 139 gfp_t gfp) 140 { 141 struct afs_call *call; 142 int o; 143 144 call = kzalloc(sizeof(*call), gfp); 145 if (!call) 146 return NULL; 147 148 call->type = type; 149 call->net = net; 150 call->debug_id = atomic_inc_return(&rxrpc_debug_id); 151 refcount_set(&call->ref, 1); 152 INIT_WORK(&call->async_work, afs_process_async_call); 153 INIT_WORK(&call->free_work, afs_deferred_free_worker); 154 init_waitqueue_head(&call->waitq); 155 spin_lock_init(&call->state_lock); 156 call->iter = &call->def_iter; 157 158 o = atomic_inc_return(&net->nr_outstanding_calls); 159 trace_afs_call(call->debug_id, afs_call_trace_alloc, 1, o, 160 __builtin_return_address(0)); 161 return call; 162 } 163 164 static void afs_free_call(struct afs_call *call) 165 { 166 struct afs_net *net = call->net; 167 int o; 168 169 ASSERT(!work_pending(&call->async_work)); 170 171 rxrpc_kernel_put_peer(call->peer); 172 173 if (call->rxcall) { 174 rxrpc_kernel_shutdown_call(net->socket, call->rxcall); 175 rxrpc_kernel_put_call(net->socket, call->rxcall); 176 call->rxcall = NULL; 177 } 178 if (call->type->destructor) 179 call->type->destructor(call); 180 181 afs_unuse_server_notime(call->net, call->server, afs_server_trace_put_call); 182 kfree(call->request); 183 184 o = atomic_read(&net->nr_outstanding_calls); 185 trace_afs_call(call->debug_id, afs_call_trace_free, 0, o, 186 __builtin_return_address(0)); 187 kfree(call); 188 189 o = atomic_dec_return(&net->nr_outstanding_calls); 190 if (o == 0) 191 wake_up_var(&net->nr_outstanding_calls); 192 } 193 194 /* 195 * Dispose of a reference on a call. 196 */ 197 void afs_put_call(struct afs_call *call) 198 { 199 struct afs_net *net = call->net; 200 unsigned int debug_id = call->debug_id; 201 bool zero; 202 int r, o; 203 204 zero = __refcount_dec_and_test(&call->ref, &r); 205 o = atomic_read(&net->nr_outstanding_calls); 206 trace_afs_call(debug_id, afs_call_trace_put, r - 1, o, 207 __builtin_return_address(0)); 208 if (zero) 209 afs_free_call(call); 210 } 211 212 static void afs_deferred_free_worker(struct work_struct *work) 213 { 214 struct afs_call *call = container_of(work, struct afs_call, free_work); 215 216 afs_free_call(call); 217 } 218 219 /* 220 * Dispose of a reference on a call, deferring the cleanup to a workqueue 221 * to avoid lock recursion. 222 */ 223 void afs_deferred_put_call(struct afs_call *call) 224 { 225 struct afs_net *net = call->net; 226 unsigned int debug_id = call->debug_id; 227 bool zero; 228 int r, o; 229 230 zero = __refcount_dec_and_test(&call->ref, &r); 231 o = atomic_read(&net->nr_outstanding_calls); 232 trace_afs_call(debug_id, afs_call_trace_put, r - 1, o, 233 __builtin_return_address(0)); 234 if (zero) 235 schedule_work(&call->free_work); 236 } 237 238 static struct afs_call *afs_get_call(struct afs_call *call, 239 enum afs_call_trace why) 240 { 241 int r; 242 243 __refcount_inc(&call->ref, &r); 244 245 trace_afs_call(call->debug_id, why, r + 1, 246 atomic_read(&call->net->nr_outstanding_calls), 247 __builtin_return_address(0)); 248 return call; 249 } 250 251 /* 252 * Queue the call for actual work. 253 */ 254 static void afs_queue_call_work(struct afs_call *call) 255 { 256 if (call->type->work) { 257 INIT_WORK(&call->work, call->type->work); 258 259 afs_get_call(call, afs_call_trace_work); 260 if (!queue_work(afs_wq, &call->work)) 261 afs_put_call(call); 262 } 263 } 264 265 /* 266 * allocate a call with flat request and reply buffers 267 */ 268 struct afs_call *afs_alloc_flat_call(struct afs_net *net, 269 const struct afs_call_type *type, 270 size_t request_size, size_t reply_max) 271 { 272 struct afs_call *call; 273 274 call = afs_alloc_call(net, type, GFP_NOFS); 275 if (!call) 276 goto nomem_call; 277 278 if (request_size) { 279 call->request_size = request_size; 280 call->request = kmalloc(request_size, GFP_NOFS); 281 if (!call->request) 282 goto nomem_free; 283 } 284 285 if (reply_max) { 286 call->reply_max = reply_max; 287 call->buffer = kmalloc(reply_max, GFP_NOFS); 288 if (!call->buffer) 289 goto nomem_free; 290 } 291 292 afs_extract_to_buf(call, call->reply_max); 293 call->operation_ID = type->op; 294 init_waitqueue_head(&call->waitq); 295 return call; 296 297 nomem_free: 298 afs_put_call(call); 299 nomem_call: 300 return NULL; 301 } 302 303 /* 304 * clean up a call with flat buffer 305 */ 306 void afs_flat_call_destructor(struct afs_call *call) 307 { 308 _enter(""); 309 310 kfree(call->request); 311 call->request = NULL; 312 kfree(call->buffer); 313 call->buffer = NULL; 314 } 315 316 /* 317 * Advance the AFS call state when the RxRPC call ends the transmit phase. 318 */ 319 static void afs_notify_end_request_tx(struct sock *sock, 320 struct rxrpc_call *rxcall, 321 unsigned long call_user_ID) 322 { 323 struct afs_call *call = (struct afs_call *)call_user_ID; 324 325 afs_set_call_state(call, AFS_CALL_CL_REQUESTING, AFS_CALL_CL_AWAIT_REPLY); 326 } 327 328 /* 329 * Initiate a call and synchronously queue up the parameters for dispatch. Any 330 * error is stored into the call struct, which the caller must check for. 331 */ 332 void afs_make_call(struct afs_call *call, gfp_t gfp) 333 { 334 struct rxrpc_call *rxcall; 335 struct msghdr msg; 336 struct kvec iov[1]; 337 size_t len; 338 s64 tx_total_len; 339 int ret; 340 341 _enter(",{%pISp+%u},", rxrpc_kernel_remote_addr(call->peer), call->service_id); 342 343 ASSERT(call->type != NULL); 344 ASSERT(call->type->name != NULL); 345 346 _debug("____MAKE %p{%s,%x} [%d]____", 347 call, call->type->name, key_serial(call->key), 348 atomic_read(&call->net->nr_outstanding_calls)); 349 350 trace_afs_make_call(call); 351 352 /* Work out the length we're going to transmit. This is awkward for 353 * calls such as FS.StoreData where there's an extra injection of data 354 * after the initial fixed part. 355 */ 356 tx_total_len = call->request_size; 357 if (call->write_iter) 358 tx_total_len += iov_iter_count(call->write_iter); 359 360 /* If the call is going to be asynchronous, we need an extra ref for 361 * the call to hold itself so the caller need not hang on to its ref. 362 */ 363 if (call->async) { 364 afs_get_call(call, afs_call_trace_get); 365 call->drop_ref = true; 366 } 367 368 /* create a call */ 369 rxcall = rxrpc_kernel_begin_call(call->net->socket, call->peer, call->key, 370 (unsigned long)call, 371 tx_total_len, 372 call->max_lifespan, 373 gfp, 374 (call->async ? 375 afs_wake_up_async_call : 376 afs_wake_up_call_waiter), 377 call->service_id, 378 call->upgrade, 379 (call->intr ? RXRPC_PREINTERRUPTIBLE : 380 RXRPC_UNINTERRUPTIBLE), 381 call->debug_id); 382 if (IS_ERR(rxcall)) { 383 ret = PTR_ERR(rxcall); 384 call->error = ret; 385 goto error_kill_call; 386 } 387 388 call->rxcall = rxcall; 389 call->issue_time = ktime_get_real(); 390 391 /* send the request */ 392 iov[0].iov_base = call->request; 393 iov[0].iov_len = call->request_size; 394 395 msg.msg_name = NULL; 396 msg.msg_namelen = 0; 397 iov_iter_kvec(&msg.msg_iter, ITER_SOURCE, iov, 1, call->request_size); 398 msg.msg_control = NULL; 399 msg.msg_controllen = 0; 400 msg.msg_flags = MSG_WAITALL | (call->write_iter ? MSG_MORE : 0); 401 402 ret = rxrpc_kernel_send_data(call->net->socket, rxcall, 403 &msg, call->request_size, 404 afs_notify_end_request_tx); 405 if (ret < 0) 406 goto error_do_abort; 407 408 if (call->write_iter) { 409 msg.msg_iter = *call->write_iter; 410 msg.msg_flags &= ~MSG_MORE; 411 trace_afs_send_data(call, &msg); 412 413 ret = rxrpc_kernel_send_data(call->net->socket, 414 call->rxcall, &msg, 415 iov_iter_count(&msg.msg_iter), 416 afs_notify_end_request_tx); 417 *call->write_iter = msg.msg_iter; 418 419 trace_afs_sent_data(call, &msg, ret); 420 if (ret < 0) 421 goto error_do_abort; 422 } 423 424 /* Note that at this point, we may have received the reply or an abort 425 * - and an asynchronous call may already have completed. 426 * 427 * afs_wait_for_call_to_complete(call) 428 * must be called to synchronously clean up. 429 */ 430 return; 431 432 error_do_abort: 433 if (ret != -ECONNABORTED) { 434 rxrpc_kernel_abort_call(call->net->socket, rxcall, 435 RX_USER_ABORT, ret, 436 afs_abort_send_data_error); 437 } else { 438 len = 0; 439 iov_iter_kvec(&msg.msg_iter, ITER_DEST, NULL, 0, 0); 440 rxrpc_kernel_recv_data(call->net->socket, rxcall, 441 &msg.msg_iter, &len, false, 442 &call->abort_code, &call->service_id); 443 call->responded = true; 444 } 445 call->error = ret; 446 trace_afs_call_done(call); 447 error_kill_call: 448 if (call->type->done) 449 call->type->done(call); 450 451 /* We need to dispose of the extra ref we grabbed for an async call. 452 * The call, however, might be queued on afs_async_calls and we need to 453 * make sure we don't get any more notifications that might requeue it. 454 */ 455 if (call->rxcall) 456 rxrpc_kernel_shutdown_call(call->net->socket, call->rxcall); 457 if (call->async) { 458 if (cancel_work_sync(&call->async_work)) 459 afs_put_call(call); 460 afs_set_call_complete(call, ret, 0); 461 } 462 463 call->error = ret; 464 call->state = AFS_CALL_COMPLETE; 465 _leave(" = %d", ret); 466 } 467 468 /* 469 * Log remote abort codes that indicate that we have a protocol disagreement 470 * with the server. 471 */ 472 static void afs_log_error(struct afs_call *call, s32 remote_abort) 473 { 474 static int max = 0; 475 const char *msg; 476 int m; 477 478 switch (remote_abort) { 479 case RX_EOF: msg = "unexpected EOF"; break; 480 case RXGEN_CC_MARSHAL: msg = "client marshalling"; break; 481 case RXGEN_CC_UNMARSHAL: msg = "client unmarshalling"; break; 482 case RXGEN_SS_MARSHAL: msg = "server marshalling"; break; 483 case RXGEN_SS_UNMARSHAL: msg = "server unmarshalling"; break; 484 case RXGEN_DECODE: msg = "opcode decode"; break; 485 case RXGEN_SS_XDRFREE: msg = "server XDR cleanup"; break; 486 case RXGEN_CC_XDRFREE: msg = "client XDR cleanup"; break; 487 case -32: msg = "insufficient data"; break; 488 default: 489 return; 490 } 491 492 m = max; 493 if (m < 3) { 494 max = m + 1; 495 pr_notice("kAFS: Peer reported %s failure on %s [%pISp]\n", 496 msg, call->type->name, 497 rxrpc_kernel_remote_addr(call->peer)); 498 } 499 } 500 501 /* 502 * deliver messages to a call 503 */ 504 static void afs_deliver_to_call(struct afs_call *call) 505 { 506 enum afs_call_state state; 507 size_t len; 508 u32 abort_code, remote_abort = 0; 509 int ret; 510 511 _enter("%s", call->type->name); 512 513 while (state = READ_ONCE(call->state), 514 state == AFS_CALL_CL_AWAIT_REPLY || 515 state == AFS_CALL_SV_AWAIT_OP_ID || 516 state == AFS_CALL_SV_AWAIT_REQUEST || 517 state == AFS_CALL_SV_AWAIT_ACK 518 ) { 519 if (state == AFS_CALL_SV_AWAIT_ACK) { 520 len = 0; 521 iov_iter_kvec(&call->def_iter, ITER_DEST, NULL, 0, 0); 522 ret = rxrpc_kernel_recv_data(call->net->socket, 523 call->rxcall, &call->def_iter, 524 &len, false, &remote_abort, 525 &call->service_id); 526 trace_afs_receive_data(call, &call->def_iter, false, ret); 527 528 if (ret == -EINPROGRESS || ret == -EAGAIN) 529 return; 530 if (ret < 0 || ret == 1) { 531 if (ret == 1) 532 ret = 0; 533 goto call_complete; 534 } 535 return; 536 } 537 538 ret = call->type->deliver(call); 539 state = READ_ONCE(call->state); 540 if (ret == 0 && call->unmarshalling_error) 541 ret = -EBADMSG; 542 switch (ret) { 543 case 0: 544 call->responded = true; 545 afs_queue_call_work(call); 546 if (state == AFS_CALL_CL_PROC_REPLY) { 547 if (call->op) 548 set_bit(AFS_SERVER_FL_MAY_HAVE_CB, 549 &call->op->server->flags); 550 goto call_complete; 551 } 552 ASSERTCMP(state, >, AFS_CALL_CL_PROC_REPLY); 553 goto done; 554 case -EINPROGRESS: 555 case -EAGAIN: 556 goto out; 557 case -ECONNABORTED: 558 ASSERTCMP(state, ==, AFS_CALL_COMPLETE); 559 call->responded = true; 560 afs_log_error(call, call->abort_code); 561 goto done; 562 case -ENOTSUPP: 563 call->responded = true; 564 abort_code = RXGEN_OPCODE; 565 rxrpc_kernel_abort_call(call->net->socket, call->rxcall, 566 abort_code, ret, 567 afs_abort_op_not_supported); 568 goto local_abort; 569 case -EIO: 570 pr_err("kAFS: Call %u in bad state %u\n", 571 call->debug_id, state); 572 fallthrough; 573 case -ENODATA: 574 case -EBADMSG: 575 case -EMSGSIZE: 576 case -ENOMEM: 577 case -EFAULT: 578 abort_code = RXGEN_CC_UNMARSHAL; 579 if (state != AFS_CALL_CL_AWAIT_REPLY) 580 abort_code = RXGEN_SS_UNMARSHAL; 581 rxrpc_kernel_abort_call(call->net->socket, call->rxcall, 582 abort_code, ret, 583 afs_abort_unmarshal_error); 584 goto local_abort; 585 default: 586 abort_code = RX_CALL_DEAD; 587 rxrpc_kernel_abort_call(call->net->socket, call->rxcall, 588 abort_code, ret, 589 afs_abort_general_error); 590 goto local_abort; 591 } 592 } 593 594 done: 595 if (call->type->done) 596 call->type->done(call); 597 out: 598 _leave(""); 599 return; 600 601 local_abort: 602 abort_code = 0; 603 call_complete: 604 afs_set_call_complete(call, ret, remote_abort); 605 state = AFS_CALL_COMPLETE; 606 goto done; 607 } 608 609 /* 610 * Wait synchronously for a call to complete. 611 */ 612 void afs_wait_for_call_to_complete(struct afs_call *call) 613 { 614 bool rxrpc_complete = false; 615 616 _enter(""); 617 618 if (!afs_check_call_state(call, AFS_CALL_COMPLETE)) { 619 DECLARE_WAITQUEUE(myself, current); 620 621 add_wait_queue(&call->waitq, &myself); 622 for (;;) { 623 set_current_state(TASK_UNINTERRUPTIBLE); 624 625 /* deliver any messages that are in the queue */ 626 if (!afs_check_call_state(call, AFS_CALL_COMPLETE) && 627 call->need_attention) { 628 call->need_attention = false; 629 __set_current_state(TASK_RUNNING); 630 afs_deliver_to_call(call); 631 continue; 632 } 633 634 if (afs_check_call_state(call, AFS_CALL_COMPLETE)) 635 break; 636 637 if (!rxrpc_kernel_check_life(call->net->socket, call->rxcall)) { 638 /* rxrpc terminated the call. */ 639 rxrpc_complete = true; 640 break; 641 } 642 643 schedule(); 644 } 645 646 remove_wait_queue(&call->waitq, &myself); 647 __set_current_state(TASK_RUNNING); 648 } 649 650 if (!afs_check_call_state(call, AFS_CALL_COMPLETE)) { 651 if (rxrpc_complete) { 652 afs_set_call_complete(call, call->error, call->abort_code); 653 } else { 654 /* Kill off the call if it's still live. */ 655 _debug("call interrupted"); 656 if (rxrpc_kernel_abort_call(call->net->socket, call->rxcall, 657 RX_USER_ABORT, -EINTR, 658 afs_abort_interrupted)) 659 afs_set_call_complete(call, -EINTR, 0); 660 } 661 } 662 } 663 664 /* 665 * wake up a waiting call 666 */ 667 static void afs_wake_up_call_waiter(struct sock *sk, struct rxrpc_call *rxcall, 668 unsigned long call_user_ID) 669 { 670 struct afs_call *call = (struct afs_call *)call_user_ID; 671 672 call->need_attention = true; 673 wake_up(&call->waitq); 674 } 675 676 /* 677 * Wake up an asynchronous call. The caller is holding the call notify 678 * spinlock around this, so we can't call afs_put_call(). 679 */ 680 static void afs_wake_up_async_call(struct sock *sk, struct rxrpc_call *rxcall, 681 unsigned long call_user_ID) 682 { 683 struct afs_call *call = (struct afs_call *)call_user_ID; 684 int r; 685 686 trace_afs_notify_call(rxcall, call); 687 call->need_attention = true; 688 689 if (__refcount_inc_not_zero(&call->ref, &r)) { 690 trace_afs_call(call->debug_id, afs_call_trace_wake, r + 1, 691 atomic_read(&call->net->nr_outstanding_calls), 692 __builtin_return_address(0)); 693 694 if (!queue_work(afs_async_calls, &call->async_work)) 695 afs_deferred_put_call(call); 696 } 697 } 698 699 /* 700 * Perform I/O processing on an asynchronous call. The work item carries a ref 701 * to the call struct that we either need to release or to pass on. 702 */ 703 static void afs_process_async_call(struct work_struct *work) 704 { 705 struct afs_call *call = container_of(work, struct afs_call, async_work); 706 707 _enter(""); 708 709 if (call->state < AFS_CALL_COMPLETE && call->need_attention) { 710 call->need_attention = false; 711 afs_deliver_to_call(call); 712 } 713 714 afs_put_call(call); 715 _leave(""); 716 } 717 718 static void afs_rx_attach(struct rxrpc_call *rxcall, unsigned long user_call_ID) 719 { 720 struct afs_call *call = (struct afs_call *)user_call_ID; 721 722 call->rxcall = rxcall; 723 } 724 725 /* 726 * Charge the incoming call preallocation. 727 */ 728 void afs_charge_preallocation(struct work_struct *work) 729 { 730 struct afs_net *net = 731 container_of(work, struct afs_net, charge_preallocation_work); 732 struct afs_call *call = net->spare_incoming_call; 733 734 for (;;) { 735 if (!call) { 736 call = afs_alloc_call(net, &afs_RXCMxxxx, GFP_KERNEL); 737 if (!call) 738 break; 739 740 call->drop_ref = true; 741 call->async = true; 742 call->state = AFS_CALL_SV_AWAIT_OP_ID; 743 init_waitqueue_head(&call->waitq); 744 afs_extract_to_tmp(call); 745 } 746 747 if (rxrpc_kernel_charge_accept(net->socket, 748 afs_wake_up_async_call, 749 afs_rx_attach, 750 (unsigned long)call, 751 GFP_KERNEL, 752 call->debug_id) < 0) 753 break; 754 call = NULL; 755 } 756 net->spare_incoming_call = call; 757 } 758 759 /* 760 * Discard a preallocated call when a socket is shut down. 761 */ 762 static void afs_rx_discard_new_call(struct rxrpc_call *rxcall, 763 unsigned long user_call_ID) 764 { 765 struct afs_call *call = (struct afs_call *)user_call_ID; 766 767 call->rxcall = NULL; 768 afs_put_call(call); 769 } 770 771 /* 772 * Notification of an incoming call. 773 */ 774 static void afs_rx_new_call(struct sock *sk, struct rxrpc_call *rxcall, 775 unsigned long user_call_ID) 776 { 777 struct afs_net *net = afs_sock2net(sk); 778 779 queue_work(afs_wq, &net->charge_preallocation_work); 780 } 781 782 /* 783 * Grab the operation ID from an incoming cache manager call. The socket 784 * buffer is discarded on error or if we don't yet have sufficient data. 785 */ 786 static int afs_deliver_cm_op_id(struct afs_call *call) 787 { 788 int ret; 789 790 _enter("{%zu}", iov_iter_count(call->iter)); 791 792 /* the operation ID forms the first four bytes of the request data */ 793 ret = afs_extract_data(call, true); 794 if (ret < 0) 795 return ret; 796 797 call->operation_ID = ntohl(call->tmp); 798 afs_set_call_state(call, AFS_CALL_SV_AWAIT_OP_ID, AFS_CALL_SV_AWAIT_REQUEST); 799 800 /* ask the cache manager to route the call (it'll change the call type 801 * if successful) */ 802 if (!afs_cm_incoming_call(call)) 803 return -ENOTSUPP; 804 805 trace_afs_cb_call(call); 806 807 /* pass responsibility for the remainer of this message off to the 808 * cache manager op */ 809 return call->type->deliver(call); 810 } 811 812 /* 813 * Advance the AFS call state when an RxRPC service call ends the transmit 814 * phase. 815 */ 816 static void afs_notify_end_reply_tx(struct sock *sock, 817 struct rxrpc_call *rxcall, 818 unsigned long call_user_ID) 819 { 820 struct afs_call *call = (struct afs_call *)call_user_ID; 821 822 afs_set_call_state(call, AFS_CALL_SV_REPLYING, AFS_CALL_SV_AWAIT_ACK); 823 } 824 825 /* 826 * send an empty reply 827 */ 828 void afs_send_empty_reply(struct afs_call *call) 829 { 830 struct afs_net *net = call->net; 831 struct msghdr msg; 832 833 _enter(""); 834 835 rxrpc_kernel_set_tx_length(net->socket, call->rxcall, 0); 836 837 msg.msg_name = NULL; 838 msg.msg_namelen = 0; 839 iov_iter_kvec(&msg.msg_iter, ITER_SOURCE, NULL, 0, 0); 840 msg.msg_control = NULL; 841 msg.msg_controllen = 0; 842 msg.msg_flags = 0; 843 844 switch (rxrpc_kernel_send_data(net->socket, call->rxcall, &msg, 0, 845 afs_notify_end_reply_tx)) { 846 case 0: 847 _leave(" [replied]"); 848 return; 849 850 case -ENOMEM: 851 _debug("oom"); 852 rxrpc_kernel_abort_call(net->socket, call->rxcall, 853 RXGEN_SS_MARSHAL, -ENOMEM, 854 afs_abort_oom); 855 fallthrough; 856 default: 857 _leave(" [error]"); 858 return; 859 } 860 } 861 862 /* 863 * send a simple reply 864 */ 865 void afs_send_simple_reply(struct afs_call *call, const void *buf, size_t len) 866 { 867 struct afs_net *net = call->net; 868 struct msghdr msg; 869 struct kvec iov[1]; 870 int n; 871 872 _enter(""); 873 874 rxrpc_kernel_set_tx_length(net->socket, call->rxcall, len); 875 876 iov[0].iov_base = (void *) buf; 877 iov[0].iov_len = len; 878 msg.msg_name = NULL; 879 msg.msg_namelen = 0; 880 iov_iter_kvec(&msg.msg_iter, ITER_SOURCE, iov, 1, len); 881 msg.msg_control = NULL; 882 msg.msg_controllen = 0; 883 msg.msg_flags = 0; 884 885 n = rxrpc_kernel_send_data(net->socket, call->rxcall, &msg, len, 886 afs_notify_end_reply_tx); 887 if (n >= 0) { 888 /* Success */ 889 _leave(" [replied]"); 890 return; 891 } 892 893 if (n == -ENOMEM) { 894 _debug("oom"); 895 rxrpc_kernel_abort_call(net->socket, call->rxcall, 896 RXGEN_SS_MARSHAL, -ENOMEM, 897 afs_abort_oom); 898 } 899 _leave(" [error]"); 900 } 901 902 /* 903 * Extract a piece of data from the received data socket buffers. 904 */ 905 int afs_extract_data(struct afs_call *call, bool want_more) 906 { 907 struct afs_net *net = call->net; 908 struct iov_iter *iter = call->iter; 909 enum afs_call_state state; 910 u32 remote_abort = 0; 911 int ret; 912 913 _enter("{%s,%zu,%zu},%d", 914 call->type->name, call->iov_len, iov_iter_count(iter), want_more); 915 916 ret = rxrpc_kernel_recv_data(net->socket, call->rxcall, iter, 917 &call->iov_len, want_more, &remote_abort, 918 &call->service_id); 919 trace_afs_receive_data(call, call->iter, want_more, ret); 920 if (ret == 0 || ret == -EAGAIN) 921 return ret; 922 923 state = READ_ONCE(call->state); 924 if (ret == 1) { 925 switch (state) { 926 case AFS_CALL_CL_AWAIT_REPLY: 927 afs_set_call_state(call, state, AFS_CALL_CL_PROC_REPLY); 928 break; 929 case AFS_CALL_SV_AWAIT_REQUEST: 930 afs_set_call_state(call, state, AFS_CALL_SV_REPLYING); 931 break; 932 case AFS_CALL_COMPLETE: 933 kdebug("prem complete %d", call->error); 934 return afs_io_error(call, afs_io_error_extract); 935 default: 936 break; 937 } 938 return 0; 939 } 940 941 afs_set_call_complete(call, ret, remote_abort); 942 return ret; 943 } 944 945 /* 946 * Log protocol error production. 947 */ 948 noinline int afs_protocol_error(struct afs_call *call, 949 enum afs_eproto_cause cause) 950 { 951 trace_afs_protocol_error(call, cause); 952 if (call) 953 call->unmarshalling_error = true; 954 return -EBADMSG; 955 } 956