projects/linux·blob master · refs

877 lines · 21.7 KiB

1// SPDX-License-Identifier: GPL-2.0-or-later
2/* AF_RXRPC sendmsg() implementation.
3 *
4 * Copyright (C) 2007, 2016 Red Hat, Inc. All Rights Reserved.
5 * Written by David Howells ([email protected])
6 */
7
8#define pr_fmt(fmt) KBUILD_MODNAME ": " fmt
9
10#include <linux/net.h>
11#include <linux/gfp.h>
12#include <linux/skbuff.h>
13#include <linux/export.h>
14#include <linux/sched/signal.h>
15
16#include <net/sock.h>
17#include <net/af_rxrpc.h>
18#include "ar-internal.h"
19
20/*
21 * Propose an abort to be made in the I/O thread.
22 */
23bool rxrpc_propose_abort(struct rxrpc_call *call, s32 abort_code, int error,
24 enum rxrpc_abort_reason why)
25{
26 _enter("{%d},%d,%d,%u", call->debug_id, abort_code, error, why);
27
28 if (!call->send_abort && !rxrpc_call_is_complete(call)) {
29 call->send_abort_why = why;
30 call->send_abort_err = error;
31 call->send_abort_seq = 0;
32 trace_rxrpc_abort_call(call, abort_code);
33 /* Request abort locklessly vs rxrpc_input_call_event(). */
34 smp_store_release(&call->send_abort, abort_code);
35 rxrpc_poke_call(call, rxrpc_call_poke_abort);
36 return true;
37 }
38
39 return false;
40}
41
42/*
43 * Wait for a call to become connected. Interruption here doesn't cause the
44 * call to be aborted.
45 */
46static int rxrpc_wait_to_be_connected(struct rxrpc_call *call, long *timeo)
47{
48 DECLARE_WAITQUEUE(myself, current);
49 int ret = 0;
50
51 _enter("%d", call->debug_id);
52
53 if (rxrpc_call_state(call) != RXRPC_CALL_CLIENT_AWAIT_CONN)
54 goto no_wait;
55
56 add_wait_queue_exclusive(&call->waitq, &myself);
57
58 for (;;) {
59 switch (call->interruptibility) {
60 case RXRPC_INTERRUPTIBLE:
61 case RXRPC_PREINTERRUPTIBLE:
62 set_current_state(TASK_INTERRUPTIBLE);
63 break;
64 case RXRPC_UNINTERRUPTIBLE:
65 default:
66 set_current_state(TASK_UNINTERRUPTIBLE);
67 break;
68 }
69
70 if (rxrpc_call_state(call) != RXRPC_CALL_CLIENT_AWAIT_CONN)
71 break;
72 if ((call->interruptibility == RXRPC_INTERRUPTIBLE ||
73 call->interruptibility == RXRPC_PREINTERRUPTIBLE) &&
74 signal_pending(current)) {
75 ret = sock_intr_errno(*timeo);
76 break;
77 }
78 *timeo = schedule_timeout(*timeo);
79 }
80
81 remove_wait_queue(&call->waitq, &myself);
82 __set_current_state(TASK_RUNNING);
83
84no_wait:
85 if (ret == 0 && rxrpc_call_is_complete(call))
86 ret = call->error;
87
88 _leave(" = %d", ret);
89 return ret;
90}
91
92/*
93 * Return true if there's sufficient Tx queue space.
94 */
95static bool rxrpc_check_tx_space(struct rxrpc_call *call, rxrpc_seq_t *_tx_win)
96{
97 rxrpc_seq_t tx_bottom = READ_ONCE(call->tx_bottom);
98
99 if (_tx_win)
100 *_tx_win = tx_bottom;
101 return call->send_top - tx_bottom < 256;
102}
103
104/*
105 * Wait for space to appear in the Tx queue or a signal to occur.
106 */
107static int rxrpc_wait_for_tx_window_intr(struct rxrpc_sock *rx,
108 struct rxrpc_call *call,
109 long *timeo)
110{
111 for (;;) {
112 set_current_state(TASK_INTERRUPTIBLE);
113 if (rxrpc_check_tx_space(call, NULL))
114 return 0;
115
116 if (rxrpc_call_is_complete(call))
117 return call->error;
118
119 if (signal_pending(current))
120 return sock_intr_errno(*timeo);
121
122 trace_rxrpc_txqueue(call, rxrpc_txqueue_wait);
123 *timeo = schedule_timeout(*timeo);
124 }
125}
126
127/*
128 * Wait for space to appear in the Tx queue uninterruptibly, but with
129 * a timeout of 2*RTT if no progress was made and a signal occurred.
130 */
131static int rxrpc_wait_for_tx_window_waitall(struct rxrpc_sock *rx,
132 struct rxrpc_call *call)
133{
134 rxrpc_seq_t tx_start, tx_win;
135 signed long rtt, timeout;
136
137 rtt = READ_ONCE(call->srtt_us) >> 3;
138 rtt = usecs_to_jiffies(rtt) * 2;
139 if (rtt < 2)
140 rtt = 2;
141
142 timeout = rtt;
143 tx_start = READ_ONCE(call->tx_bottom);
144
145 for (;;) {
146 set_current_state(TASK_UNINTERRUPTIBLE);
147
148 if (rxrpc_check_tx_space(call, &tx_win))
149 return 0;
150
151 if (rxrpc_call_is_complete(call))
152 return call->error;
153
154 if (timeout == 0 &&
155 tx_win == tx_start && signal_pending(current))
156 return -EINTR;
157
158 if (tx_win != tx_start) {
159 timeout = rtt;
160 tx_start = tx_win;
161 }
162
163 trace_rxrpc_txqueue(call, rxrpc_txqueue_wait);
164 timeout = schedule_timeout(timeout);
165 }
166}
167
168/*
169 * Wait for space to appear in the Tx queue uninterruptibly.
170 */
171static int rxrpc_wait_for_tx_window_nonintr(struct rxrpc_sock *rx,
172 struct rxrpc_call *call,
173 long *timeo)
174{
175 for (;;) {
176 set_current_state(TASK_UNINTERRUPTIBLE);
177 if (rxrpc_check_tx_space(call, NULL))
178 return 0;
179
180 if (rxrpc_call_is_complete(call))
181 return call->error;
182
183 trace_rxrpc_txqueue(call, rxrpc_txqueue_wait);
184 *timeo = schedule_timeout(*timeo);
185 }
186}
187
188/*
189 * wait for space to appear in the transmit/ACK window
190 * - caller holds the socket locked
191 */
192static int rxrpc_wait_for_tx_window(struct rxrpc_sock *rx,
193 struct rxrpc_call *call,
194 long *timeo,
195 bool waitall)
196{
197 DECLARE_WAITQUEUE(myself, current);
198 int ret;
199
200 _enter(",{%u,%u,%u}",
201 call->tx_bottom, call->tx_top, call->tx_winsize);
202
203 add_wait_queue(&call->waitq, &myself);
204
205 switch (call->interruptibility) {
206 case RXRPC_INTERRUPTIBLE:
207 if (waitall)
208 ret = rxrpc_wait_for_tx_window_waitall(rx, call);
209 else
210 ret = rxrpc_wait_for_tx_window_intr(rx, call, timeo);
211 break;
212 case RXRPC_PREINTERRUPTIBLE:
213 case RXRPC_UNINTERRUPTIBLE:
214 default:
215 ret = rxrpc_wait_for_tx_window_nonintr(rx, call, timeo);
216 break;
217 }
218
219 remove_wait_queue(&call->waitq, &myself);
220 set_current_state(TASK_RUNNING);
221 _leave(" = %d", ret);
222 return ret;
223}
224
225/*
226 * Notify the owner of the call that the transmit phase is ended and the last
227 * packet has been queued.
228 */
229static void rxrpc_notify_end_tx(struct rxrpc_sock *rx, struct rxrpc_call *call,
230 rxrpc_notify_end_tx_t notify_end_tx)
231{
232 if (notify_end_tx)
233 notify_end_tx(&rx->sk, call, call->user_call_ID);
234}
235
236/*
237 * Queue a DATA packet for transmission, set the resend timeout and send
238 * the packet immediately. Returns the error from rxrpc_send_data_packet()
239 * in case the caller wants to do something with it.
240 */
241static void rxrpc_queue_packet(struct rxrpc_sock *rx, struct rxrpc_call *call,
242 struct rxrpc_txbuf *txb,
243 rxrpc_notify_end_tx_t notify_end_tx)
244{
245 struct rxrpc_txqueue *sq = call->send_queue;
246 rxrpc_seq_t seq = txb->seq;
247 bool poke, last = txb->flags & RXRPC_LAST_PACKET;
248 int ix = seq & RXRPC_TXQ_MASK;
249 rxrpc_inc_stat(call->rxnet, stat_tx_data);
250
251 ASSERTCMP(txb->seq, ==, call->send_top + 1);
252
253 if (last)
254 trace_rxrpc_txqueue(call, rxrpc_txqueue_queue_last);
255 else
256 trace_rxrpc_txqueue(call, rxrpc_txqueue_queue);
257
258 if (WARN_ON_ONCE(sq->bufs[ix]))
259 trace_rxrpc_tq(call, sq, seq, rxrpc_tq_queue_dup);
260 else
261 trace_rxrpc_tq(call, sq, seq, rxrpc_tq_queue);
262
263 /* Add the packet to the call's output buffer */
264 poke = (READ_ONCE(call->tx_bottom) == call->send_top);
265 sq->bufs[ix] = txb;
266 /* Order send_top after the queue->next pointer and txb content. */
267 smp_store_release(&call->send_top, seq);
268 if (last) {
269 set_bit(RXRPC_CALL_TX_NO_MORE, &call->flags);
270 rxrpc_notify_end_tx(rx, call, notify_end_tx);
271 call->send_queue = NULL;
272 }
273
274 if (poke)
275 rxrpc_poke_call(call, rxrpc_call_poke_start);
276}
277
278/*
279 * Allocate a new txqueue unit and add it to the transmission queue.
280 */
281static int rxrpc_alloc_txqueue(struct sock *sk, struct rxrpc_call *call)
282{
283 struct rxrpc_txqueue *tq;
284
285 tq = kzalloc_obj(*tq, sk->sk_allocation);
286 if (!tq)
287 return -ENOMEM;
288
289 tq->xmit_ts_base = KTIME_MIN;
290 for (int i = 0; i < RXRPC_NR_TXQUEUE; i++)
291 tq->segment_xmit_ts[i] = UINT_MAX;
292
293 if (call->send_queue) {
294 tq->qbase = call->send_top + 1;
295 call->send_queue->next = tq;
296 call->send_queue = tq;
297 } else if (WARN_ON(call->tx_queue)) {
298 kfree(tq);
299 return -ENOMEM;
300 } else {
301 /* We start at seq 1, so pretend seq 0 is hard-acked. */
302 tq->nr_reported_acks = 1;
303 tq->segment_acked = 1UL;
304 tq->qbase = 0;
305 call->tx_qbase = 0;
306 call->send_queue = tq;
307 call->tx_qtail = tq;
308 call->tx_queue = tq;
309 }
310
311 trace_rxrpc_tq(call, tq, call->send_top, rxrpc_tq_alloc);
312 return 0;
313}
314
315/*
316 * send data through a socket
317 * - must be called in process context
318 * - The caller holds the call user access mutex, but not the socket lock.
319 */
320static int rxrpc_send_data(struct rxrpc_sock *rx,
321 struct rxrpc_call *call,
322 struct msghdr *msg, size_t len,
323 rxrpc_notify_end_tx_t notify_end_tx,
324 bool *_dropped_lock)
325{
326 struct rxrpc_txbuf *txb;
327 struct sock *sk = &rx->sk;
328 enum rxrpc_call_state state;
329 long timeo;
330 bool more = msg->msg_flags & MSG_MORE;
331 int ret, copied = 0;
332
333 if (test_bit(RXRPC_CALL_TX_NO_MORE, &call->flags)) {
334 trace_rxrpc_abort(call->debug_id, rxrpc_sendmsg_late_send,
335 call->cid, call->call_id, call->rx_consumed,
336 0, -EPROTO);
337 return -EPROTO;
338 }
339
340 timeo = sock_sndtimeo(sk, msg->msg_flags & MSG_DONTWAIT);
341
342 ret = rxrpc_wait_to_be_connected(call, &timeo);
343 if (ret < 0)
344 return ret;
345
346 if (call->conn->state == RXRPC_CONN_CLIENT_UNSECURED) {
347 ret = rxrpc_init_client_conn_security(call->conn);
348 if (ret < 0)
349 return ret;
350 }
351
352 /* this should be in poll */
353 sk_clear_bit(SOCKWQ_ASYNC_NOSPACE, sk);
354
355reload:
356 txb = call->tx_pending;
357 call->tx_pending = NULL;
358 if (txb)
359 rxrpc_see_txbuf(txb, rxrpc_txbuf_see_send_more);
360
361 ret = -EPIPE;
362 if (sk->sk_shutdown & SEND_SHUTDOWN)
363 goto maybe_error;
364 state = rxrpc_call_state(call);
365 ret = -ESHUTDOWN;
366 if (state >= RXRPC_CALL_COMPLETE)
367 goto maybe_error;
368 ret = -EPROTO;
369 if (state != RXRPC_CALL_CLIENT_PRE_SEND &&
370 state != RXRPC_CALL_CLIENT_SEND_REQUEST &&
371 state != RXRPC_CALL_SERVER_ACK_REQUEST &&
372 state != RXRPC_CALL_SERVER_SEND_REPLY) {
373 /* Request phase complete for this client call */
374 trace_rxrpc_abort(call->debug_id, rxrpc_sendmsg_late_send,
375 call->cid, call->call_id, call->rx_consumed,
376 0, -EPROTO);
377 goto maybe_error;
378 }
379
380 ret = -EMSGSIZE;
381 if (call->tx_total_len != -1) {
382 if (len - copied > call->tx_total_len)
383 goto maybe_error;
384 if (!more && len - copied != call->tx_total_len)
385 goto maybe_error;
386 }
387
388 do {
389 if (!txb) {
390 size_t remain;
391
392 _debug("alloc");
393
394 if (!rxrpc_check_tx_space(call, NULL))
395 goto wait_for_space;
396
397 /* See if we need to begin/extend the Tx queue. */
398 if (!call->send_queue || !((call->send_top + 1) & RXRPC_TXQ_MASK)) {
399 ret = rxrpc_alloc_txqueue(sk, call);
400 if (ret < 0)
401 goto maybe_error;
402 }
403
404 /* Work out the maximum size of a packet. Assume that
405 * the security header is going to be in the padded
406 * region (enc blocksize), but the trailer is not.
407 */
408 remain = more ? INT_MAX : msg_data_left(msg);
409 txb = call->conn->security->alloc_txbuf(call, remain, sk->sk_allocation);
410 if (!txb) {
411 ret = -ENOMEM;
412 goto maybe_error;
413 }
414 }
415
416 _debug("append");
417
418 /* append next segment of data to the current buffer */
419 if (msg_data_left(msg) > 0) {
420 size_t copy = umin(txb->space, msg_data_left(msg));
421
422 _debug("add %zu", copy);
423 if (!copy_from_iter_full(txb->data + txb->offset,
424 copy, &msg->msg_iter))
425 goto efault;
426 _debug("added");
427 txb->space -= copy;
428 txb->len += copy;
429 txb->offset += copy;
430 copied += copy;
431 if (call->tx_total_len != -1)
432 call->tx_total_len -= copy;
433 }
434
435 /* check for the far side aborting the call or a network error
436 * occurring */
437 if (rxrpc_call_is_complete(call))
438 goto call_terminated;
439
440 /* add the packet to the send queue if it's now full */
441 if (!txb->space ||
442 (msg_data_left(msg) == 0 && !more)) {
443 if (msg_data_left(msg) == 0 && !more)
444 txb->flags |= RXRPC_LAST_PACKET;
445
446 ret = call->security->secure_packet(call, txb);
447 if (ret < 0)
448 goto out;
449 rxrpc_queue_packet(rx, call, txb, notify_end_tx);
450 txb = NULL;
451 }
452 } while (msg_data_left(msg) > 0);
453
454success:
455 ret = copied;
456 if (rxrpc_call_is_complete(call) &&
457 call->error < 0)
458 ret = call->error;
459out:
460 call->tx_pending = txb;
461 _leave(" = %d", ret);
462 return ret;
463
464call_terminated:
465 rxrpc_put_txbuf(txb, rxrpc_txbuf_put_send_aborted);
466 _leave(" = %d", call->error);
467 return call->error;
468
469maybe_error:
470 if (copied)
471 goto success;
472 goto out;
473
474efault:
475 ret = -EFAULT;
476 goto out;
477
478wait_for_space:
479 ret = -EAGAIN;
480 if (msg->msg_flags & MSG_DONTWAIT)
481 goto maybe_error;
482 mutex_unlock(&call->user_mutex);
483 *_dropped_lock = true;
484 ret = rxrpc_wait_for_tx_window(rx, call, &timeo,
485 msg->msg_flags & MSG_WAITALL);
486 if (ret < 0)
487 goto maybe_error;
488 if (call->interruptibility == RXRPC_INTERRUPTIBLE) {
489 if (mutex_lock_interruptible(&call->user_mutex) < 0) {
490 ret = sock_intr_errno(timeo);
491 goto maybe_error;
492 }
493 } else {
494 mutex_lock(&call->user_mutex);
495 }
496 *_dropped_lock = false;
497 goto reload;
498}
499
500/*
501 * extract control messages from the sendmsg() control buffer
502 */
503static int rxrpc_sendmsg_cmsg(struct msghdr *msg, struct rxrpc_send_params *p)
504{
505 struct cmsghdr *cmsg;
506 bool got_user_ID = false;
507 int len;
508
509 if (msg->msg_controllen == 0)
510 return -EINVAL;
511
512 for_each_cmsghdr(cmsg, msg) {
513 if (!CMSG_OK(msg, cmsg))
514 return -EINVAL;
515
516 len = cmsg->cmsg_len - sizeof(struct cmsghdr);
517 _debug("CMSG %d, %d, %d",
518 cmsg->cmsg_level, cmsg->cmsg_type, len);
519
520 if (cmsg->cmsg_level != SOL_RXRPC)
521 continue;
522
523 switch (cmsg->cmsg_type) {
524 case RXRPC_USER_CALL_ID:
525 if (msg->msg_flags & MSG_CMSG_COMPAT) {
526 if (len != sizeof(u32))
527 return -EINVAL;
528 p->call.user_call_ID = *(u32 *)CMSG_DATA(cmsg);
529 } else {
530 if (len != sizeof(unsigned long))
531 return -EINVAL;
532 p->call.user_call_ID = *(unsigned long *)
533 CMSG_DATA(cmsg);
534 }
535 got_user_ID = true;
536 break;
537
538 case RXRPC_ABORT:
539 if (p->command != RXRPC_CMD_SEND_DATA)
540 return -EINVAL;
541 p->command = RXRPC_CMD_SEND_ABORT;
542 if (len != sizeof(p->abort_code))
543 return -EINVAL;
544 p->abort_code = *(unsigned int *)CMSG_DATA(cmsg);
545 if (p->abort_code == 0)
546 return -EINVAL;
547 break;
548
549 case RXRPC_CHARGE_ACCEPT:
550 if (p->command != RXRPC_CMD_SEND_DATA)
551 return -EINVAL;
552 p->command = RXRPC_CMD_CHARGE_ACCEPT;
553 if (len != 0)
554 return -EINVAL;
555 break;
556
557 case RXRPC_EXCLUSIVE_CALL:
558 p->exclusive = true;
559 if (len != 0)
560 return -EINVAL;
561 break;
562
563 case RXRPC_UPGRADE_SERVICE:
564 p->upgrade = true;
565 if (len != 0)
566 return -EINVAL;
567 break;
568
569 case RXRPC_TX_LENGTH:
570 if (p->call.tx_total_len != -1 || len != sizeof(__s64))
571 return -EINVAL;
572 p->call.tx_total_len = *(__s64 *)CMSG_DATA(cmsg);
573 if (p->call.tx_total_len < 0)
574 return -EINVAL;
575 break;
576
577 case RXRPC_SET_CALL_TIMEOUT:
578 if (len & 3 || len < 4 || len > 12)
579 return -EINVAL;
580 memcpy(&p->call.timeouts, CMSG_DATA(cmsg), len);
581 p->call.nr_timeouts = len / 4;
582 if (p->call.timeouts.hard > INT_MAX / HZ)
583 return -ERANGE;
584 if (p->call.nr_timeouts >= 2 && p->call.timeouts.idle > 60 * 60 * 1000)
585 return -ERANGE;
586 if (p->call.nr_timeouts >= 3 && p->call.timeouts.normal > 60 * 60 * 1000)
587 return -ERANGE;
588 break;
589
590 default:
591 return -EINVAL;
592 }
593 }
594
595 if (!got_user_ID)
596 return -EINVAL;
597 if (p->call.tx_total_len != -1 && p->command != RXRPC_CMD_SEND_DATA)
598 return -EINVAL;
599 _leave(" = 0");
600 return 0;
601}
602
603/*
604 * Create a new client call for sendmsg().
605 * - Called with the socket lock held, which it must release.
606 * - If it returns a call, the call's lock will need releasing by the caller.
607 */
608static struct rxrpc_call *
609rxrpc_new_client_call_for_sendmsg(struct rxrpc_sock *rx, struct msghdr *msg,
610 struct rxrpc_send_params *p)
611 __releases(&rx->sk.sk_lock)
612 __acquires(&call->user_mutex)
613{
614 struct rxrpc_conn_parameters cp;
615 struct rxrpc_peer *peer;
616 struct rxrpc_call *call;
617 struct key *key;
618
619 DECLARE_SOCKADDR(struct sockaddr_rxrpc *, srx, msg->msg_name);
620
621 _enter("");
622
623 if (!msg->msg_name) {
624 release_sock(&rx->sk);
625 return ERR_PTR(-EDESTADDRREQ);
626 }
627
628 peer = rxrpc_lookup_peer(rx->local, srx, GFP_KERNEL);
629 if (!peer) {
630 release_sock(&rx->sk);
631 return ERR_PTR(-ENOMEM);
632 }
633
634 key = rx->key;
635 if (key && !rx->key->payload.data[0])
636 key = NULL;
637
638 memset(&cp, 0, sizeof(cp));
639 cp.local = rx->local;
640 cp.peer = peer;
641 cp.key = key;
642 cp.security_level = rx->min_sec_level;
643 cp.exclusive = rx->exclusive | p->exclusive;
644 cp.upgrade = p->upgrade;
645 cp.service_id = srx->srx_service;
646 call = rxrpc_new_client_call(rx, &cp, &p->call, GFP_KERNEL,
647 atomic_inc_return(&rxrpc_debug_id));
648 /* The socket is now unlocked */
649
650 rxrpc_put_peer(peer, rxrpc_peer_put_application);
651 _leave(" = %p\n", call);
652 return call;
653}
654
655/*
656 * send a message forming part of a client call through an RxRPC socket
657 * - caller holds the socket locked
658 * - the socket may be either a client socket or a server socket
659 */
660int rxrpc_do_sendmsg(struct rxrpc_sock *rx, struct msghdr *msg, size_t len)
661{
662 struct rxrpc_call *call;
663 bool dropped_lock = false;
664 int ret;
665
666 struct rxrpc_send_params p = {
667 .call.tx_total_len = -1,
668 .call.user_call_ID = 0,
669 .call.nr_timeouts = 0,
670 .call.interruptibility = RXRPC_INTERRUPTIBLE,
671 .abort_code = 0,
672 .command = RXRPC_CMD_SEND_DATA,
673 .exclusive = false,
674 .upgrade = false,
675 };
676
677 _enter("");
678
679 ret = rxrpc_sendmsg_cmsg(msg, &p);
680 if (ret < 0)
681 goto error_release_sock;
682
683 if (p.command == RXRPC_CMD_CHARGE_ACCEPT) {
684 ret = -EINVAL;
685 if (rx->sk.sk_state != RXRPC_SERVER_LISTENING)
686 goto error_release_sock;
687 ret = rxrpc_user_charge_accept(rx, p.call.user_call_ID);
688 goto error_release_sock;
689 }
690
691 call = rxrpc_find_call_by_user_ID(rx, p.call.user_call_ID);
692 if (!call) {
693 ret = -EBADSLT;
694 if (p.command != RXRPC_CMD_SEND_DATA)
695 goto error_release_sock;
696 call = rxrpc_new_client_call_for_sendmsg(rx, msg, &p);
697 /* The socket is now unlocked... */
698 if (IS_ERR(call))
699 return PTR_ERR(call);
700 /* ... and we have the call lock. */
701 p.call.nr_timeouts = 0;
702 ret = 0;
703 if (rxrpc_call_is_complete(call))
704 goto out_put_unlock;
705 } else {
706 switch (rxrpc_call_state(call)) {
707 case RXRPC_CALL_CLIENT_AWAIT_CONN:
708 case RXRPC_CALL_SERVER_RECV_REQUEST:
709 if (p.command == RXRPC_CMD_SEND_ABORT)
710 break;
711 fallthrough;
712 case RXRPC_CALL_UNINITIALISED:
713 case RXRPC_CALL_SERVER_PREALLOC:
714 rxrpc_put_call(call, rxrpc_call_put_sendmsg);
715 ret = -EBUSY;
716 goto error_release_sock;
717 default:
718 break;
719 }
720
721 ret = mutex_lock_interruptible(&call->user_mutex);
722 release_sock(&rx->sk);
723 if (ret < 0) {
724 ret = -ERESTARTSYS;
725 goto error_put;
726 }
727
728 if (p.call.tx_total_len != -1) {
729 ret = -EINVAL;
730 if (call->tx_total_len != -1 ||
731 call->tx_pending ||
732 call->tx_top != 0)
733 goto out_put_unlock;
734 call->tx_total_len = p.call.tx_total_len;
735 }
736 }
737
738 switch (p.call.nr_timeouts) {
739 case 3:
740 WRITE_ONCE(call->next_rx_timo, p.call.timeouts.normal);
741 fallthrough;
742 case 2:
743 WRITE_ONCE(call->next_req_timo, p.call.timeouts.idle);
744 fallthrough;
745 case 1:
746 if (p.call.timeouts.hard > 0) {
747 ktime_t delay = ms_to_ktime(p.call.timeouts.hard * MSEC_PER_SEC);
748
749 WRITE_ONCE(call->expect_term_by,
750 ktime_add(p.call.timeouts.hard,
751 ktime_get_real()));
752 trace_rxrpc_timer_set(call, delay, rxrpc_timer_trace_hard);
753 rxrpc_poke_call(call, rxrpc_call_poke_set_timeout);
754
755 }
756 break;
757 }
758
759 if (rxrpc_call_is_complete(call)) {
760 /* it's too late for this call */
761 ret = -ESHUTDOWN;
762 goto out_put_unlock;
763 }
764
765 switch (p.command) {
766 case RXRPC_CMD_SEND_ABORT:
767 rxrpc_propose_abort(call, p.abort_code, -ECONNABORTED,
768 rxrpc_abort_call_sendmsg);
769 ret = 0;
770 break;
771 case RXRPC_CMD_SEND_DATA:
772 ret = rxrpc_send_data(rx, call, msg, len, NULL, &dropped_lock);
773 break;
774 default:
775 ret = -EINVAL;
776 break;
777 }
778
779out_put_unlock:
780 if (!dropped_lock)
781 mutex_unlock(&call->user_mutex);
782error_put:
783 rxrpc_put_call(call, rxrpc_call_put_sendmsg);
784 _leave(" = %d", ret);
785 return ret;
786
787error_release_sock:
788 release_sock(&rx->sk);
789 return ret;
790}
791
792/**
793 * rxrpc_kernel_send_data - Allow a kernel service to send data on a call
794 * @sock: The socket the call is on
795 * @call: The call to send data through
796 * @msg: The data to send
797 * @len: The amount of data to send
798 * @notify_end_tx: Notification that the last packet is queued.
799 *
800 * Allow a kernel service to send data on a call. The call must be in an state
801 * appropriate to sending data. No control data should be supplied in @msg,
802 * nor should an address be supplied. MSG_MORE should be flagged if there's
803 * more data to come, otherwise this data will end the transmission phase.
804 *
805 * Return: %0 if successful and a negative error code otherwise.
806 */
807int rxrpc_kernel_send_data(struct socket *sock, struct rxrpc_call *call,
808 struct msghdr *msg, size_t len,
809 rxrpc_notify_end_tx_t notify_end_tx)
810{
811 bool dropped_lock = false;
812 int ret;
813
814 _enter("{%d},", call->debug_id);
815
816 ASSERTCMP(msg->msg_name, ==, NULL);
817 ASSERTCMP(msg->msg_control, ==, NULL);
818
819 mutex_lock(&call->user_mutex);
820
821 ret = rxrpc_send_data(rxrpc_sk(sock->sk), call, msg, len,
822 notify_end_tx, &dropped_lock);
823 if (ret == -ESHUTDOWN)
824 ret = call->error;
825
826 if (!dropped_lock)
827 mutex_unlock(&call->user_mutex);
828 _leave(" = %d", ret);
829 return ret;
830}
831EXPORT_SYMBOL(rxrpc_kernel_send_data);
832
833/**
834 * rxrpc_kernel_abort_call - Allow a kernel service to abort a call
835 * @sock: The socket the call is on
836 * @call: The call to be aborted
837 * @abort_code: The abort code to stick into the ABORT packet
838 * @error: Local error value
839 * @why: Indication as to why.
840 *
841 * Allow a kernel service to abort a call if it's still in an abortable state.
842 *
843 * Return: %true if the call was aborted, %false if it was already complete.
844 */
845bool rxrpc_kernel_abort_call(struct socket *sock, struct rxrpc_call *call,
846 u32 abort_code, int error, enum rxrpc_abort_reason why)
847{
848 bool aborted;
849
850 _enter("{%d},%d,%d,%u", call->debug_id, abort_code, error, why);
851
852 mutex_lock(&call->user_mutex);
853 aborted = rxrpc_propose_abort(call, abort_code, error, why);
854 mutex_unlock(&call->user_mutex);
855 return aborted;
856}
857EXPORT_SYMBOL(rxrpc_kernel_abort_call);
858
859/**
860 * rxrpc_kernel_set_tx_length - Set the total Tx length on a call
861 * @sock: The socket the call is on
862 * @call: The call to be informed
863 * @tx_total_len: The amount of data to be transmitted for this call
864 *
865 * Allow a kernel service to set the total transmit length on a call. This
866 * allows buffer-to-packet encrypt-and-copy to be performed.
867 *
868 * This function is primarily for use for setting the reply length since the
869 * request length can be set when beginning the call.
870 */
871void rxrpc_kernel_set_tx_length(struct socket *sock, struct rxrpc_call *call,
872 s64 tx_total_len)
873{
874 WARN_ON(call->tx_total_len != -1);
875 call->tx_total_len = tx_total_len;
876}
877EXPORT_SYMBOL(rxrpc_kernel_set_tx_length);