1 /* AF_RXRPC sendmsg() implementation.
3 * Copyright (C) 2007, 2016 Red Hat, Inc. All Rights Reserved.
4 * Written by David Howells (dhowells@redhat.com)
6 * This program is free software; you can redistribute it and/or
7 * modify it under the terms of the GNU General Public Licence
8 * as published by the Free Software Foundation; either version
9 * 2 of the Licence, or (at your option) any later version.
12 #define pr_fmt(fmt) KBUILD_MODNAME ": " fmt
14 #include <linux/net.h>
15 #include <linux/gfp.h>
16 #include <linux/skbuff.h>
17 #include <linux/export.h>
18 #include <linux/sched/signal.h>
21 #include <net/af_rxrpc.h>
22 #include "ar-internal.h"
25 RXRPC_CMD_SEND_DATA
, /* send data message */
26 RXRPC_CMD_SEND_ABORT
, /* request abort generation */
27 RXRPC_CMD_ACCEPT
, /* [server] accept incoming call */
28 RXRPC_CMD_REJECT_BUSY
, /* [server] reject a call as busy */
31 struct rxrpc_send_params
{
32 s64 tx_total_len
; /* Total Tx data length (if send data) */
33 unsigned long user_call_ID
; /* User's call ID */
34 u32 abort_code
; /* Abort code to Tx (if abort) */
35 enum rxrpc_command command
: 8; /* The command to implement */
36 bool exclusive
; /* Shared or exclusive call */
37 bool upgrade
; /* If the connection is upgradeable */
41 * Wait for space to appear in the Tx queue or a signal to occur.
43 static int rxrpc_wait_for_tx_window_intr(struct rxrpc_sock
*rx
,
44 struct rxrpc_call
*call
,
48 set_current_state(TASK_INTERRUPTIBLE
);
49 if (call
->tx_top
- call
->tx_hard_ack
<
50 min_t(unsigned int, call
->tx_winsize
,
51 call
->cong_cwnd
+ call
->cong_extra
))
54 if (call
->state
>= RXRPC_CALL_COMPLETE
)
57 if (signal_pending(current
))
58 return sock_intr_errno(*timeo
);
60 trace_rxrpc_transmit(call
, rxrpc_transmit_wait
);
61 mutex_unlock(&call
->user_mutex
);
62 *timeo
= schedule_timeout(*timeo
);
63 if (mutex_lock_interruptible(&call
->user_mutex
) < 0)
64 return sock_intr_errno(*timeo
);
69 * Wait for space to appear in the Tx queue uninterruptibly, but with
70 * a timeout of 2*RTT if no progress was made and a signal occurred.
72 static int rxrpc_wait_for_tx_window_nonintr(struct rxrpc_sock
*rx
,
73 struct rxrpc_call
*call
)
75 rxrpc_seq_t tx_start
, tx_win
;
76 signed long rtt2
, timeout
;
79 rtt
= READ_ONCE(call
->peer
->rtt
);
80 rtt2
= nsecs_to_jiffies64(rtt
) * 2;
85 tx_start
= READ_ONCE(call
->tx_hard_ack
);
88 set_current_state(TASK_UNINTERRUPTIBLE
);
90 tx_win
= READ_ONCE(call
->tx_hard_ack
);
91 if (call
->tx_top
- tx_win
<
92 min_t(unsigned int, call
->tx_winsize
,
93 call
->cong_cwnd
+ call
->cong_extra
))
96 if (call
->state
>= RXRPC_CALL_COMPLETE
)
100 tx_win
== tx_start
&& signal_pending(current
))
103 if (tx_win
!= tx_start
) {
108 trace_rxrpc_transmit(call
, rxrpc_transmit_wait
);
109 timeout
= schedule_timeout(timeout
);
114 * wait for space to appear in the transmit/ACK window
115 * - caller holds the socket locked
117 static int rxrpc_wait_for_tx_window(struct rxrpc_sock
*rx
,
118 struct rxrpc_call
*call
,
122 DECLARE_WAITQUEUE(myself
, current
);
125 _enter(",{%u,%u,%u}",
126 call
->tx_hard_ack
, call
->tx_top
, call
->tx_winsize
);
128 add_wait_queue(&call
->waitq
, &myself
);
131 ret
= rxrpc_wait_for_tx_window_nonintr(rx
, call
);
133 ret
= rxrpc_wait_for_tx_window_intr(rx
, call
, timeo
);
135 remove_wait_queue(&call
->waitq
, &myself
);
136 set_current_state(TASK_RUNNING
);
137 _leave(" = %d", ret
);
142 * Schedule an instant Tx resend.
144 static inline void rxrpc_instant_resend(struct rxrpc_call
*call
, int ix
)
146 spin_lock_bh(&call
->lock
);
148 if (call
->state
< RXRPC_CALL_COMPLETE
) {
149 call
->rxtx_annotations
[ix
] = RXRPC_TX_ANNO_RETRANS
;
150 if (!test_and_set_bit(RXRPC_CALL_EV_RESEND
, &call
->events
))
151 rxrpc_queue_call(call
);
154 spin_unlock_bh(&call
->lock
);
158 * Notify the owner of the call that the transmit phase is ended and the last
159 * packet has been queued.
161 static void rxrpc_notify_end_tx(struct rxrpc_sock
*rx
, struct rxrpc_call
*call
,
162 rxrpc_notify_end_tx_t notify_end_tx
)
165 notify_end_tx(&rx
->sk
, call
, call
->user_call_ID
);
169 * Queue a DATA packet for transmission, set the resend timeout and send the
172 static void rxrpc_queue_packet(struct rxrpc_sock
*rx
, struct rxrpc_call
*call
,
173 struct sk_buff
*skb
, bool last
,
174 rxrpc_notify_end_tx_t notify_end_tx
)
176 struct rxrpc_skb_priv
*sp
= rxrpc_skb(skb
);
177 rxrpc_seq_t seq
= sp
->hdr
.seq
;
179 u8 annotation
= RXRPC_TX_ANNO_UNACK
;
181 _net("queue skb %p [%d]", skb
, seq
);
183 ASSERTCMP(seq
, ==, call
->tx_top
+ 1);
186 annotation
|= RXRPC_TX_ANNO_LAST
;
187 set_bit(RXRPC_CALL_TX_LASTQ
, &call
->flags
);
190 /* We have to set the timestamp before queueing as the retransmit
191 * algorithm can see the packet as soon as we queue it.
193 skb
->tstamp
= ktime_get_real();
195 ix
= seq
& RXRPC_RXTX_BUFF_MASK
;
196 rxrpc_get_skb(skb
, rxrpc_skb_tx_got
);
197 call
->rxtx_annotations
[ix
] = annotation
;
199 call
->rxtx_buffer
[ix
] = skb
;
202 trace_rxrpc_transmit(call
, rxrpc_transmit_queue_last
);
204 trace_rxrpc_transmit(call
, rxrpc_transmit_queue
);
206 if (last
|| call
->state
== RXRPC_CALL_SERVER_ACK_REQUEST
) {
207 _debug("________awaiting reply/ACK__________");
208 write_lock_bh(&call
->state_lock
);
209 switch (call
->state
) {
210 case RXRPC_CALL_CLIENT_SEND_REQUEST
:
211 call
->state
= RXRPC_CALL_CLIENT_AWAIT_REPLY
;
212 rxrpc_notify_end_tx(rx
, call
, notify_end_tx
);
214 case RXRPC_CALL_SERVER_ACK_REQUEST
:
215 call
->state
= RXRPC_CALL_SERVER_SEND_REPLY
;
216 call
->ack_at
= call
->expire_at
;
217 if (call
->ackr_reason
== RXRPC_ACK_DELAY
)
218 call
->ackr_reason
= 0;
219 __rxrpc_set_timer(call
, rxrpc_timer_init_for_send_reply
,
224 case RXRPC_CALL_SERVER_SEND_REPLY
:
225 call
->state
= RXRPC_CALL_SERVER_AWAIT_ACK
;
226 rxrpc_notify_end_tx(rx
, call
, notify_end_tx
);
231 write_unlock_bh(&call
->state_lock
);
234 if (seq
== 1 && rxrpc_is_client_call(call
))
235 rxrpc_expose_client_call(call
);
237 ret
= rxrpc_send_data_packet(call
, skb
, false);
239 _debug("need instant resend %d", ret
);
240 rxrpc_instant_resend(call
, ix
);
242 ktime_t now
= ktime_get_real(), resend_at
;
244 resend_at
= ktime_add_ms(now
, rxrpc_resend_timeout
);
246 if (ktime_before(resend_at
, call
->resend_at
)) {
247 call
->resend_at
= resend_at
;
248 rxrpc_set_timer(call
, rxrpc_timer_set_for_send
, now
);
252 rxrpc_free_skb(skb
, rxrpc_skb_tx_freed
);
257 * send data through a socket
258 * - must be called in process context
259 * - The caller holds the call user access mutex, but not the socket lock.
261 static int rxrpc_send_data(struct rxrpc_sock
*rx
,
262 struct rxrpc_call
*call
,
263 struct msghdr
*msg
, size_t len
,
264 rxrpc_notify_end_tx_t notify_end_tx
)
266 struct rxrpc_skb_priv
*sp
;
268 struct sock
*sk
= &rx
->sk
;
273 timeo
= sock_sndtimeo(sk
, msg
->msg_flags
& MSG_DONTWAIT
);
275 /* this should be in poll */
276 sk_clear_bit(SOCKWQ_ASYNC_NOSPACE
, sk
);
278 if (sk
->sk_err
|| (sk
->sk_shutdown
& SEND_SHUTDOWN
))
281 more
= msg
->msg_flags
& MSG_MORE
;
283 if (call
->tx_total_len
!= -1) {
284 if (len
> call
->tx_total_len
)
286 if (!more
&& len
!= call
->tx_total_len
)
290 skb
= call
->tx_pending
;
291 call
->tx_pending
= NULL
;
292 rxrpc_see_skb(skb
, rxrpc_skb_tx_seen
);
296 /* Check to see if there's a ping ACK to reply to. */
297 if (call
->ackr_reason
== RXRPC_ACK_PING_RESPONSE
)
298 rxrpc_send_ack_packet(call
, false);
301 size_t size
, chunk
, max
, space
;
305 if (call
->tx_top
- call
->tx_hard_ack
>=
306 min_t(unsigned int, call
->tx_winsize
,
307 call
->cong_cwnd
+ call
->cong_extra
)) {
309 if (msg
->msg_flags
& MSG_DONTWAIT
)
311 ret
= rxrpc_wait_for_tx_window(rx
, call
,
313 msg
->msg_flags
& MSG_WAITALL
);
318 max
= RXRPC_JUMBO_DATALEN
;
319 max
-= call
->conn
->security_size
;
320 max
&= ~(call
->conn
->size_align
- 1UL);
323 if (chunk
> msg_data_left(msg
) && !more
)
324 chunk
= msg_data_left(msg
);
326 space
= chunk
+ call
->conn
->size_align
;
327 space
&= ~(call
->conn
->size_align
- 1UL);
329 size
= space
+ call
->conn
->security_size
;
331 _debug("SIZE: %zu/%zu/%zu", chunk
, space
, size
);
333 /* create a buffer that we can retain until it's ACK'd */
334 skb
= sock_alloc_send_skb(
335 sk
, size
, msg
->msg_flags
& MSG_DONTWAIT
, &ret
);
339 rxrpc_new_skb(skb
, rxrpc_skb_tx_new
);
341 _debug("ALLOC SEND %p", skb
);
343 ASSERTCMP(skb
->mark
, ==, 0);
345 _debug("HS: %u", call
->conn
->security_size
);
346 skb_reserve(skb
, call
->conn
->security_size
);
347 skb
->len
+= call
->conn
->security_size
;
351 if (sp
->remain
> skb_tailroom(skb
))
352 sp
->remain
= skb_tailroom(skb
);
354 _net("skb: hr %d, tr %d, hl %d, rm %d",
360 skb
->ip_summed
= CHECKSUM_UNNECESSARY
;
366 /* append next segment of data to the current buffer */
367 if (msg_data_left(msg
) > 0) {
368 int copy
= skb_tailroom(skb
);
369 ASSERTCMP(copy
, >, 0);
370 if (copy
> msg_data_left(msg
))
371 copy
= msg_data_left(msg
);
372 if (copy
> sp
->remain
)
376 ret
= skb_add_data(skb
, &msg
->msg_iter
, copy
);
383 if (call
->tx_total_len
!= -1)
384 call
->tx_total_len
-= copy
;
387 /* add the packet to the send queue if it's now full */
388 if (sp
->remain
<= 0 ||
389 (msg_data_left(msg
) == 0 && !more
)) {
390 struct rxrpc_connection
*conn
= call
->conn
;
394 /* pad out if we're using security */
395 if (conn
->security_ix
) {
396 pad
= conn
->security_size
+ skb
->mark
;
397 pad
= conn
->size_align
- pad
;
398 pad
&= conn
->size_align
- 1;
399 _debug("pad %zu", pad
);
401 skb_put_zero(skb
, pad
);
404 seq
= call
->tx_top
+ 1;
408 sp
->hdr
.flags
= conn
->out_clientflag
;
410 if (msg_data_left(msg
) == 0 && !more
)
411 sp
->hdr
.flags
|= RXRPC_LAST_PACKET
;
412 else if (call
->tx_top
- call
->tx_hard_ack
<
414 sp
->hdr
.flags
|= RXRPC_MORE_PACKETS
;
416 ret
= conn
->security
->secure_packet(
417 call
, skb
, skb
->mark
, skb
->head
);
421 rxrpc_queue_packet(rx
, call
, skb
,
422 !msg_data_left(msg
) && !more
,
427 /* Check for the far side aborting the call or a network error
428 * occurring. If this happens, save any packet that was under
429 * construction so that in the case of a network error, the
430 * call can be retried or redirected.
432 if (call
->state
== RXRPC_CALL_COMPLETE
) {
436 } while (msg_data_left(msg
) > 0);
441 call
->tx_pending
= skb
;
442 _leave(" = %d", ret
);
456 * extract control messages from the sendmsg() control buffer
458 static int rxrpc_sendmsg_cmsg(struct msghdr
*msg
, struct rxrpc_send_params
*p
)
460 struct cmsghdr
*cmsg
;
461 bool got_user_ID
= false;
464 if (msg
->msg_controllen
== 0)
467 for_each_cmsghdr(cmsg
, msg
) {
468 if (!CMSG_OK(msg
, cmsg
))
471 len
= cmsg
->cmsg_len
- sizeof(struct cmsghdr
);
472 _debug("CMSG %d, %d, %d",
473 cmsg
->cmsg_level
, cmsg
->cmsg_type
, len
);
475 if (cmsg
->cmsg_level
!= SOL_RXRPC
)
478 switch (cmsg
->cmsg_type
) {
479 case RXRPC_USER_CALL_ID
:
480 if (msg
->msg_flags
& MSG_CMSG_COMPAT
) {
481 if (len
!= sizeof(u32
))
483 p
->user_call_ID
= *(u32
*)CMSG_DATA(cmsg
);
485 if (len
!= sizeof(unsigned long))
487 p
->user_call_ID
= *(unsigned long *)
494 if (p
->command
!= RXRPC_CMD_SEND_DATA
)
496 p
->command
= RXRPC_CMD_SEND_ABORT
;
497 if (len
!= sizeof(p
->abort_code
))
499 p
->abort_code
= *(unsigned int *)CMSG_DATA(cmsg
);
500 if (p
->abort_code
== 0)
505 if (p
->command
!= RXRPC_CMD_SEND_DATA
)
507 p
->command
= RXRPC_CMD_ACCEPT
;
512 case RXRPC_EXCLUSIVE_CALL
:
518 case RXRPC_UPGRADE_SERVICE
:
524 case RXRPC_TX_LENGTH
:
525 if (p
->tx_total_len
!= -1 || len
!= sizeof(__s64
))
527 p
->tx_total_len
= *(__s64
*)CMSG_DATA(cmsg
);
528 if (p
->tx_total_len
< 0)
539 if (p
->tx_total_len
!= -1 && p
->command
!= RXRPC_CMD_SEND_DATA
)
546 * Create a new client call for sendmsg().
547 * - Called with the socket lock held, which it must release.
548 * - If it returns a call, the call's lock will need releasing by the caller.
550 static struct rxrpc_call
*
551 rxrpc_new_client_call_for_sendmsg(struct rxrpc_sock
*rx
, struct msghdr
*msg
,
552 struct rxrpc_send_params
*p
)
553 __releases(&rx
->sk
.sk_lock
.slock
)
555 struct rxrpc_conn_parameters cp
;
556 struct rxrpc_call
*call
;
559 DECLARE_SOCKADDR(struct sockaddr_rxrpc
*, srx
, msg
->msg_name
);
563 if (!msg
->msg_name
) {
564 release_sock(&rx
->sk
);
565 return ERR_PTR(-EDESTADDRREQ
);
569 if (key
&& !rx
->key
->payload
.data
[0])
572 memset(&cp
, 0, sizeof(cp
));
573 cp
.local
= rx
->local
;
575 cp
.security_level
= rx
->min_sec_level
;
576 cp
.exclusive
= rx
->exclusive
| p
->exclusive
;
577 cp
.upgrade
= p
->upgrade
;
578 cp
.service_id
= srx
->srx_service
;
579 call
= rxrpc_new_client_call(rx
, &cp
, srx
, p
->user_call_ID
,
580 p
->tx_total_len
, GFP_KERNEL
);
581 /* The socket is now unlocked */
583 _leave(" = %p\n", call
);
588 * send a message forming part of a client call through an RxRPC socket
589 * - caller holds the socket locked
590 * - the socket may be either a client socket or a server socket
592 int rxrpc_do_sendmsg(struct rxrpc_sock
*rx
, struct msghdr
*msg
, size_t len
)
593 __releases(&rx
->sk
.sk_lock
.slock
)
595 enum rxrpc_call_state state
;
596 struct rxrpc_call
*call
;
599 struct rxrpc_send_params p
= {
603 .command
= RXRPC_CMD_SEND_DATA
,
610 ret
= rxrpc_sendmsg_cmsg(msg
, &p
);
612 goto error_release_sock
;
614 if (p
.command
== RXRPC_CMD_ACCEPT
) {
616 if (rx
->sk
.sk_state
!= RXRPC_SERVER_LISTENING
)
617 goto error_release_sock
;
618 call
= rxrpc_accept_call(rx
, p
.user_call_ID
, NULL
);
619 /* The socket is now unlocked. */
621 return PTR_ERR(call
);
626 call
= rxrpc_find_call_by_user_ID(rx
, p
.user_call_ID
);
629 if (p
.command
!= RXRPC_CMD_SEND_DATA
)
630 goto error_release_sock
;
631 call
= rxrpc_new_client_call_for_sendmsg(rx
, msg
, &p
);
632 /* The socket is now unlocked... */
634 return PTR_ERR(call
);
635 /* ... and we have the call lock. */
637 switch (READ_ONCE(call
->state
)) {
638 case RXRPC_CALL_UNINITIALISED
:
639 case RXRPC_CALL_CLIENT_AWAIT_CONN
:
640 case RXRPC_CALL_SERVER_PREALLOC
:
641 case RXRPC_CALL_SERVER_SECURING
:
642 case RXRPC_CALL_SERVER_ACCEPTING
:
644 goto error_release_sock
;
649 ret
= mutex_lock_interruptible(&call
->user_mutex
);
650 release_sock(&rx
->sk
);
656 if (p
.tx_total_len
!= -1) {
658 if (call
->tx_total_len
!= -1 ||
662 call
->tx_total_len
= p
.tx_total_len
;
666 state
= READ_ONCE(call
->state
);
667 _debug("CALL %d USR %lx ST %d on CONN %p",
668 call
->debug_id
, call
->user_call_ID
, state
, call
->conn
);
670 if (state
>= RXRPC_CALL_COMPLETE
) {
671 /* it's too late for this call */
673 } else if (p
.command
== RXRPC_CMD_SEND_ABORT
) {
675 if (rxrpc_abort_call("CMD", call
, 0, p
.abort_code
, -ECONNABORTED
))
676 ret
= rxrpc_send_abort_packet(call
);
677 } else if (p
.command
!= RXRPC_CMD_SEND_DATA
) {
679 } else if (rxrpc_is_client_call(call
) &&
680 state
!= RXRPC_CALL_CLIENT_SEND_REQUEST
) {
681 /* request phase complete for this client call */
683 } else if (rxrpc_is_service_call(call
) &&
684 state
!= RXRPC_CALL_SERVER_ACK_REQUEST
&&
685 state
!= RXRPC_CALL_SERVER_SEND_REPLY
) {
686 /* Reply phase not begun or not complete for service call. */
689 ret
= rxrpc_send_data(rx
, call
, msg
, len
, NULL
);
693 mutex_unlock(&call
->user_mutex
);
695 rxrpc_put_call(call
, rxrpc_call_put
);
696 _leave(" = %d", ret
);
700 release_sock(&rx
->sk
);
705 * rxrpc_kernel_send_data - Allow a kernel service to send data on a call
706 * @sock: The socket the call is on
707 * @call: The call to send data through
708 * @msg: The data to send
709 * @len: The amount of data to send
710 * @notify_end_tx: Notification that the last packet is queued.
712 * Allow a kernel service to send data on a call. The call must be in an state
713 * appropriate to sending data. No control data should be supplied in @msg,
714 * nor should an address be supplied. MSG_MORE should be flagged if there's
715 * more data to come, otherwise this data will end the transmission phase.
717 int rxrpc_kernel_send_data(struct socket
*sock
, struct rxrpc_call
*call
,
718 struct msghdr
*msg
, size_t len
,
719 rxrpc_notify_end_tx_t notify_end_tx
)
723 _enter("{%d,%s},", call
->debug_id
, rxrpc_call_states
[call
->state
]);
725 ASSERTCMP(msg
->msg_name
, ==, NULL
);
726 ASSERTCMP(msg
->msg_control
, ==, NULL
);
728 mutex_lock(&call
->user_mutex
);
730 _debug("CALL %d USR %lx ST %d on CONN %p",
731 call
->debug_id
, call
->user_call_ID
, call
->state
, call
->conn
);
733 switch (READ_ONCE(call
->state
)) {
734 case RXRPC_CALL_CLIENT_SEND_REQUEST
:
735 case RXRPC_CALL_SERVER_ACK_REQUEST
:
736 case RXRPC_CALL_SERVER_SEND_REPLY
:
737 ret
= rxrpc_send_data(rxrpc_sk(sock
->sk
), call
, msg
, len
,
740 case RXRPC_CALL_COMPLETE
:
741 read_lock_bh(&call
->state_lock
);
743 read_unlock_bh(&call
->state_lock
);
746 /* Request phase complete for this client call */
747 trace_rxrpc_rx_eproto(call
, 0, tracepoint_string("late_send"));
752 mutex_unlock(&call
->user_mutex
);
753 _leave(" = %d", ret
);
756 EXPORT_SYMBOL(rxrpc_kernel_send_data
);
759 * rxrpc_kernel_abort_call - Allow a kernel service to abort a call
760 * @sock: The socket the call is on
761 * @call: The call to be aborted
762 * @abort_code: The abort code to stick into the ABORT packet
763 * @error: Local error value
764 * @why: 3-char string indicating why.
766 * Allow a kernel service to abort a call, if it's still in an abortable state
767 * and return true if the call was aborted, false if it was already complete.
769 bool rxrpc_kernel_abort_call(struct socket
*sock
, struct rxrpc_call
*call
,
770 u32 abort_code
, int error
, const char *why
)
774 _enter("{%d},%d,%d,%s", call
->debug_id
, abort_code
, error
, why
);
776 mutex_lock(&call
->user_mutex
);
778 aborted
= rxrpc_abort_call(why
, call
, 0, abort_code
, error
);
780 rxrpc_send_abort_packet(call
);
782 mutex_unlock(&call
->user_mutex
);
785 EXPORT_SYMBOL(rxrpc_kernel_abort_call
);
788 * rxrpc_kernel_set_tx_length - Set the total Tx length on a call
789 * @sock: The socket the call is on
790 * @call: The call to be informed
791 * @tx_total_len: The amount of data to be transmitted for this call
793 * Allow a kernel service to set the total transmit length on a call. This
794 * allows buffer-to-packet encrypt-and-copy to be performed.
796 * This function is primarily for use for setting the reply length since the
797 * request length can be set when beginning the call.
799 void rxrpc_kernel_set_tx_length(struct socket
*sock
, struct rxrpc_call
*call
,
802 WARN_ON(call
->tx_total_len
!= -1);
803 call
->tx_total_len
= tx_total_len
;
805 EXPORT_SYMBOL(rxrpc_kernel_set_tx_length
);