Merge tag 'pci-v6.16-fixes-3' of git://git.kernel.org/pub/scm/linux/kernel/git/pci/pci
[linux-block.git] / net / rxrpc / io_thread.c
CommitLineData
96b2d69b
DH
1// SPDX-License-Identifier: GPL-2.0-or-later
2/* RxRPC packet reception
3 *
a275da62 4 * Copyright (C) 2007, 2016, 2022 Red Hat, Inc. All Rights Reserved.
96b2d69b
DH
5 * Written by David Howells (dhowells@redhat.com)
6 */
7
8#define pr_fmt(fmt) KBUILD_MODNAME ": " fmt
9
10#include "ar-internal.h"
11
5e6ef4f1
DH
12static int rxrpc_input_packet_on_conn(struct rxrpc_connection *conn,
13 struct sockaddr_rxrpc *peer_srx,
14 struct sk_buff *skb);
15
446b3e14
DH
16/*
17 * handle data received on the local endpoint
18 * - may be called in interrupt context
19 *
20 * [!] Note that as this is called from the encap_rcv hook, the socket is not
21 * held locked by the caller and nothing prevents sk_user_data on the UDP from
22 * being cleared in the middle of processing this function.
23 *
24 * Called with the RCU read lock held from the IP layer via UDP.
25 */
26int rxrpc_encap_rcv(struct sock *udp_sk, struct sk_buff *skb)
27{
af094824 28 struct sk_buff_head *rx_queue;
446b3e14 29 struct rxrpc_local *local = rcu_dereference_sk_user_data(udp_sk);
bc212465 30 struct task_struct *io_thread;
446b3e14
DH
31
32 if (unlikely(!local)) {
33 kfree_skb(skb);
34 return 0;
35 }
bc212465
DH
36 io_thread = READ_ONCE(local->io_thread);
37 if (!io_thread) {
38 kfree_skb(skb);
39 return 0;
40 }
446b3e14
DH
41 if (skb->tstamp == 0)
42 skb->tstamp = ktime_get_real();
43
44 skb->mark = RXRPC_SKB_MARK_PACKET;
45 rxrpc_new_skb(skb, rxrpc_skb_new_encap_rcv);
af094824
DH
46 rx_queue = &local->rx_queue;
47#ifdef CONFIG_AF_RXRPC_INJECT_RX_DELAY
48 if (rxrpc_inject_rx_delay ||
49 !skb_queue_empty(&local->rx_delay_queue)) {
50 skb->tstamp = ktime_add_ms(skb->tstamp, rxrpc_inject_rx_delay);
51 rx_queue = &local->rx_delay_queue;
52 }
53#endif
54
55 skb_queue_tail(rx_queue, skb);
bc212465 56 wake_up_process(io_thread);
446b3e14
DH
57 return 0;
58}
59
ff734825
DH
60/*
61 * Handle an error received on the local endpoint.
62 */
63void rxrpc_error_report(struct sock *sk)
64{
65 struct rxrpc_local *local;
66 struct sk_buff *skb;
67
68 rcu_read_lock();
69 local = rcu_dereference_sk_user_data(sk);
70 if (unlikely(!local)) {
71 rcu_read_unlock();
72 return;
73 }
74
75 while ((skb = skb_dequeue(&sk->sk_error_queue))) {
76 skb->mark = RXRPC_SKB_MARK_ERROR;
77 rxrpc_new_skb(skb, rxrpc_skb_new_error_report);
78 skb_queue_tail(&local->rx_queue, skb);
79 }
80
81 rxrpc_wake_up_io_thread(local);
82 rcu_read_unlock();
83}
84
57af281e
DH
85/*
86 * Directly produce an abort from a packet.
87 */
88bool rxrpc_direct_abort(struct sk_buff *skb, enum rxrpc_abort_reason why,
89 s32 abort_code, int err)
90{
91 struct rxrpc_skb_priv *sp = rxrpc_skb(skb);
92
93 trace_rxrpc_abort(0, why, sp->hdr.cid, sp->hdr.callNumber, sp->hdr.seq,
94 abort_code, err);
95 skb->mark = RXRPC_SKB_MARK_REJECT_ABORT;
96 skb->priority = abort_code;
97 return false;
98}
99
100static bool rxrpc_bad_message(struct sk_buff *skb, enum rxrpc_abort_reason why)
101{
102 return rxrpc_direct_abort(skb, why, RX_PROTOCOL_ERROR, -EBADMSG);
103}
104
105#define just_discard true
106
96b2d69b 107/*
5e6ef4f1 108 * Process event packets targeted at a local endpoint.
96b2d69b 109 */
57af281e 110static bool rxrpc_input_version(struct rxrpc_local *local, struct sk_buff *skb)
96b2d69b 111{
5e6ef4f1
DH
112 struct rxrpc_skb_priv *sp = rxrpc_skb(skb);
113 char v;
96b2d69b 114
5e6ef4f1 115 _enter("");
96b2d69b 116
5e6ef4f1
DH
117 rxrpc_see_skb(skb, rxrpc_skb_see_version);
118 if (skb_copy_bits(skb, sizeof(struct rxrpc_wire_header), &v, 1) >= 0) {
119 if (v == 0)
120 rxrpc_send_version_request(local, &sp->hdr, skb);
96b2d69b 121 }
57af281e
DH
122
123 return true;
96b2d69b
DH
124}
125
126/*
127 * Extract the wire header from a packet and translate the byte order.
128 */
57af281e
DH
129static bool rxrpc_extract_header(struct rxrpc_skb_priv *sp,
130 struct sk_buff *skb)
96b2d69b
DH
131{
132 struct rxrpc_wire_header whdr;
4b68137a 133 struct rxrpc_ackpacket ack;
96b2d69b
DH
134
135 /* dig out the RxRPC connection details */
57af281e
DH
136 if (skb_copy_bits(skb, 0, &whdr, sizeof(whdr)) < 0)
137 return rxrpc_bad_message(skb, rxrpc_badmsg_short_hdr);
96b2d69b
DH
138
139 memset(sp, 0, sizeof(*sp));
140 sp->hdr.epoch = ntohl(whdr.epoch);
141 sp->hdr.cid = ntohl(whdr.cid);
142 sp->hdr.callNumber = ntohl(whdr.callNumber);
143 sp->hdr.seq = ntohl(whdr.seq);
144 sp->hdr.serial = ntohl(whdr.serial);
145 sp->hdr.flags = whdr.flags;
146 sp->hdr.type = whdr.type;
147 sp->hdr.userStatus = whdr.userStatus;
148 sp->hdr.securityIndex = whdr.securityIndex;
149 sp->hdr._rsvd = ntohs(whdr._rsvd);
150 sp->hdr.serviceId = ntohs(whdr.serviceId);
4b68137a
DH
151
152 if (sp->hdr.type == RXRPC_PACKET_TYPE_ACK) {
153 if (skb_copy_bits(skb, sizeof(whdr), &ack, sizeof(ack)) < 0)
154 return rxrpc_bad_message(skb, rxrpc_badmsg_short_ack);
155 sp->ack.first_ack = ntohl(ack.firstPacket);
156 sp->ack.prev_ack = ntohl(ack.previousPacket);
157 sp->ack.acked_serial = ntohl(ack.serial);
158 sp->ack.reason = ack.reason;
159 sp->ack.nr_acks = ack.nAcks;
160 }
57af281e 161 return true;
96b2d69b
DH
162}
163
164/*
165 * Extract the abort code from an ABORT packet and stash it in skb->priority.
166 */
167static bool rxrpc_extract_abort(struct sk_buff *skb)
168{
169 __be32 wtmp;
170
171 if (skb_copy_bits(skb, sizeof(struct rxrpc_wire_header),
172 &wtmp, sizeof(wtmp)) < 0)
173 return false;
174 skb->priority = ntohl(wtmp);
175 return true;
176}
177
178/*
446b3e14 179 * Process packets received on the local endpoint
96b2d69b 180 */
57af281e 181static bool rxrpc_input_packet(struct rxrpc_local *local, struct sk_buff **_skb)
96b2d69b 182{
96b2d69b 183 struct rxrpc_connection *conn;
393a2a20 184 struct sockaddr_rxrpc peer_srx;
96b2d69b
DH
185 struct rxrpc_skb_priv *sp;
186 struct rxrpc_peer *peer = NULL;
2d1faf7a 187 struct sk_buff *skb = *_skb;
57af281e 188 bool ret = false;
96b2d69b 189
96b2d69b
DH
190 skb_pull(skb, sizeof(struct udphdr));
191
96b2d69b
DH
192 sp = rxrpc_skb(skb);
193
194 /* dig out the RxRPC connection details */
57af281e
DH
195 if (!rxrpc_extract_header(sp, skb))
196 return just_discard;
96b2d69b
DH
197
198 if (IS_ENABLED(CONFIG_AF_RXRPC_INJECT_LOSS)) {
199 static int lose;
200 if ((lose++ & 7) == 7) {
201 trace_rxrpc_rx_lose(sp);
57af281e 202 return just_discard;
96b2d69b
DH
203 }
204 }
205
96b2d69b
DH
206 trace_rxrpc_rx_packet(sp);
207
208 switch (sp->hdr.type) {
209 case RXRPC_PACKET_TYPE_VERSION:
210 if (rxrpc_to_client(sp))
57af281e
DH
211 return just_discard;
212 return rxrpc_input_version(local, skb);
96b2d69b
DH
213
214 case RXRPC_PACKET_TYPE_BUSY:
215 if (rxrpc_to_server(sp))
57af281e 216 return just_discard;
96b2d69b
DH
217 fallthrough;
218 case RXRPC_PACKET_TYPE_ACK:
219 case RXRPC_PACKET_TYPE_ACKALL:
220 if (sp->hdr.callNumber == 0)
57af281e 221 return rxrpc_bad_message(skb, rxrpc_badmsg_zero_call);
96b2d69b
DH
222 break;
223 case RXRPC_PACKET_TYPE_ABORT:
224 if (!rxrpc_extract_abort(skb))
57af281e 225 return just_discard; /* Just discard if malformed */
96b2d69b
DH
226 break;
227
228 case RXRPC_PACKET_TYPE_DATA:
57af281e
DH
229 if (sp->hdr.callNumber == 0)
230 return rxrpc_bad_message(skb, rxrpc_badmsg_zero_call);
231 if (sp->hdr.seq == 0)
232 return rxrpc_bad_message(skb, rxrpc_badmsg_zero_seq);
96b2d69b
DH
233
234 /* Unshare the packet so that it can be modified for in-place
235 * decryption.
236 */
237 if (sp->hdr.securityIndex != 0) {
2d1faf7a
DH
238 skb = skb_unshare(skb, GFP_ATOMIC);
239 if (!skb) {
240 rxrpc_eaten_skb(*_skb, rxrpc_skb_eaten_by_unshare_nomem);
241 *_skb = NULL;
57af281e 242 return just_discard;
96b2d69b
DH
243 }
244
2d1faf7a
DH
245 if (skb != *_skb) {
246 rxrpc_eaten_skb(*_skb, rxrpc_skb_eaten_by_unshare);
247 *_skb = skb;
96b2d69b
DH
248 rxrpc_new_skb(skb, rxrpc_skb_new_unshared);
249 sp = rxrpc_skb(skb);
250 }
251 }
252 break;
253
254 case RXRPC_PACKET_TYPE_CHALLENGE:
255 if (rxrpc_to_server(sp))
57af281e 256 return just_discard;
96b2d69b
DH
257 break;
258 case RXRPC_PACKET_TYPE_RESPONSE:
259 if (rxrpc_to_client(sp))
57af281e 260 return just_discard;
96b2d69b
DH
261 break;
262
263 /* Packet types 9-11 should just be ignored. */
264 case RXRPC_PACKET_TYPE_PARAMS:
265 case RXRPC_PACKET_TYPE_10:
266 case RXRPC_PACKET_TYPE_11:
57af281e 267 return just_discard;
96b2d69b
DH
268
269 default:
57af281e 270 return rxrpc_bad_message(skb, rxrpc_badmsg_unsupported_packet);
96b2d69b
DH
271 }
272
273 if (sp->hdr.serviceId == 0)
57af281e 274 return rxrpc_bad_message(skb, rxrpc_badmsg_zero_service);
96b2d69b 275
393a2a20 276 if (WARN_ON_ONCE(rxrpc_extract_addr_from_skb(&peer_srx, skb) < 0))
57af281e 277 return just_discard; /* Unsupported address type. */
393a2a20
DH
278
279 if (peer_srx.transport.family != local->srx.transport.family &&
280 (peer_srx.transport.family == AF_INET &&
281 local->srx.transport.family != AF_INET6)) {
282 pr_warn_ratelimited("AF_RXRPC: Protocol mismatch %u not %u\n",
283 peer_srx.transport.family,
284 local->srx.transport.family);
57af281e 285 return just_discard; /* Wrong address type. */
5e6ef4f1
DH
286 }
287
288 if (rxrpc_to_client(sp)) {
289 rcu_read_lock();
290 conn = rxrpc_find_client_connection_rcu(local, &peer_srx, skb);
291 conn = rxrpc_get_connection_maybe(conn, rxrpc_conn_get_call_input);
292 rcu_read_unlock();
57af281e
DH
293 if (!conn)
294 return rxrpc_protocol_error(skb, rxrpc_eproto_no_client_conn);
5e6ef4f1
DH
295
296 ret = rxrpc_input_packet_on_conn(conn, &peer_srx, skb);
297 rxrpc_put_connection(conn, rxrpc_conn_put_call_input);
298 return ret;
393a2a20
DH
299 }
300
5e6ef4f1
DH
301 /* We need to look up service connections by the full protocol
302 * parameter set. We look up the peer first as an intermediate step
303 * and then the connection from the peer's tree.
304 */
cd21effb
DH
305 rcu_read_lock();
306
5e6ef4f1
DH
307 peer = rxrpc_lookup_peer_rcu(local, &peer_srx);
308 if (!peer) {
309 rcu_read_unlock();
310 return rxrpc_new_incoming_call(local, NULL, NULL, &peer_srx, skb);
96b2d69b
DH
311 }
312
5e6ef4f1
DH
313 conn = rxrpc_find_service_conn_rcu(peer, skb);
314 conn = rxrpc_get_connection_maybe(conn, rxrpc_conn_get_call_input);
96b2d69b 315 if (conn) {
5e6ef4f1
DH
316 rcu_read_unlock();
317 ret = rxrpc_input_packet_on_conn(conn, &peer_srx, skb);
318 rxrpc_put_connection(conn, rxrpc_conn_put_call_input);
319 return ret;
320 }
96b2d69b 321
5e6ef4f1
DH
322 peer = rxrpc_get_peer_maybe(peer, rxrpc_peer_get_input);
323 rcu_read_unlock();
96b2d69b 324
5e6ef4f1
DH
325 ret = rxrpc_new_incoming_call(local, peer, NULL, &peer_srx, skb);
326 rxrpc_put_peer(peer, rxrpc_peer_put_input);
57af281e 327 return ret;
5e6ef4f1 328}
96b2d69b 329
5e6ef4f1
DH
330/*
331 * Deal with a packet that's associated with an extant connection.
332 */
333static int rxrpc_input_packet_on_conn(struct rxrpc_connection *conn,
334 struct sockaddr_rxrpc *peer_srx,
335 struct sk_buff *skb)
336{
337 struct rxrpc_skb_priv *sp = rxrpc_skb(skb);
338 struct rxrpc_channel *chan;
339 struct rxrpc_call *call = NULL;
340 unsigned int channel;
96b2d69b 341
5e6ef4f1 342 if (sp->hdr.securityIndex != conn->security_ix)
57af281e
DH
343 return rxrpc_direct_abort(skb, rxrpc_eproto_wrong_security,
344 RXKADINCONSISTENCY, -EBADMSG);
96b2d69b 345
5e6ef4f1
DH
346 if (sp->hdr.serviceId != conn->service_id) {
347 int old_id;
96b2d69b 348
5e6ef4f1 349 if (!test_bit(RXRPC_CONN_PROBING_FOR_UPGRADE, &conn->flags))
57af281e
DH
350 return rxrpc_protocol_error(skb, rxrpc_eproto_reupgrade);
351
5e6ef4f1
DH
352 old_id = cmpxchg(&conn->service_id, conn->orig_service_id,
353 sp->hdr.serviceId);
5e6ef4f1
DH
354 if (old_id != conn->orig_service_id &&
355 old_id != sp->hdr.serviceId)
57af281e 356 return rxrpc_protocol_error(skb, rxrpc_eproto_bad_upgrade);
5e6ef4f1 357 }
96b2d69b 358
5e6ef4f1
DH
359 if (after(sp->hdr.serial, conn->hi_serial))
360 conn->hi_serial = sp->hdr.serial;
96b2d69b 361
5e6ef4f1
DH
362 /* It's a connection-level packet if the call number is 0. */
363 if (sp->hdr.callNumber == 0)
364 return rxrpc_input_conn_packet(conn, skb);
365
eeaedc54
DH
366 /* Deal with path MTU discovery probing. */
367 if (sp->hdr.type == RXRPC_PACKET_TYPE_ACK &&
368 conn->pmtud_probe &&
369 after_eq(sp->ack.acked_serial, conn->pmtud_probe))
370 rxrpc_input_probe_for_pmtud(conn, sp->ack.acked_serial, false);
371
5e6ef4f1
DH
372 /* Call-bound packets are routed by connection channel. */
373 channel = sp->hdr.cid & RXRPC_CHANNELMASK;
374 chan = &conn->channels[channel];
375
376 /* Ignore really old calls */
377 if (sp->hdr.callNumber < chan->last_call)
57af281e 378 return just_discard;
5e6ef4f1
DH
379
380 if (sp->hdr.callNumber == chan->last_call) {
381 if (chan->call ||
382 sp->hdr.type == RXRPC_PACKET_TYPE_ABORT)
57af281e 383 return just_discard;
96b2d69b 384
5e6ef4f1
DH
385 /* For the previous service call, if completed successfully, we
386 * discard all further packets.
387 */
388 if (rxrpc_conn_is_service(conn) &&
389 chan->last_type == RXRPC_PACKET_TYPE_ACK)
57af281e 390 return just_discard;
96b2d69b 391
5e6ef4f1
DH
392 /* But otherwise we need to retransmit the final packet from
393 * data cached in the connection record.
394 */
395 if (sp->hdr.type == RXRPC_PACKET_TYPE_DATA)
396 trace_rxrpc_rx_data(chan->call_debug_id,
397 sp->hdr.seq,
398 sp->hdr.serial,
399 sp->hdr.flags);
30df927b 400 rxrpc_conn_retransmit_call(conn, skb, channel);
57af281e 401 return just_discard;
5e6ef4f1 402 }
96b2d69b 403
9d35d880 404 call = rxrpc_try_get_call(chan->call, rxrpc_call_get_input);
5e6ef4f1
DH
405
406 if (sp->hdr.callNumber > chan->call_id) {
407 if (rxrpc_to_client(sp)) {
408 rxrpc_put_call(call, rxrpc_call_put_input);
57af281e
DH
409 return rxrpc_protocol_error(skb,
410 rxrpc_eproto_unexpected_implicit_end);
5e6ef4f1 411 }
cd21effb 412
96b2d69b 413 if (call) {
5e6ef4f1
DH
414 rxrpc_implicit_end_call(call, skb);
415 rxrpc_put_call(call, rxrpc_call_put_input);
416 call = NULL;
96b2d69b
DH
417 }
418 }
419
cd21effb 420 if (!call) {
5e6ef4f1 421 if (rxrpc_to_client(sp))
57af281e
DH
422 return rxrpc_protocol_error(skb, rxrpc_eproto_no_client_call);
423 return rxrpc_new_incoming_call(conn->local, conn->peer, conn,
424 peer_srx, skb);
96b2d69b
DH
425 }
426
9e3cccd1 427 rxrpc_queue_rx_call_packet(call, skb);
cd21effb 428 rxrpc_put_call(call, rxrpc_call_put_input);
9e3cccd1 429 return true;
96b2d69b 430}
a275da62
DH
431
432/*
433 * I/O and event handling thread.
434 */
435int rxrpc_io_thread(void *data)
436{
f2cce89a 437 struct rxrpc_connection *conn;
a275da62
DH
438 struct sk_buff_head rx_queue;
439 struct rxrpc_local *local = data;
15f661dc 440 struct rxrpc_call *call;
a275da62 441 struct sk_buff *skb;
af094824
DH
442#ifdef CONFIG_AF_RXRPC_INJECT_RX_DELAY
443 ktime_t now;
444#endif
743d1768 445 bool should_stop;
9e3cccd1
DH
446 LIST_HEAD(conn_attend_q);
447 LIST_HEAD(call_attend_q);
a275da62 448
8fbcc833
DH
449 complete(&local->io_thread_ready);
450
a275da62
DH
451 skb_queue_head_init(&rx_queue);
452
453 set_user_nice(current, MIN_NICE);
454
455 for (;;) {
456 rxrpc_inc_stat(local->rxnet, stat_io_loop);
457
9e3cccd1
DH
458 /* Inject a delay into packets if requested. */
459#ifdef CONFIG_AF_RXRPC_INJECT_RX_DELAY
460 now = ktime_get_real();
461 while ((skb = skb_peek(&local->rx_delay_queue))) {
462 if (ktime_before(now, skb->tstamp))
463 break;
464 skb = skb_dequeue(&local->rx_delay_queue);
465 skb_queue_tail(&local->rx_queue, skb);
f2cce89a 466 }
9e3cccd1 467#endif
f2cce89a 468
9e3cccd1
DH
469 if (!skb_queue_empty(&local->rx_queue)) {
470 spin_lock_irq(&local->rx_queue.lock);
471 skb_queue_splice_tail_init(&local->rx_queue, &rx_queue);
472 spin_unlock_irq(&local->rx_queue.lock);
7c482665 473 trace_rxrpc_iothread_rx(local, skb_queue_len(&rx_queue));
15f661dc
DH
474 }
475
9e3cccd1
DH
476 /* Distribute packets and errors. */
477 while ((skb = __skb_dequeue(&rx_queue))) {
2953d3b8 478 struct rxrpc_skb_priv *sp = rxrpc_skb(skb);
446b3e14
DH
479 switch (skb->mark) {
480 case RXRPC_SKB_MARK_PACKET:
2d1faf7a 481 skb->priority = 0;
57af281e
DH
482 if (!rxrpc_input_packet(local, &skb))
483 rxrpc_reject_packet(local, skb);
2d1faf7a
DH
484 trace_rxrpc_rx_done(skb->mark, skb->priority);
485 rxrpc_free_skb(skb, rxrpc_skb_put_input);
446b3e14 486 break;
ff734825
DH
487 case RXRPC_SKB_MARK_ERROR:
488 rxrpc_input_error(local, skb);
489 rxrpc_free_skb(skb, rxrpc_skb_put_error_report);
490 break;
2953d3b8 491 case RXRPC_SKB_MARK_SERVICE_CONN_SECURED:
5800b1cf
DH
492 rxrpc_input_conn_event(sp->poke_conn, skb);
493 rxrpc_put_connection(sp->poke_conn, rxrpc_conn_put_poke);
2953d3b8 494 rxrpc_free_skb(skb, rxrpc_skb_put_conn_secured);
f2cce89a 495 break;
446b3e14
DH
496 default:
497 WARN_ON_ONCE(1);
498 rxrpc_free_skb(skb, rxrpc_skb_put_unknown);
499 break;
500 }
a275da62
DH
501 }
502
9e3cccd1 503 /* Deal with connections that want immediate attention. */
5800b1cf
DH
504 if (!list_empty_careful(&local->conn_attend_q)) {
505 spin_lock_irq(&local->lock);
506 list_splice_tail_init(&local->conn_attend_q, &conn_attend_q);
507 spin_unlock_irq(&local->lock);
508 }
9e3cccd1
DH
509
510 while ((conn = list_first_entry_or_null(&conn_attend_q,
511 struct rxrpc_connection,
512 attend_link))) {
d920270a 513 spin_lock_irq(&local->lock);
9e3cccd1 514 list_del_init(&conn->attend_link);
d920270a 515 spin_unlock_irq(&local->lock);
9e3cccd1
DH
516 rxrpc_input_conn_event(conn, NULL);
517 rxrpc_put_connection(conn, rxrpc_conn_put_poke);
af094824 518 }
af094824 519
9e3cccd1
DH
520 if (test_and_clear_bit(RXRPC_CLIENT_CONN_REAP_TIMER,
521 &local->client_conn_flags))
522 rxrpc_discard_expired_client_conns(local);
523
524 /* Deal with calls that want immediate attention. */
a2ea9a90 525 spin_lock_irq(&local->lock);
9e3cccd1 526 list_splice_tail_init(&local->call_attend_q, &call_attend_q);
a2ea9a90 527 spin_unlock_irq(&local->lock);
9e3cccd1
DH
528
529 while ((call = list_first_entry_or_null(&call_attend_q,
530 struct rxrpc_call,
531 attend_link))) {
d920270a 532 spin_lock_irq(&local->lock);
9e3cccd1 533 list_del_init(&call->attend_link);
d920270a 534 spin_unlock_irq(&local->lock);
9e3cccd1
DH
535 trace_rxrpc_call_poked(call);
536 rxrpc_input_call_event(call);
537 rxrpc_put_call(call, rxrpc_call_put_poke);
a275da62
DH
538 }
539
9e3cccd1
DH
540 if (!list_empty(&local->new_client_calls))
541 rxrpc_connect_client_calls(local);
542
a275da62 543 set_current_state(TASK_INTERRUPTIBLE);
743d1768 544 should_stop = kthread_should_stop();
15f661dc 545 if (!skb_queue_empty(&local->rx_queue) ||
f2cce89a 546 !list_empty(&local->call_attend_q) ||
9d35d880
DH
547 !list_empty(&local->conn_attend_q) ||
548 !list_empty(&local->new_client_calls) ||
549 test_bit(RXRPC_CLIENT_CONN_REAP_TIMER,
550 &local->client_conn_flags)) {
a275da62
DH
551 __set_current_state(TASK_RUNNING);
552 continue;
553 }
554
743d1768 555 if (should_stop)
a275da62 556 break;
af094824
DH
557
558#ifdef CONFIG_AF_RXRPC_INJECT_RX_DELAY
559 skb = skb_peek(&local->rx_delay_queue);
560 if (skb) {
561 unsigned long timeout;
562 ktime_t tstamp = skb->tstamp;
563 ktime_t now = ktime_get_real();
564 s64 delay_ns = ktime_to_ns(ktime_sub(tstamp, now));
565
566 if (delay_ns <= 0) {
567 __set_current_state(TASK_RUNNING);
568 continue;
569 }
570
571 timeout = nsecs_to_jiffies(delay_ns);
29e03ec7 572 timeout = umax(timeout, 1);
af094824
DH
573 schedule_timeout(timeout);
574 __set_current_state(TASK_RUNNING);
575 continue;
576 }
577#endif
578
a275da62
DH
579 schedule();
580 }
581
582 __set_current_state(TASK_RUNNING);
583 rxrpc_see_local(local, rxrpc_local_stop);
584 rxrpc_destroy_local(local);
bc212465 585 WRITE_ONCE(local->io_thread, NULL);
a275da62
DH
586 rxrpc_see_local(local, rxrpc_local_stopped);
587 return 0;
588}