net/rxrpc/call_event.c

   1 // SPDX-License-Identifier: GPL-2.0-or-later
   2 /* Management of Tx window, Tx resend, ACKs and out-of-sequence reception
   3  *
   4  * Copyright (C) 2007 Red Hat, Inc. All Rights Reserved.
   5  * Written by David Howells (dhowells@redhat.com)
   6  */
   7
   8 #define pr_fmt(fmt) KBUILD_MODNAME ": " fmt
   9
  10 #include <linux/module.h>
  11 #include <linux/circ_buf.h>
  12 #include <linux/net.h>
  13 #include <linux/skbuff.h>
  14 #include <linux/slab.h>
  15 #include <linux/udp.h>
  16 #include <net/sock.h>
  17 #include <net/af_rxrpc.h>
  18 #include "ar-internal.h"
  19
  20 /*
  21  * Propose a PING ACK be sent.
  22  */
  23 void rxrpc_propose_ping(struct rxrpc_call *call, u32 serial,
  24                         enum rxrpc_propose_ack_trace why)
  25 {
  26         unsigned long now = jiffies;
  27         unsigned long ping_at = now + rxrpc_idle_ack_delay;
  28
  29         if (time_before(ping_at, call->ping_at)) {
  30                 WRITE_ONCE(call->ping_at, ping_at);
  31                 rxrpc_reduce_call_timer(call, ping_at, now,
  32                                         rxrpc_timer_set_for_ping);
  33                 trace_rxrpc_propose_ack(call, why, RXRPC_ACK_PING, serial);
  34         }
  35 }
  36
  37 /*
  38  * Propose a DELAY ACK be sent in the future.
  39  */
  40 void rxrpc_propose_delay_ACK(struct rxrpc_call *call, rxrpc_serial_t serial,
  41                              enum rxrpc_propose_ack_trace why)
  42 {
  43         unsigned long expiry = rxrpc_soft_ack_delay;
  44         unsigned long now = jiffies, ack_at;
  45
  46         call->ackr_serial = serial;
  47
  48         if (rxrpc_soft_ack_delay < expiry)
  49                 expiry = rxrpc_soft_ack_delay;
  50         if (call->peer->srtt_us != 0)
  51                 ack_at = usecs_to_jiffies(call->peer->srtt_us >> 3);
  52         else
  53                 ack_at = expiry;
  54
  55         ack_at += READ_ONCE(call->tx_backoff);
  56         ack_at += now;
  57         if (time_before(ack_at, call->delay_ack_at)) {
  58                 WRITE_ONCE(call->delay_ack_at, ack_at);
  59                 rxrpc_reduce_call_timer(call, ack_at, now,
  60                                         rxrpc_timer_set_for_ack);
  61         }
  62
  63         trace_rxrpc_propose_ack(call, why, RXRPC_ACK_DELAY, serial);
  64 }
  65
  66 /*
  67  * Queue an ACK for immediate transmission.
  68  */
  69 void rxrpc_send_ACK(struct rxrpc_call *call, u8 ack_reason,
  70                     rxrpc_serial_t serial, enum rxrpc_propose_ack_trace why)
  71 {
  72         struct rxrpc_local *local = call->conn->local;
  73         struct rxrpc_txbuf *txb;
  74
  75         if (test_bit(RXRPC_CALL_DISCONNECTED, &call->flags))
  76                 return;
  77
  78         rxrpc_inc_stat(call->rxnet, stat_tx_acks[ack_reason]);
  79
  80         txb = rxrpc_alloc_txbuf(call, RXRPC_PACKET_TYPE_ACK,
  81                                 rcu_read_lock_held() ? GFP_ATOMIC | __GFP_NOWARN : GFP_NOFS);
  82         if (!txb) {
  83                 kleave(" = -ENOMEM");
  84                 return;
  85         }
  86
  87         txb->ack_why            = why;
  88         txb->wire.seq           = 0;
  89         txb->wire.type          = RXRPC_PACKET_TYPE_ACK;
  90         txb->wire.flags         |= RXRPC_SLOW_START_OK;
  91         txb->ack.bufferSpace    = 0;
  92         txb->ack.maxSkew        = 0;
  93         txb->ack.firstPacket    = 0;
  94         txb->ack.previousPacket = 0;
  95         txb->ack.serial         = htonl(serial);
  96         txb->ack.reason         = ack_reason;
  97         txb->ack.nAcks          = 0;
  98
  99         if (!rxrpc_try_get_call(call, rxrpc_call_get_send_ack)) {
 100                 rxrpc_put_txbuf(txb, rxrpc_txbuf_put_nomem);
 101                 return;
 102         }
 103
 104         spin_lock(&local->ack_tx_lock);
 105         list_add_tail(&txb->tx_link, &local->ack_tx_queue);
 106         spin_unlock(&local->ack_tx_lock);
 107         trace_rxrpc_send_ack(call, why, ack_reason, serial);
 108
 109         rxrpc_wake_up_io_thread(local);
 110 }
 111
 112 /*
 113  * Handle congestion being detected by the retransmit timeout.
 114  */
 115 static void rxrpc_congestion_timeout(struct rxrpc_call *call)
 116 {
 117         set_bit(RXRPC_CALL_RETRANS_TIMEOUT, &call->flags);
 118 }
 119
 120 /*
 121  * Perform retransmission of NAK'd and unack'd packets.
 122  */
 123 void rxrpc_resend(struct rxrpc_call *call, struct sk_buff *ack_skb)
 124 {
 125         struct rxrpc_ackpacket *ack = NULL;
 126         struct rxrpc_txbuf *txb;
 127         unsigned long resend_at;
 128         rxrpc_seq_t transmitted = READ_ONCE(call->tx_transmitted);
 129         ktime_t now, max_age, oldest, ack_ts;
 130         bool unacked = false;
 131         unsigned int i;
 132         LIST_HEAD(retrans_queue);
 133
 134         _enter("{%d,%d}", call->acks_hard_ack, call->tx_top);
 135
 136         now = ktime_get_real();
 137         max_age = ktime_sub_us(now, jiffies_to_usecs(call->peer->rto_j));
 138         oldest = now;
 139
 140         if (list_empty(&call->tx_buffer))
 141                 goto no_resend;
 142
 143         if (list_empty(&call->tx_buffer))
 144                 goto no_further_resend;
 145
 146         trace_rxrpc_resend(call, ack_skb);
 147         txb = list_first_entry(&call->tx_buffer, struct rxrpc_txbuf, call_link);
 148
 149         /* Scan the soft ACK table without dropping the lock and resend any
 150          * explicitly NAK'd packets.
 151          */
 152         if (ack_skb) {
 153                 ack = (void *)ack_skb->data + sizeof(struct rxrpc_wire_header);
 154
 155                 for (i = 0; i < ack->nAcks; i++) {
 156                         rxrpc_seq_t seq;
 157
 158                         if (ack->acks[i] & 1)
 159                                 continue;
 160                         seq = ntohl(ack->firstPacket) + i;
 161                         if (after(txb->seq, transmitted))
 162                                 break;
 163                         if (after(txb->seq, seq))
 164                                 continue; /* A new hard ACK probably came in */
 165                         list_for_each_entry_from(txb, &call->tx_buffer, call_link) {
 166                                 if (txb->seq == seq)
 167                                         goto found_txb;
 168                         }
 169                         goto no_further_resend;
 170
 171                 found_txb:
 172                         if (after(ntohl(txb->wire.serial), call->acks_highest_serial))
 173                                 continue; /* Ack point not yet reached */
 174
 175                         rxrpc_see_txbuf(txb, rxrpc_txbuf_see_unacked);
 176
 177                         if (list_empty(&txb->tx_link)) {
 178                                 list_add_tail(&txb->tx_link, &retrans_queue);
 179                                 set_bit(RXRPC_TXBUF_RESENT, &txb->flags);
 180                         }
 181
 182                         trace_rxrpc_retransmit(call, txb->seq,
 183                                                ktime_to_ns(ktime_sub(txb->last_sent,
 184                                                                      max_age)));
 185
 186                         if (list_is_last(&txb->call_link, &call->tx_buffer))
 187                                 goto no_further_resend;
 188                         txb = list_next_entry(txb, call_link);
 189                 }
 190         }
 191
 192         /* Fast-forward through the Tx queue to the point the peer says it has
 193          * seen.  Anything between the soft-ACK table and that point will get
 194          * ACK'd or NACK'd in due course, so don't worry about it here; here we
 195          * need to consider retransmitting anything beyond that point.
 196          *
 197          * Note that ACK for a packet can beat the update of tx_transmitted.
 198          */
 199         if (after_eq(READ_ONCE(call->acks_prev_seq), READ_ONCE(call->tx_transmitted)))
 200                 goto no_further_resend;
 201
 202         list_for_each_entry_from(txb, &call->tx_buffer, call_link) {
 203                 if (before_eq(txb->seq, READ_ONCE(call->acks_prev_seq)))
 204                         continue;
 205                 if (after(txb->seq, READ_ONCE(call->tx_transmitted)))
 206                         break; /* Not transmitted yet */
 207
 208                 if (ack && ack->reason == RXRPC_ACK_PING_RESPONSE &&
 209                     before(ntohl(txb->wire.serial), ntohl(ack->serial)))
 210                         goto do_resend; /* Wasn't accounted for by a more recent ping. */
 211
 212                 if (ktime_after(txb->last_sent, max_age)) {
 213                         if (ktime_before(txb->last_sent, oldest))
 214                                 oldest = txb->last_sent;
 215                         continue;
 216                 }
 217
 218         do_resend:
 219                 unacked = true;
 220                 if (list_empty(&txb->tx_link)) {
 221                         list_add_tail(&txb->tx_link, &retrans_queue);
 222                         set_bit(RXRPC_TXBUF_RESENT, &txb->flags);
 223                         rxrpc_inc_stat(call->rxnet, stat_tx_data_retrans);
 224                 }
 225         }
 226
 227 no_further_resend:
 228 no_resend:
 229         resend_at = nsecs_to_jiffies(ktime_to_ns(ktime_sub(now, oldest)));
 230         resend_at += jiffies + rxrpc_get_rto_backoff(call->peer,
 231                                                      !list_empty(&retrans_queue));
 232         WRITE_ONCE(call->resend_at, resend_at);
 233
 234         if (unacked)
 235                 rxrpc_congestion_timeout(call);
 236
 237         /* If there was nothing that needed retransmission then it's likely
 238          * that an ACK got lost somewhere.  Send a ping to find out instead of
 239          * retransmitting data.
 240          */
 241         if (list_empty(&retrans_queue)) {
 242                 rxrpc_reduce_call_timer(call, resend_at, jiffies,
 243                                         rxrpc_timer_set_for_resend);
 244                 ack_ts = ktime_sub(now, call->acks_latest_ts);
 245                 if (ktime_to_us(ack_ts) < (call->peer->srtt_us >> 3))
 246                         goto out;
 247                 rxrpc_send_ACK(call, RXRPC_ACK_PING, 0,
 248                                rxrpc_propose_ack_ping_for_lost_ack);
 249                 goto out;
 250         }
 251
 252         /* Retransmit the queue */
 253         while ((txb = list_first_entry_or_null(&retrans_queue,
 254                                                struct rxrpc_txbuf, tx_link))) {
 255                 list_del_init(&txb->tx_link);
 256                 rxrpc_transmit_one(call, txb);
 257         }
 258
 259 out:
 260         _leave("");
 261 }
 262
 263 static bool rxrpc_tx_window_has_space(struct rxrpc_call *call)
 264 {
 265         unsigned int winsize = min_t(unsigned int, call->tx_winsize,
 266                                      call->cong_cwnd + call->cong_extra);
 267         rxrpc_seq_t window = call->acks_hard_ack, wtop = window + winsize;
 268         rxrpc_seq_t tx_top = call->tx_top;
 269         int space;
 270
 271         space = wtop - tx_top;
 272         return space > 0;
 273 }
 274
 275 /*
 276  * Decant some if the sendmsg prepared queue into the transmission buffer.
 277  */
 278 static void rxrpc_decant_prepared_tx(struct rxrpc_call *call)
 279 {
 280         struct rxrpc_txbuf *txb;
 281
 282         if (rxrpc_is_client_call(call) &&
 283             !test_bit(RXRPC_CALL_EXPOSED, &call->flags))
 284                 rxrpc_expose_client_call(call);
 285
 286         while ((txb = list_first_entry_or_null(&call->tx_sendmsg,
 287                                                struct rxrpc_txbuf, call_link))) {
 288                 spin_lock(&call->tx_lock);
 289                 list_del(&txb->call_link);
 290                 spin_unlock(&call->tx_lock);
 291
 292                 call->tx_top = txb->seq;
 293                 list_add_tail(&txb->call_link, &call->tx_buffer);
 294
 295                 rxrpc_transmit_one(call, txb);
 296
 297                 // TODO: Drain the transmission buffers.  Do this somewhere better
 298                 if (after(call->acks_hard_ack, call->tx_bottom + 16))
 299                         rxrpc_shrink_call_tx_buffer(call);
 300
 301                 if (!rxrpc_tx_window_has_space(call))
 302                         break;
 303         }
 304 }
 305
 306 static void rxrpc_transmit_some_data(struct rxrpc_call *call)
 307 {
 308         switch (call->state) {
 309         case RXRPC_CALL_SERVER_ACK_REQUEST:
 310                 if (list_empty(&call->tx_sendmsg))
 311                         return;
 312                 fallthrough;
 313
 314         case RXRPC_CALL_SERVER_SEND_REPLY:
 315         case RXRPC_CALL_SERVER_AWAIT_ACK:
 316         case RXRPC_CALL_CLIENT_SEND_REQUEST:
 317         case RXRPC_CALL_CLIENT_AWAIT_REPLY:
 318                 if (!rxrpc_tx_window_has_space(call))
 319                         return;
 320                 if (list_empty(&call->tx_sendmsg))
 321                         return;
 322                 rxrpc_decant_prepared_tx(call);
 323                 break;
 324         default:
 325                 return;
 326         }
 327 }
 328
 329 /*
 330  * Ping the other end to fill our RTT cache and to retrieve the rwind
 331  * and MTU parameters.
 332  */
 333 static void rxrpc_send_initial_ping(struct rxrpc_call *call)
 334 {
 335         if (call->peer->rtt_count < 3 ||
 336             ktime_before(ktime_add_ms(call->peer->rtt_last_req, 1000),
 337                          ktime_get_real()))
 338                 rxrpc_send_ACK(call, RXRPC_ACK_PING, 0,
 339                                rxrpc_propose_ack_ping_for_params);
 340 }
 341
 342 /*
 343  * Handle retransmission and deferred ACK/abort generation.
 344  */
 345 void rxrpc_input_call_event(struct rxrpc_call *call, struct sk_buff *skb)
 346 {
 347         unsigned long now, next, t;
 348         rxrpc_serial_t ackr_serial;
 349         bool resend = false, expired = false;
 350
 351         rxrpc_see_call(call, rxrpc_call_see_input);
 352
 353         //printk("\n--------------------\n");
 354         _enter("{%d,%s,%lx}",
 355                call->debug_id, rxrpc_call_states[call->state], call->events);
 356
 357         if (call->state == RXRPC_CALL_COMPLETE)
 358                 goto out;
 359
 360         if (skb && skb->mark == RXRPC_SKB_MARK_ERROR)
 361                 goto out;
 362
 363         /* If we see our async-event poke, check for timeout trippage. */
 364         now = jiffies;
 365         t = READ_ONCE(call->expect_rx_by);
 366         if (time_after_eq(now, t)) {
 367                 trace_rxrpc_timer(call, rxrpc_timer_exp_normal, now);
 368                 expired = true;
 369         }
 370
 371         t = READ_ONCE(call->expect_req_by);
 372         if (call->state == RXRPC_CALL_SERVER_RECV_REQUEST &&
 373             time_after_eq(now, t)) {
 374                 trace_rxrpc_timer(call, rxrpc_timer_exp_idle, now);
 375                 expired = true;
 376         }
 377
 378         t = READ_ONCE(call->expect_term_by);
 379         if (time_after_eq(now, t)) {
 380                 trace_rxrpc_timer(call, rxrpc_timer_exp_hard, now);
 381                 expired = true;
 382         }
 383
 384         t = READ_ONCE(call->delay_ack_at);
 385         if (time_after_eq(now, t)) {
 386                 trace_rxrpc_timer(call, rxrpc_timer_exp_ack, now);
 387                 cmpxchg(&call->delay_ack_at, t, now + MAX_JIFFY_OFFSET);
 388                 ackr_serial = xchg(&call->ackr_serial, 0);
 389                 rxrpc_send_ACK(call, RXRPC_ACK_DELAY, ackr_serial,
 390                                rxrpc_propose_ack_ping_for_lost_ack);
 391         }
 392
 393         t = READ_ONCE(call->ack_lost_at);
 394         if (time_after_eq(now, t)) {
 395                 trace_rxrpc_timer(call, rxrpc_timer_exp_lost_ack, now);
 396                 cmpxchg(&call->ack_lost_at, t, now + MAX_JIFFY_OFFSET);
 397                 set_bit(RXRPC_CALL_EV_ACK_LOST, &call->events);
 398         }
 399
 400         t = READ_ONCE(call->keepalive_at);
 401         if (time_after_eq(now, t)) {
 402                 trace_rxrpc_timer(call, rxrpc_timer_exp_keepalive, now);
 403                 cmpxchg(&call->keepalive_at, t, now + MAX_JIFFY_OFFSET);
 404                 rxrpc_send_ACK(call, RXRPC_ACK_PING, 0,
 405                                rxrpc_propose_ack_ping_for_keepalive);
 406         }
 407
 408         t = READ_ONCE(call->ping_at);
 409         if (time_after_eq(now, t)) {
 410                 trace_rxrpc_timer(call, rxrpc_timer_exp_ping, now);
 411                 cmpxchg(&call->ping_at, t, now + MAX_JIFFY_OFFSET);
 412                 rxrpc_send_ACK(call, RXRPC_ACK_PING, 0,
 413                                rxrpc_propose_ack_ping_for_keepalive);
 414         }
 415
 416         t = READ_ONCE(call->resend_at);
 417         if (time_after_eq(now, t)) {
 418                 trace_rxrpc_timer(call, rxrpc_timer_exp_resend, now);
 419                 cmpxchg(&call->resend_at, t, now + MAX_JIFFY_OFFSET);
 420                 resend = true;
 421         }
 422
 423         if (skb)
 424                 rxrpc_input_call_packet(call, skb);
 425
 426         rxrpc_transmit_some_data(call);
 427
 428         if (test_and_clear_bit(RXRPC_CALL_EV_INITIAL_PING, &call->events))
 429                 rxrpc_send_initial_ping(call);
 430
 431         /* Process events */
 432         if (expired) {
 433                 if (test_bit(RXRPC_CALL_RX_HEARD, &call->flags) &&
 434                     (int)call->conn->hi_serial - (int)call->rx_serial > 0) {
 435                         trace_rxrpc_call_reset(call);
 436                         rxrpc_abort_call("EXP", call, 0, RX_CALL_DEAD, -ECONNRESET);
 437                 } else {
 438                         rxrpc_abort_call("EXP", call, 0, RX_CALL_TIMEOUT, -ETIME);
 439                 }
 440                 rxrpc_send_abort_packet(call);
 441                 goto out;
 442         }
 443
 444         if (test_and_clear_bit(RXRPC_CALL_EV_ACK_LOST, &call->events))
 445                 rxrpc_send_ACK(call, RXRPC_ACK_PING, 0,
 446                                rxrpc_propose_ack_ping_for_lost_ack);
 447
 448         if (resend && call->state != RXRPC_CALL_CLIENT_RECV_REPLY)
 449                 rxrpc_resend(call, NULL);
 450
 451         if (test_and_clear_bit(RXRPC_CALL_RX_IS_IDLE, &call->flags))
 452                 rxrpc_send_ACK(call, RXRPC_ACK_IDLE, 0,
 453                                rxrpc_propose_ack_rx_idle);
 454
 455         if (atomic_read(&call->ackr_nr_unacked) > 2)
 456                 rxrpc_send_ACK(call, RXRPC_ACK_IDLE, 0,
 457                                rxrpc_propose_ack_input_data);
 458
 459         /* Make sure the timer is restarted */
 460         if (call->state != RXRPC_CALL_COMPLETE) {
 461                 next = call->expect_rx_by;
 462
 463 #define set(T) { t = READ_ONCE(T); if (time_before(t, next)) next = t; }
 464
 465                 set(call->expect_req_by);
 466                 set(call->expect_term_by);
 467                 set(call->delay_ack_at);
 468                 set(call->ack_lost_at);
 469                 set(call->resend_at);
 470                 set(call->keepalive_at);
 471                 set(call->ping_at);
 472
 473                 now = jiffies;
 474                 if (time_after_eq(now, next))
 475                         rxrpc_poke_call(call, rxrpc_call_poke_timer_now);
 476
 477                 rxrpc_reduce_call_timer(call, next, now, rxrpc_timer_restart);
 478         }
 479
 480 out:
 481         if (call->state == RXRPC_CALL_COMPLETE)
 482                 del_timer_sync(&call->timer);
 483         if (call->acks_hard_ack != call->tx_bottom)
 484                 rxrpc_shrink_call_tx_buffer(call);
 485         _leave("");
 486 }