1 // SPDX-License-Identifier: GPL-2.0-or-later
2 /* RxRPC packet transmission
4 * Copyright (C) 2007 Red Hat, Inc. All Rights Reserved.
5 * Written by David Howells (dhowells@redhat.com)
8 #define pr_fmt(fmt) KBUILD_MODNAME ": " fmt
10 #include <linux/net.h>
11 #include <linux/gfp.h>
12 #include <linux/skbuff.h>
13 #include <linux/export.h>
15 #include <net/af_rxrpc.h>
17 #include "ar-internal.h"
19 struct rxrpc_ack_buffer {
20 struct rxrpc_wire_header whdr;
21 struct rxrpc_ackpacket ack;
24 struct rxrpc_ackinfo ackinfo;
27 extern int udpv6_sendmsg(struct sock *sk, struct msghdr *msg, size_t len);
29 static ssize_t do_udp_sendmsg(struct socket *sk, struct msghdr *msg, size_t len)
31 #if IS_ENABLED(CONFIG_AF_RXRPC_IPV6)
32 struct sockaddr *sa = msg->msg_name;
34 if (sa->sa_family == AF_INET6)
35 return udpv6_sendmsg(sk->sk, msg, len);
37 return udp_sendmsg(sk->sk, msg, len);
40 struct rxrpc_abort_buffer {
41 struct rxrpc_wire_header whdr;
45 static const char rxrpc_keepalive_string[] = "";
48 * Increase Tx backoff on transmission failure and clear it on success.
50 static void rxrpc_tx_backoff(struct rxrpc_call *call, int ret)
53 u16 tx_backoff = READ_ONCE(call->tx_backoff);
56 WRITE_ONCE(call->tx_backoff, tx_backoff + 1);
58 WRITE_ONCE(call->tx_backoff, 0);
63 * Arrange for a keepalive ping a certain time after we last transmitted. This
64 * lets the far side know we're still interested in this call and helps keep
65 * the route through any intervening firewall open.
67 * Receiving a response to the ping will prevent the ->expect_rx_by timer from
70 static void rxrpc_set_keepalive(struct rxrpc_call *call)
72 unsigned long now = jiffies, keepalive_at = call->next_rx_timo / 6;
75 WRITE_ONCE(call->keepalive_at, keepalive_at);
76 rxrpc_reduce_call_timer(call, keepalive_at, now,
77 rxrpc_timer_set_for_keepalive);
81 * Fill out an ACK packet.
83 static size_t rxrpc_fill_out_ack(struct rxrpc_connection *conn,
84 struct rxrpc_call *call,
85 struct rxrpc_ack_buffer *pkt,
86 rxrpc_seq_t *_hard_ack,
90 rxrpc_serial_t serial;
92 rxrpc_seq_t hard_ack, top, seq;
97 tmp = atomic_xchg(&call->ackr_nr_unacked, 0);
98 tmp |= atomic_xchg(&call->ackr_nr_consumed, 0);
99 if (!tmp && (reason == RXRPC_ACK_DELAY ||
100 reason == RXRPC_ACK_IDLE)) {
101 rxrpc_inc_stat(call->rxnet, stat_tx_ack_skip);
105 rxrpc_inc_stat(call->rxnet, stat_tx_ack_fill);
107 /* Barrier against rxrpc_input_data(). */
108 serial = call->ackr_serial;
109 hard_ack = READ_ONCE(call->rx_hard_ack);
110 top = smp_load_acquire(&call->rx_top);
111 *_hard_ack = hard_ack;
114 pkt->ack.bufferSpace = htons(0);
115 pkt->ack.maxSkew = htons(0);
116 pkt->ack.firstPacket = htonl(hard_ack + 1);
117 pkt->ack.previousPacket = htonl(call->ackr_highest_seq);
118 pkt->ack.serial = htonl(serial);
119 pkt->ack.reason = reason;
120 pkt->ack.nAcks = top - hard_ack;
122 if (reason == RXRPC_ACK_PING)
123 pkt->whdr.flags |= RXRPC_REQUEST_ACK;
125 if (after(top, hard_ack)) {
128 ix = seq & RXRPC_RXTX_BUFF_MASK;
129 if (call->rxtx_buffer[ix])
130 *ackp++ = RXRPC_ACK_TYPE_ACK;
132 *ackp++ = RXRPC_ACK_TYPE_NACK;
134 } while (before_eq(seq, top));
137 mtu = conn->params.peer->if_mtu;
138 mtu -= conn->params.peer->hdrsize;
139 jmax = (call->nr_jumbo_bad > 3) ? 1 : rxrpc_rx_jumbo_max;
140 pkt->ackinfo.rxMTU = htonl(rxrpc_rx_mtu);
141 pkt->ackinfo.maxMTU = htonl(mtu);
142 pkt->ackinfo.rwind = htonl(call->rx_winsize);
143 pkt->ackinfo.jumbo_max = htonl(jmax);
148 return top - hard_ack + 3;
152 * Record the beginning of an RTT probe.
154 static int rxrpc_begin_rtt_probe(struct rxrpc_call *call, rxrpc_serial_t serial,
155 enum rxrpc_rtt_tx_trace why)
157 unsigned long avail = call->rtt_avail;
160 if (!(avail & RXRPC_CALL_RTT_AVAIL_MASK))
163 rtt_slot = __ffs(avail & RXRPC_CALL_RTT_AVAIL_MASK);
164 if (!test_and_clear_bit(rtt_slot, &call->rtt_avail))
167 call->rtt_serial[rtt_slot] = serial;
168 call->rtt_sent_at[rtt_slot] = ktime_get_real();
169 smp_wmb(); /* Write data before avail bit */
170 set_bit(rtt_slot + RXRPC_CALL_RTT_PEND_SHIFT, &call->rtt_avail);
172 trace_rxrpc_rtt_tx(call, why, rtt_slot, serial);
176 trace_rxrpc_rtt_tx(call, rxrpc_rtt_tx_no_slot, rtt_slot, serial);
181 * Cancel an RTT probe.
183 static void rxrpc_cancel_rtt_probe(struct rxrpc_call *call,
184 rxrpc_serial_t serial, int rtt_slot)
186 if (rtt_slot != -1) {
187 clear_bit(rtt_slot + RXRPC_CALL_RTT_PEND_SHIFT, &call->rtt_avail);
188 smp_wmb(); /* Clear pending bit before setting slot */
189 set_bit(rtt_slot, &call->rtt_avail);
190 trace_rxrpc_rtt_tx(call, rxrpc_rtt_tx_cancel, rtt_slot, serial);
195 * Send an ACK call packet.
197 int rxrpc_send_ack_packet(struct rxrpc_call *call, bool ping,
198 rxrpc_serial_t *_serial)
200 struct rxrpc_connection *conn;
201 struct rxrpc_ack_buffer *pkt;
204 rxrpc_serial_t serial;
205 rxrpc_seq_t hard_ack, top;
207 int ret, rtt_slot = -1;
210 if (test_bit(RXRPC_CALL_DISCONNECTED, &call->flags))
213 pkt = kzalloc(sizeof(*pkt), GFP_KERNEL);
219 msg.msg_name = &call->peer->srx.transport;
220 msg.msg_namelen = call->peer->srx.transport_len;
221 msg.msg_control = NULL;
222 msg.msg_controllen = 0;
225 pkt->whdr.epoch = htonl(conn->proto.epoch);
226 pkt->whdr.cid = htonl(call->cid);
227 pkt->whdr.callNumber = htonl(call->call_id);
229 pkt->whdr.type = RXRPC_PACKET_TYPE_ACK;
230 pkt->whdr.flags = RXRPC_SLOW_START_OK | conn->out_clientflag;
231 pkt->whdr.userStatus = 0;
232 pkt->whdr.securityIndex = call->security_ix;
234 pkt->whdr.serviceId = htons(call->service_id);
236 spin_lock_bh(&call->lock);
238 reason = RXRPC_ACK_PING;
240 reason = call->ackr_reason;
241 if (!call->ackr_reason) {
242 spin_unlock_bh(&call->lock);
246 call->ackr_reason = 0;
248 n = rxrpc_fill_out_ack(conn, call, pkt, &hard_ack, &top, reason);
250 spin_unlock_bh(&call->lock);
256 iov[0].iov_base = pkt;
257 iov[0].iov_len = sizeof(pkt->whdr) + sizeof(pkt->ack) + n;
258 iov[1].iov_base = &pkt->ackinfo;
259 iov[1].iov_len = sizeof(pkt->ackinfo);
260 len = iov[0].iov_len + iov[1].iov_len;
262 serial = atomic_inc_return(&conn->serial);
263 pkt->whdr.serial = htonl(serial);
264 trace_rxrpc_tx_ack(call->debug_id, serial,
265 ntohl(pkt->ack.firstPacket),
266 ntohl(pkt->ack.serial),
267 pkt->ack.reason, pkt->ack.nAcks);
272 rtt_slot = rxrpc_begin_rtt_probe(call, serial, rxrpc_rtt_tx_ping);
274 rxrpc_inc_stat(call->rxnet, stat_tx_ack_send);
276 iov_iter_kvec(&msg.msg_iter, WRITE, iov, 2, len);
277 ret = do_udp_sendmsg(conn->params.local->socket, &msg, len);
278 call->peer->last_tx_at = ktime_get_seconds();
280 trace_rxrpc_tx_fail(call->debug_id, serial, ret,
281 rxrpc_tx_point_call_ack);
283 trace_rxrpc_tx_packet(call->debug_id, &pkt->whdr,
284 rxrpc_tx_point_call_ack);
285 rxrpc_tx_backoff(call, ret);
287 if (call->state < RXRPC_CALL_COMPLETE) {
289 rxrpc_cancel_rtt_probe(call, serial, rtt_slot);
290 rxrpc_propose_ACK(call, pkt->ack.reason,
291 ntohl(pkt->ack.serial),
293 rxrpc_propose_ack_retry_tx);
296 rxrpc_set_keepalive(call);
305 * Send an ABORT call packet.
307 int rxrpc_send_abort_packet(struct rxrpc_call *call)
309 struct rxrpc_connection *conn;
310 struct rxrpc_abort_buffer pkt;
313 rxrpc_serial_t serial;
316 /* Don't bother sending aborts for a client call once the server has
317 * hard-ACK'd all of its request data. After that point, we're not
318 * going to stop the operation proceeding, and whilst we might limit
319 * the reply, it's not worth it if we can send a new call on the same
320 * channel instead, thereby closing off this call.
322 if (rxrpc_is_client_call(call) &&
323 test_bit(RXRPC_CALL_TX_LAST, &call->flags))
326 if (test_bit(RXRPC_CALL_DISCONNECTED, &call->flags))
331 msg.msg_name = &call->peer->srx.transport;
332 msg.msg_namelen = call->peer->srx.transport_len;
333 msg.msg_control = NULL;
334 msg.msg_controllen = 0;
337 pkt.whdr.epoch = htonl(conn->proto.epoch);
338 pkt.whdr.cid = htonl(call->cid);
339 pkt.whdr.callNumber = htonl(call->call_id);
341 pkt.whdr.type = RXRPC_PACKET_TYPE_ABORT;
342 pkt.whdr.flags = conn->out_clientflag;
343 pkt.whdr.userStatus = 0;
344 pkt.whdr.securityIndex = call->security_ix;
346 pkt.whdr.serviceId = htons(call->service_id);
347 pkt.abort_code = htonl(call->abort_code);
349 iov[0].iov_base = &pkt;
350 iov[0].iov_len = sizeof(pkt);
352 serial = atomic_inc_return(&conn->serial);
353 pkt.whdr.serial = htonl(serial);
355 iov_iter_kvec(&msg.msg_iter, WRITE, iov, 1, sizeof(pkt));
356 ret = do_udp_sendmsg(conn->params.local->socket, &msg, sizeof(pkt));
357 conn->params.peer->last_tx_at = ktime_get_seconds();
359 trace_rxrpc_tx_fail(call->debug_id, serial, ret,
360 rxrpc_tx_point_call_abort);
362 trace_rxrpc_tx_packet(call->debug_id, &pkt.whdr,
363 rxrpc_tx_point_call_abort);
364 rxrpc_tx_backoff(call, ret);
369 * send a packet through the transport endpoint
371 int rxrpc_send_data_packet(struct rxrpc_call *call, struct sk_buff *skb,
374 enum rxrpc_req_ack_trace why;
375 struct rxrpc_connection *conn = call->conn;
376 struct rxrpc_wire_header whdr;
377 struct rxrpc_skb_priv *sp = rxrpc_skb(skb);
380 rxrpc_serial_t serial;
382 int ret, rtt_slot = -1;
384 _enter(",{%d}", skb->len);
386 if (hlist_unhashed(&call->error_link)) {
387 spin_lock_bh(&call->peer->lock);
388 hlist_add_head_rcu(&call->error_link, &call->peer->error_targets);
389 spin_unlock_bh(&call->peer->lock);
392 /* Each transmission of a Tx packet needs a new serial number */
393 serial = atomic_inc_return(&conn->serial);
395 whdr.epoch = htonl(conn->proto.epoch);
396 whdr.cid = htonl(call->cid);
397 whdr.callNumber = htonl(call->call_id);
398 whdr.seq = htonl(sp->hdr.seq);
399 whdr.serial = htonl(serial);
400 whdr.type = RXRPC_PACKET_TYPE_DATA;
401 whdr.flags = sp->hdr.flags;
403 whdr.securityIndex = call->security_ix;
404 whdr._rsvd = htons(sp->hdr._rsvd);
405 whdr.serviceId = htons(call->service_id);
407 if (test_bit(RXRPC_CONN_PROBING_FOR_UPGRADE, &conn->flags) &&
409 whdr.userStatus = RXRPC_USERSTATUS_SERVICE_UPGRADE;
411 iov[0].iov_base = &whdr;
412 iov[0].iov_len = sizeof(whdr);
413 iov[1].iov_base = skb->head;
414 iov[1].iov_len = skb->len;
415 len = iov[0].iov_len + iov[1].iov_len;
416 iov_iter_kvec(&msg.msg_iter, WRITE, iov, 2, len);
418 msg.msg_name = &call->peer->srx.transport;
419 msg.msg_namelen = call->peer->srx.transport_len;
420 msg.msg_control = NULL;
421 msg.msg_controllen = 0;
424 /* If our RTT cache needs working on, request an ACK. Also request
425 * ACKs if a DATA packet appears to have been lost.
427 * However, we mustn't request an ACK on the last reply packet of a
428 * service call, lest OpenAFS incorrectly send us an ACK with some
429 * soft-ACKs in it and then never follow up with a proper hard ACK.
431 if (whdr.flags & RXRPC_REQUEST_ACK)
432 why = rxrpc_reqack_already_on;
433 else if ((whdr.flags & RXRPC_LAST_PACKET) && rxrpc_to_client(sp))
434 why = rxrpc_reqack_no_srv_last;
435 else if (test_and_clear_bit(RXRPC_CALL_EV_ACK_LOST, &call->events))
436 why = rxrpc_reqack_ack_lost;
438 why = rxrpc_reqack_retrans;
439 else if (call->cong_mode == RXRPC_CALL_SLOW_START && call->cong_cwnd <= 2)
440 why = rxrpc_reqack_slow_start;
441 else if (call->tx_winsize <= 2)
442 why = rxrpc_reqack_small_txwin;
443 else if (call->peer->rtt_count < 3 && sp->hdr.seq & 1)
444 why = rxrpc_reqack_more_rtt;
445 else if (ktime_before(ktime_add_ms(call->peer->rtt_last_req, 1000), ktime_get_real()))
446 why = rxrpc_reqack_old_rtt;
448 goto dont_set_request_ack;
450 rxrpc_inc_stat(call->rxnet, stat_why_req_ack[why]);
451 trace_rxrpc_req_ack(call->debug_id, sp->hdr.seq, why);
452 if (why != rxrpc_reqack_no_srv_last)
453 whdr.flags |= RXRPC_REQUEST_ACK;
454 dont_set_request_ack:
456 if (IS_ENABLED(CONFIG_AF_RXRPC_INJECT_LOSS)) {
458 if ((lose++ & 7) == 7) {
460 trace_rxrpc_tx_data(call, sp->hdr.seq, serial,
461 whdr.flags, retrans, true);
466 trace_rxrpc_tx_data(call, sp->hdr.seq, serial, whdr.flags, retrans,
469 /* send the packet with the don't fragment bit set if we currently
470 * think it's small enough */
471 if (iov[1].iov_len >= call->peer->maxdata)
472 goto send_fragmentable;
474 down_read(&conn->params.local->defrag_sem);
476 sp->hdr.serial = serial;
477 smp_wmb(); /* Set serial before timestamp */
478 skb->tstamp = ktime_get_real();
479 if (whdr.flags & RXRPC_REQUEST_ACK)
480 rtt_slot = rxrpc_begin_rtt_probe(call, serial, rxrpc_rtt_tx_data);
482 /* send the packet by UDP
483 * - returns -EMSGSIZE if UDP would have to fragment the packet
484 * to go out of the interface
485 * - in which case, we'll have processed the ICMP error
486 * message and update the peer record
488 rxrpc_inc_stat(call->rxnet, stat_tx_data_send);
489 ret = do_udp_sendmsg(conn->params.local->socket, &msg, len);
490 conn->params.peer->last_tx_at = ktime_get_seconds();
492 up_read(&conn->params.local->defrag_sem);
494 rxrpc_cancel_rtt_probe(call, serial, rtt_slot);
495 trace_rxrpc_tx_fail(call->debug_id, serial, ret,
496 rxrpc_tx_point_call_data_nofrag);
498 trace_rxrpc_tx_packet(call->debug_id, &whdr,
499 rxrpc_tx_point_call_data_nofrag);
502 rxrpc_tx_backoff(call, ret);
503 if (ret == -EMSGSIZE)
504 goto send_fragmentable;
508 if (whdr.flags & RXRPC_REQUEST_ACK) {
509 call->peer->rtt_last_req = skb->tstamp;
510 if (call->peer->rtt_count > 1) {
511 unsigned long nowj = jiffies, ack_lost_at;
513 ack_lost_at = rxrpc_get_rto_backoff(call->peer, false);
515 WRITE_ONCE(call->ack_lost_at, ack_lost_at);
516 rxrpc_reduce_call_timer(call, ack_lost_at, nowj,
517 rxrpc_timer_set_for_lost_ack);
521 if (sp->hdr.seq == 1 &&
522 !test_and_set_bit(RXRPC_CALL_BEGAN_RX_TIMER,
524 unsigned long nowj = jiffies, expect_rx_by;
526 expect_rx_by = nowj + call->next_rx_timo;
527 WRITE_ONCE(call->expect_rx_by, expect_rx_by);
528 rxrpc_reduce_call_timer(call, expect_rx_by, nowj,
529 rxrpc_timer_set_for_normal);
532 rxrpc_set_keepalive(call);
534 /* Cancel the call if the initial transmission fails,
535 * particularly if that's due to network routing issues that
536 * aren't going away anytime soon. The layer above can arrange
537 * the retransmission.
539 if (!test_and_set_bit(RXRPC_CALL_BEGAN_RX_TIMER, &call->flags))
540 rxrpc_set_call_completion(call, RXRPC_CALL_LOCAL_ERROR,
544 _leave(" = %d [%u]", ret, call->peer->maxdata);
548 /* attempt to send this message with fragmentation enabled */
549 _debug("send fragment");
551 down_write(&conn->params.local->defrag_sem);
553 sp->hdr.serial = serial;
554 smp_wmb(); /* Set serial before timestamp */
555 skb->tstamp = ktime_get_real();
556 if (whdr.flags & RXRPC_REQUEST_ACK)
557 rtt_slot = rxrpc_begin_rtt_probe(call, serial, rxrpc_rtt_tx_data);
559 switch (conn->params.local->srx.transport.family) {
562 ip_sock_set_mtu_discover(conn->params.local->socket->sk,
564 rxrpc_inc_stat(call->rxnet, stat_tx_data_send_frag);
565 ret = do_udp_sendmsg(conn->params.local->socket, &msg, len);
566 conn->params.peer->last_tx_at = ktime_get_seconds();
568 ip_sock_set_mtu_discover(conn->params.local->socket->sk,
577 rxrpc_cancel_rtt_probe(call, serial, rtt_slot);
578 trace_rxrpc_tx_fail(call->debug_id, serial, ret,
579 rxrpc_tx_point_call_data_frag);
581 trace_rxrpc_tx_packet(call->debug_id, &whdr,
582 rxrpc_tx_point_call_data_frag);
584 rxrpc_tx_backoff(call, ret);
586 up_write(&conn->params.local->defrag_sem);
591 * reject packets through the local endpoint
593 void rxrpc_reject_packets(struct rxrpc_local *local)
595 struct sockaddr_rxrpc srx;
596 struct rxrpc_skb_priv *sp;
597 struct rxrpc_wire_header whdr;
605 _enter("%d", local->debug_id);
607 iov[0].iov_base = &whdr;
608 iov[0].iov_len = sizeof(whdr);
609 iov[1].iov_base = &code;
610 iov[1].iov_len = sizeof(code);
612 msg.msg_name = &srx.transport;
613 msg.msg_control = NULL;
614 msg.msg_controllen = 0;
617 memset(&whdr, 0, sizeof(whdr));
619 while ((skb = skb_dequeue(&local->reject_queue))) {
620 rxrpc_see_skb(skb, rxrpc_skb_seen);
624 case RXRPC_SKB_MARK_REJECT_BUSY:
625 whdr.type = RXRPC_PACKET_TYPE_BUSY;
629 case RXRPC_SKB_MARK_REJECT_ABORT:
630 whdr.type = RXRPC_PACKET_TYPE_ABORT;
631 code = htonl(skb->priority);
632 size = sizeof(whdr) + sizeof(code);
636 rxrpc_free_skb(skb, rxrpc_skb_freed);
640 if (rxrpc_extract_addr_from_skb(&srx, skb) == 0) {
641 msg.msg_namelen = srx.transport_len;
643 whdr.epoch = htonl(sp->hdr.epoch);
644 whdr.cid = htonl(sp->hdr.cid);
645 whdr.callNumber = htonl(sp->hdr.callNumber);
646 whdr.serviceId = htons(sp->hdr.serviceId);
647 whdr.flags = sp->hdr.flags;
648 whdr.flags ^= RXRPC_CLIENT_INITIATED;
649 whdr.flags &= RXRPC_CLIENT_INITIATED;
651 iov_iter_kvec(&msg.msg_iter, WRITE, iov, ioc, size);
652 ret = do_udp_sendmsg(local->socket, &msg, size);
654 trace_rxrpc_tx_fail(local->debug_id, 0, ret,
655 rxrpc_tx_point_reject);
657 trace_rxrpc_tx_packet(local->debug_id, &whdr,
658 rxrpc_tx_point_reject);
661 rxrpc_free_skb(skb, rxrpc_skb_freed);
668 * Send a VERSION reply to a peer as a keepalive.
670 void rxrpc_send_keepalive(struct rxrpc_peer *peer)
672 struct rxrpc_wire_header whdr;
680 msg.msg_name = &peer->srx.transport;
681 msg.msg_namelen = peer->srx.transport_len;
682 msg.msg_control = NULL;
683 msg.msg_controllen = 0;
686 whdr.epoch = htonl(peer->local->rxnet->epoch);
691 whdr.type = RXRPC_PACKET_TYPE_VERSION; /* Not client-initiated */
692 whdr.flags = RXRPC_LAST_PACKET;
694 whdr.securityIndex = 0;
698 iov[0].iov_base = &whdr;
699 iov[0].iov_len = sizeof(whdr);
700 iov[1].iov_base = (char *)rxrpc_keepalive_string;
701 iov[1].iov_len = sizeof(rxrpc_keepalive_string);
703 len = iov[0].iov_len + iov[1].iov_len;
705 _proto("Tx VERSION (keepalive)");
707 iov_iter_kvec(&msg.msg_iter, WRITE, iov, 2, len);
708 ret = do_udp_sendmsg(peer->local->socket, &msg, len);
710 trace_rxrpc_tx_fail(peer->debug_id, 0, ret,
711 rxrpc_tx_point_version_keepalive);
713 trace_rxrpc_tx_packet(peer->debug_id, &whdr,
714 rxrpc_tx_point_version_keepalive);
716 peer->last_tx_at = ktime_get_seconds();