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 */
 23void rxrpc_propose_ping(struct rxrpc_call *call, u32 serial,
 24			enum rxrpc_propose_ack_trace why)
 25{
 26	ktime_t delay = ms_to_ktime(READ_ONCE(rxrpc_idle_ack_delay));
 27	ktime_t now = ktime_get_real();
 28	ktime_t ping_at = ktime_add(now, delay);
 29
 30	trace_rxrpc_propose_ack(call, why, RXRPC_ACK_PING, serial);
 31	if (ktime_before(ping_at, call->ping_at)) {
 32		call->ping_at = ping_at;
 33		trace_rxrpc_timer_set(call, delay, rxrpc_timer_trace_ping);
 34	}
 35}
 36
 37/*
 38 * Propose a DELAY ACK be sent in the future.
 39 */
 40void rxrpc_propose_delay_ACK(struct rxrpc_call *call, rxrpc_serial_t serial,
 41			     enum rxrpc_propose_ack_trace why)
 42{
 43	ktime_t now = ktime_get_real(), delay;
 44
 45	trace_rxrpc_propose_ack(call, why, RXRPC_ACK_DELAY, serial);
 46
 47	if (call->srtt_us)
 48		delay = (call->srtt_us >> 3) * NSEC_PER_USEC;
 49	else
 50		delay = ms_to_ktime(READ_ONCE(rxrpc_soft_ack_delay));
 51	ktime_add_ms(delay, call->tx_backoff);
 52
 53	call->delay_ack_at = ktime_add(now, delay);
 54	trace_rxrpc_timer_set(call, delay, rxrpc_timer_trace_delayed_ack);
 55}
 56
 57/*
 58 * Retransmit one or more packets.
 59 */
 60static bool rxrpc_retransmit_data(struct rxrpc_call *call,
 61				  struct rxrpc_send_data_req *req)
 62{
 63	struct rxrpc_txqueue *tq = req->tq;
 64	unsigned int ix = req->seq & RXRPC_TXQ_MASK;
 65	struct rxrpc_txbuf *txb = tq->bufs[ix];
 66
 67	_enter("%x,%x,%x,%x", tq->qbase, req->seq, ix, txb->debug_id);
 68
 69	req->retrans = true;
 70	trace_rxrpc_retransmit(call, req, txb);
 71
 72	txb->flags |= RXRPC_TXBUF_RESENT;
 73	rxrpc_send_data_packet(call, req);
 74	rxrpc_inc_stat(call->rxnet, stat_tx_data_retrans);
 75
 76	req->tq		= NULL;
 77	req->n		= 0;
 78	req->did_send	= true;
 79	req->now	= ktime_get_real();
 80	return true;
 81}
 82
 83/*
 84 * Perform retransmission of NAK'd and unack'd packets.
 85 */
 86static void rxrpc_resend(struct rxrpc_call *call)
 87{
 88	struct rxrpc_send_data_req req = {
 89		.now	= ktime_get_real(),
 90		.trace	= rxrpc_txdata_retransmit,
 91	};
 92	struct rxrpc_txqueue *tq;
 93
 94	_enter("{%d,%d}", call->tx_bottom, call->tx_top);
 95
 96	trace_rxrpc_resend(call, call->acks_highest_serial);
 97
 98	/* Scan the transmission queue, looking for lost packets. */
 99	for (tq = call->tx_queue; tq; tq = tq->next) {
100		unsigned long lost = tq->segment_lost;
101
102		if (after(tq->qbase, call->tx_transmitted))
103			break;
104
105		_debug("retr %16lx %u c=%08x [%x]",
106		       tq->segment_acked, tq->nr_reported_acks, call->debug_id, tq->qbase);
107		_debug("lost %16lx", lost);
108
109		trace_rxrpc_resend_lost(call, tq, lost);
110		while (lost) {
111			unsigned int ix = __ffs(lost);
112			struct rxrpc_txbuf *txb = tq->bufs[ix];
113
114			__clear_bit(ix, &lost);
115			rxrpc_see_txbuf(txb, rxrpc_txbuf_see_lost);
116
117			req.tq  = tq;
118			req.seq = tq->qbase + ix;
119			req.n   = 1;
120			rxrpc_retransmit_data(call, &req);
121		}
122	}
123
124	rxrpc_get_rto_backoff(call, req.did_send);
125	_leave("");
126}
127
128/*
129 * Resend the highest-seq DATA packet so far transmitted for RACK-TLP [RFC8985 7.3].
130 */
131void rxrpc_resend_tlp(struct rxrpc_call *call)
132{
133	struct rxrpc_send_data_req req = {
134		.now		= ktime_get_real(),
135		.seq		= call->tx_transmitted,
136		.n		= 1,
137		.tlp_probe	= true,
138		.trace		= rxrpc_txdata_tlp_retransmit,
139	};
140
141	/* There's a chance it'll be on the tail segment of the queue. */
142	req.tq = READ_ONCE(call->tx_qtail);
143	if (req.tq &&
144	    before(call->tx_transmitted, req.tq->qbase + RXRPC_NR_TXQUEUE)) {
145		rxrpc_retransmit_data(call, &req);
146		return;
147	}
148
149	for (req.tq = call->tx_queue; req.tq; req.tq = req.tq->next) {
150		if (after_eq(call->tx_transmitted, req.tq->qbase) &&
151		    before(call->tx_transmitted, req.tq->qbase + RXRPC_NR_TXQUEUE)) {
152			rxrpc_retransmit_data(call, &req);
153			return;
154		}
155	}
156}
157
158/*
159 * Start transmitting the reply to a service.  This cancels the need to ACK the
160 * request if we haven't yet done so.
161 */
162static void rxrpc_begin_service_reply(struct rxrpc_call *call)
163{
164	rxrpc_set_call_state(call, RXRPC_CALL_SERVER_SEND_REPLY);
165	if (call->ackr_reason == RXRPC_ACK_DELAY)
166		call->ackr_reason = 0;
167	call->delay_ack_at = KTIME_MAX;
168	trace_rxrpc_timer_can(call, rxrpc_timer_trace_delayed_ack);
169}
170
171/*
172 * Close the transmission phase.  After this point there is no more data to be
173 * transmitted in the call.
174 */
175static void rxrpc_close_tx_phase(struct rxrpc_call *call)
176{
177	_debug("________awaiting reply/ACK__________");
178
179	switch (__rxrpc_call_state(call)) {
180	case RXRPC_CALL_CLIENT_SEND_REQUEST:
181		rxrpc_set_call_state(call, RXRPC_CALL_CLIENT_AWAIT_REPLY);
182		break;
183	case RXRPC_CALL_SERVER_SEND_REPLY:
184		rxrpc_set_call_state(call, RXRPC_CALL_SERVER_AWAIT_ACK);
185		break;
186	default:
187		break;
188	}
189}
190
191/*
192 * Transmit some as-yet untransmitted data, to a maximum of the supplied limit.
193 */
194static void rxrpc_transmit_fresh_data(struct rxrpc_call *call, unsigned int limit,
195				      enum rxrpc_txdata_trace trace)
196{
197	int space = rxrpc_tx_window_space(call);
198
199	if (!test_bit(RXRPC_CALL_EXPOSED, &call->flags)) {
200		if (call->send_top == call->tx_top)
201			return;
202		rxrpc_expose_client_call(call);
203	}
204
205	while (space > 0) {
206		struct rxrpc_send_data_req req = {
207			.now	= ktime_get_real(),
208			.seq	= call->tx_transmitted + 1,
209			.n	= 0,
210			.trace	= trace,
211		};
212		struct rxrpc_txqueue *tq;
213		struct rxrpc_txbuf *txb;
214		rxrpc_seq_t send_top, seq;
215		int limit = min(space, max(call->peer->pmtud_jumbo, 1));
216
217		/* Order send_top before the contents of the new txbufs and
218		 * txqueue pointers
219		 */
220		send_top = smp_load_acquire(&call->send_top);
221		if (call->tx_top == send_top)
222			break;
223
224		trace_rxrpc_transmit(call, send_top, space);
225
226		tq = call->tx_qtail;
227		seq = call->tx_top;
228		trace_rxrpc_tq(call, tq, seq, rxrpc_tq_decant);
229
230		do {
231			int ix;
232
233			seq++;
234			ix = seq & RXRPC_TXQ_MASK;
235			if (!ix) {
236				tq = tq->next;
237				trace_rxrpc_tq(call, tq, seq, rxrpc_tq_decant_advance);
238			}
239			if (!req.tq)
240				req.tq = tq;
241			txb = tq->bufs[ix];
242			req.n++;
243			if (!txb->jumboable)
244				break;
245		} while (req.n < limit && before(seq, send_top));
246
247		if (txb->flags & RXRPC_LAST_PACKET) {
248			rxrpc_close_tx_phase(call);
249			tq = NULL;
250		}
251		call->tx_qtail = tq;
252		call->tx_top = seq;
253
254		space -= req.n;
255		rxrpc_send_data_packet(call, &req);
256	}
257}
258
259void rxrpc_transmit_some_data(struct rxrpc_call *call, unsigned int limit,
260			      enum rxrpc_txdata_trace trace)
261{
262	switch (__rxrpc_call_state(call)) {
263	case RXRPC_CALL_SERVER_ACK_REQUEST:
264		if (call->tx_bottom == READ_ONCE(call->send_top))
265			return;
266		rxrpc_begin_service_reply(call);
267		fallthrough;
268
269	case RXRPC_CALL_SERVER_SEND_REPLY:
270	case RXRPC_CALL_CLIENT_SEND_REQUEST:
271		if (!rxrpc_tx_window_space(call))
272			return;
273		if (call->tx_bottom == READ_ONCE(call->send_top)) {
274			rxrpc_inc_stat(call->rxnet, stat_tx_data_underflow);
275			return;
276		}
277		rxrpc_transmit_fresh_data(call, limit, trace);
278		break;
279	default:
280		return;
281	}
282}
283
284/*
285 * Ping the other end to fill our RTT cache and to retrieve the rwind
286 * and MTU parameters.
287 */
288static void rxrpc_send_initial_ping(struct rxrpc_call *call)
289{
290	if (call->rtt_count < 3 ||
291	    ktime_before(ktime_add_ms(call->rtt_last_req, 1000),
292			 ktime_get_real()))
293		rxrpc_send_ACK(call, RXRPC_ACK_PING, 0,
294			       rxrpc_propose_ack_ping_for_params);
295}
296
297/*
298 * Handle retransmission and deferred ACK/abort generation.
299 */
300bool rxrpc_input_call_event(struct rxrpc_call *call)
301{
302	struct sk_buff *skb;
303	ktime_t now, t;
304	bool did_receive = false, saw_ack = false;
305	s32 abort_code;
306
307	rxrpc_see_call(call, rxrpc_call_see_input);
308
309	//printk("\n--------------------\n");
310	_enter("{%d,%s,%lx}",
311	       call->debug_id, rxrpc_call_states[__rxrpc_call_state(call)],
312	       call->events);
313
314	/* Handle abort request locklessly, vs rxrpc_propose_abort(). */
315	abort_code = smp_load_acquire(&call->send_abort);
316	if (abort_code) {
317		rxrpc_abort_call(call, 0, call->send_abort, call->send_abort_err,
318				 call->send_abort_why);
319		goto out;
320	}
321
322	do {
323		skb = __skb_dequeue(&call->rx_queue);
324		if (skb) {
325			struct rxrpc_skb_priv *sp = rxrpc_skb(skb);
326
327			if (__rxrpc_call_is_complete(call) ||
328			    skb->mark == RXRPC_SKB_MARK_ERROR) {
329				rxrpc_free_skb(skb, rxrpc_skb_put_call_rx);
330				goto out;
331			}
332
333			saw_ack |= sp->hdr.type == RXRPC_PACKET_TYPE_ACK;
334
335			rxrpc_input_call_packet(call, skb);
336			rxrpc_free_skb(skb, rxrpc_skb_put_call_rx);
337			did_receive = true;
338		}
339
340		t = ktime_sub(call->rack_timo_at, ktime_get_real());
341		if (t <= 0) {
342			trace_rxrpc_timer_exp(call, t,
343					      rxrpc_timer_trace_rack_off + call->rack_timer_mode);
344			call->rack_timo_at = KTIME_MAX;
345			rxrpc_rack_timer_expired(call, t);
346		}
347
348	} while (!skb_queue_empty(&call->rx_queue));
349
350	/* If we see our async-event poke, check for timeout trippage. */
351	now = ktime_get_real();
352	t = ktime_sub(call->expect_rx_by, now);
353	if (t <= 0) {
354		trace_rxrpc_timer_exp(call, t, rxrpc_timer_trace_expect_rx);
355		goto expired;
356	}
357
358	t = ktime_sub(call->expect_req_by, now);
359	if (t <= 0) {
360		call->expect_req_by = KTIME_MAX;
361		if (__rxrpc_call_state(call) == RXRPC_CALL_SERVER_RECV_REQUEST) {
362			trace_rxrpc_timer_exp(call, t, rxrpc_timer_trace_idle);
363			goto expired;
364		}
365	}
366
367	t = ktime_sub(READ_ONCE(call->expect_term_by), now);
368	if (t <= 0) {
369		trace_rxrpc_timer_exp(call, t, rxrpc_timer_trace_hard);
370		goto expired;
371	}
372
373	t = ktime_sub(call->delay_ack_at, now);
374	if (t <= 0) {
375		trace_rxrpc_timer_exp(call, t, rxrpc_timer_trace_delayed_ack);
376		call->delay_ack_at = KTIME_MAX;
377		rxrpc_send_ACK(call, RXRPC_ACK_DELAY, 0,
378			       rxrpc_propose_ack_delayed_ack);
379	}
380
381	t = ktime_sub(call->ping_at, now);
382	if (t <= 0) {
383		trace_rxrpc_timer_exp(call, t, rxrpc_timer_trace_ping);
384		call->ping_at = KTIME_MAX;
385		rxrpc_send_ACK(call, RXRPC_ACK_PING, 0,
386			       rxrpc_propose_ack_ping_for_keepalive);
387	}
388
389	now = ktime_get_real();
390	t = ktime_sub(call->keepalive_at, now);
391	if (t <= 0) {
392		trace_rxrpc_timer_exp(call, t, rxrpc_timer_trace_keepalive);
393		call->keepalive_at = KTIME_MAX;
394		rxrpc_send_ACK(call, RXRPC_ACK_PING, 0,
395			       rxrpc_propose_ack_ping_for_keepalive);
396	}
397
398	if (test_and_clear_bit(RXRPC_CALL_EV_INITIAL_PING, &call->events))
399		rxrpc_send_initial_ping(call);
400
401	rxrpc_transmit_some_data(call, UINT_MAX, rxrpc_txdata_new_data);
402
403	if (saw_ack)
404		rxrpc_congestion_degrade(call);
405
406	if (did_receive &&
407	    (__rxrpc_call_state(call) == RXRPC_CALL_CLIENT_SEND_REQUEST ||
408	     __rxrpc_call_state(call) == RXRPC_CALL_SERVER_SEND_REPLY)) {
409		t = ktime_sub(call->rack_timo_at, ktime_get_real());
410		trace_rxrpc_rack(call, t);
411	}
412
413	/* Process events */
414	if (test_and_clear_bit(RXRPC_CALL_EV_ACK_LOST, &call->events))
415		rxrpc_send_ACK(call, RXRPC_ACK_PING, 0,
416			       rxrpc_propose_ack_ping_for_lost_ack);
417
418	if (call->tx_nr_lost > 0 &&
419	    __rxrpc_call_state(call) != RXRPC_CALL_CLIENT_RECV_REPLY &&
420	    !test_bit(RXRPC_CALL_TX_ALL_ACKED, &call->flags))
421		rxrpc_resend(call);
422
423	if (test_and_clear_bit(RXRPC_CALL_RX_IS_IDLE, &call->flags))
424		rxrpc_send_ACK(call, RXRPC_ACK_IDLE, 0,
425			       rxrpc_propose_ack_rx_idle);
426
427	if (call->ackr_nr_unacked > 2) {
428		if (call->rtt_count < 3)
429			rxrpc_send_ACK(call, RXRPC_ACK_PING, 0,
430				       rxrpc_propose_ack_ping_for_rtt);
431		else if (ktime_before(ktime_add_ms(call->rtt_last_req, 1000),
432				      ktime_get_real()))
433			rxrpc_send_ACK(call, RXRPC_ACK_PING, 0,
434				       rxrpc_propose_ack_ping_for_old_rtt);
435		else
436			rxrpc_send_ACK(call, RXRPC_ACK_IDLE, 0,
437				       rxrpc_propose_ack_input_data);
438	}
439
440	/* Make sure the timer is restarted */
441	if (!__rxrpc_call_is_complete(call)) {
442		ktime_t next = READ_ONCE(call->expect_term_by), delay;
443
444#define set(T) { ktime_t _t = (T); if (ktime_before(_t, next)) next = _t; }
445
446		set(call->expect_req_by);
447		set(call->expect_rx_by);
448		set(call->delay_ack_at);
449		set(call->rack_timo_at);
450		set(call->keepalive_at);
451		set(call->ping_at);
452
453		now = ktime_get_real();
454		delay = ktime_sub(next, now);
455		if (delay <= 0) {
456			rxrpc_poke_call(call, rxrpc_call_poke_timer_now);
457		} else {
458			unsigned long nowj = jiffies, delayj, nextj;
459
460			delayj = umax(nsecs_to_jiffies(delay), 1);
461			nextj = nowj + delayj;
462			if (time_before(nextj, call->timer.expires) ||
463			    !timer_pending(&call->timer)) {
464				trace_rxrpc_timer_restart(call, delay, delayj);
465				timer_reduce(&call->timer, nextj);
466			}
467		}
468	}
469
470out:
471	if (__rxrpc_call_is_complete(call)) {
472		timer_delete_sync(&call->timer);
473		if (!test_bit(RXRPC_CALL_DISCONNECTED, &call->flags))
474			rxrpc_disconnect_call(call);
475		if (call->security)
476			call->security->free_call_crypto(call);
477	} else {
478		if (did_receive &&
479		    call->peer->ackr_adv_pmtud &&
480		    call->peer->pmtud_pending)
481			rxrpc_send_probe_for_pmtud(call);
482	}
483	_leave("");
484	return true;
485
486expired:
487	if (test_bit(RXRPC_CALL_RX_HEARD, &call->flags) &&
488	    (int)call->conn->hi_serial - (int)call->rx_serial > 0) {
489		trace_rxrpc_call_reset(call);
490		rxrpc_abort_call(call, 0, RX_CALL_DEAD, -ECONNRESET,
491				 rxrpc_abort_call_reset);
492	} else {
493		rxrpc_abort_call(call, 0, RX_CALL_TIMEOUT, -ETIME,
494				 rxrpc_abort_call_timeout);
495	}
496	goto out;
497}