net/rxrpc/io_thread.c

   1 // SPDX-License-Identifier: GPL-2.0-or-later
   2 /* RxRPC packet reception
   3  *
   4  * Copyright (C) 2007, 2016, 2022 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 "ar-internal.h"
  11
  12 static int rxrpc_input_packet_on_conn(struct rxrpc_connection *conn,
  13                                       struct sockaddr_rxrpc *peer_srx,
  14                                       struct sk_buff *skb);
  15
  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  */
  26 int rxrpc_encap_rcv(struct sock *udp_sk, struct sk_buff *skb)
  27 {
  28         struct sk_buff_head *rx_queue;
  29         struct rxrpc_local *local = rcu_dereference_sk_user_data(udp_sk);
  30         struct task_struct *io_thread;
  31
  32         if (unlikely(!local)) {
  33                 kfree_skb(skb);
  34                 return 0;
  35         }
  36         io_thread = READ_ONCE(local->io_thread);
  37         if (!io_thread) {
  38                 kfree_skb(skb);
  39                 return 0;
  40         }
  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);
  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);
  56         wake_up_process(io_thread);
  57         return 0;
  58 }
  59
  60 /*
  61  * Handle an error received on the local endpoint.
  62  */
  63 void 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
  85 /*
  86  * Directly produce an abort from a packet.
  87  */
  88 bool 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
 100 static 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
 107 /*
 108  * Process event packets targeted at a local endpoint.
 109  */
 110 static bool rxrpc_input_version(struct rxrpc_local *local, struct sk_buff *skb)
 111 {
 112         struct rxrpc_skb_priv *sp = rxrpc_skb(skb);
 113         char v;
 114
 115         _enter("");
 116
 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);
 121         }
 122
 123         return true;
 124 }
 125
 126 /*
 127  * Extract the wire header from a packet and translate the byte order.
 128  */
 129 static bool rxrpc_extract_header(struct rxrpc_skb_priv *sp,
 130                                  struct sk_buff *skb)
 131 {
 132         struct rxrpc_wire_header whdr;
 133         struct rxrpc_ackpacket ack;
 134
 135         /* dig out the RxRPC connection details */
 136         if (skb_copy_bits(skb, 0, &whdr, sizeof(whdr)) < 0)
 137                 return rxrpc_bad_message(skb, rxrpc_badmsg_short_hdr);
 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);
 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         }
 161         return true;
 162 }
 163
 164 /*
 165  * Extract the abort code from an ABORT packet and stash it in skb->priority.
 166  */
 167 static 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 /*
 179  * Process packets received on the local endpoint
 180  */
 181 static bool rxrpc_input_packet(struct rxrpc_local *local, struct sk_buff **_skb)
 182 {
 183         struct rxrpc_connection *conn;
 184         struct sockaddr_rxrpc peer_srx;
 185         struct rxrpc_skb_priv *sp;
 186         struct rxrpc_peer *peer = NULL;
 187         struct sk_buff *skb = *_skb;
 188         bool ret = false;
 189
 190         skb_pull(skb, sizeof(struct udphdr));
 191
 192         sp = rxrpc_skb(skb);
 193
 194         /* dig out the RxRPC connection details */
 195         if (!rxrpc_extract_header(sp, skb))
 196                 return just_discard;
 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);
 202                         return just_discard;
 203                 }
 204         }
 205
 206         trace_rxrpc_rx_packet(sp);
 207
 208         switch (sp->hdr.type) {
 209         case RXRPC_PACKET_TYPE_VERSION:
 210                 if (rxrpc_to_client(sp))
 211                         return just_discard;
 212                 return rxrpc_input_version(local, skb);
 213
 214         case RXRPC_PACKET_TYPE_BUSY:
 215                 if (rxrpc_to_server(sp))
 216                         return just_discard;
 217                 fallthrough;
 218         case RXRPC_PACKET_TYPE_ACK:
 219         case RXRPC_PACKET_TYPE_ACKALL:
 220                 if (sp->hdr.callNumber == 0)
 221                         return rxrpc_bad_message(skb, rxrpc_badmsg_zero_call);
 222                 break;
 223         case RXRPC_PACKET_TYPE_ABORT:
 224                 if (!rxrpc_extract_abort(skb))
 225                         return just_discard; /* Just discard if malformed */
 226                 break;
 227
 228         case RXRPC_PACKET_TYPE_DATA:
 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);
 233
 234                 /* Unshare the packet so that it can be modified for in-place
 235                  * decryption.
 236                  */
 237                 if (sp->hdr.securityIndex != 0) {
 238                         skb = skb_unshare(skb, GFP_ATOMIC);
 239                         if (!skb) {
 240                                 rxrpc_eaten_skb(*_skb, rxrpc_skb_eaten_by_unshare_nomem);
 241                                 *_skb = NULL;
 242                                 return just_discard;
 243                         }
 244
 245                         if (skb != *_skb) {
 246                                 rxrpc_eaten_skb(*_skb, rxrpc_skb_eaten_by_unshare);
 247                                 *_skb = skb;
 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))
 256                         return just_discard;
 257                 break;
 258         case RXRPC_PACKET_TYPE_RESPONSE:
 259                 if (rxrpc_to_client(sp))
 260                         return just_discard;
 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:
 267                 return just_discard;
 268
 269         default:
 270                 return rxrpc_bad_message(skb, rxrpc_badmsg_unsupported_packet);
 271         }
 272
 273         if (sp->hdr.serviceId == 0)
 274                 return rxrpc_bad_message(skb, rxrpc_badmsg_zero_service);
 275
 276         if (WARN_ON_ONCE(rxrpc_extract_addr_from_skb(&peer_srx, skb) < 0))
 277                 return just_discard; /* Unsupported address type. */
 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);
 285                 return just_discard; /* Wrong address type. */
 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();
 293                 if (!conn)
 294                         return rxrpc_protocol_error(skb, rxrpc_eproto_no_client_conn);
 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;
 299         }
 300
 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          */
 305         rcu_read_lock();
 306
 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);
 311         }
 312
 313         conn = rxrpc_find_service_conn_rcu(peer, skb);
 314         conn = rxrpc_get_connection_maybe(conn, rxrpc_conn_get_call_input);
 315         if (conn) {
 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         }
 321
 322         peer = rxrpc_get_peer_maybe(peer, rxrpc_peer_get_input);
 323         rcu_read_unlock();
 324
 325         ret = rxrpc_new_incoming_call(local, peer, NULL, &peer_srx, skb);
 326         rxrpc_put_peer(peer, rxrpc_peer_put_input);
 327         return ret;
 328 }
 329
 330 /*
 331  * Deal with a packet that's associated with an extant connection.
 332  */
 333 static 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;
 341
 342         if (sp->hdr.securityIndex != conn->security_ix)
 343                 return rxrpc_direct_abort(skb, rxrpc_eproto_wrong_security,
 344                                           RXKADINCONSISTENCY, -EBADMSG);
 345
 346         if (sp->hdr.serviceId != conn->service_id) {
 347                 int old_id;
 348
 349                 if (!test_bit(RXRPC_CONN_PROBING_FOR_UPGRADE, &conn->flags))
 350                         return rxrpc_protocol_error(skb, rxrpc_eproto_reupgrade);
 351
 352                 old_id = cmpxchg(&conn->service_id, conn->orig_service_id,
 353                                  sp->hdr.serviceId);
 354                 if (old_id != conn->orig_service_id &&
 355                     old_id != sp->hdr.serviceId)
 356                         return rxrpc_protocol_error(skb, rxrpc_eproto_bad_upgrade);
 357         }
 358
 359         if (after(sp->hdr.serial, conn->hi_serial))
 360                 conn->hi_serial = sp->hdr.serial;
 361
 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
 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
 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)
 378                 return just_discard;
 379
 380         if (sp->hdr.callNumber == chan->last_call) {
 381                 if (chan->call ||
 382                     sp->hdr.type == RXRPC_PACKET_TYPE_ABORT)
 383                         return just_discard;
 384
 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)
 390                         return just_discard;
 391
 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);
 400                 rxrpc_conn_retransmit_call(conn, skb, channel);
 401                 return just_discard;
 402         }
 403
 404         call = rxrpc_try_get_call(chan->call, rxrpc_call_get_input);
 405
 406         if (sp->hdr.callNumber > chan->call_id) {
 407                 if (rxrpc_to_client(sp)) {
 408                         rxrpc_put_call(call, rxrpc_call_put_input);
 409                         return rxrpc_protocol_error(skb,
 410                                                     rxrpc_eproto_unexpected_implicit_end);
 411                 }
 412
 413                 if (call) {
 414                         rxrpc_implicit_end_call(call, skb);
 415                         rxrpc_put_call(call, rxrpc_call_put_input);
 416                         call = NULL;
 417                 }
 418         }
 419
 420         if (!call) {
 421                 if (rxrpc_to_client(sp))
 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);
 425         }
 426
 427         rxrpc_queue_rx_call_packet(call, skb);
 428         rxrpc_put_call(call, rxrpc_call_put_input);
 429         return true;
 430 }
 431
 432 /*
 433  * I/O and event handling thread.
 434  */
 435 int rxrpc_io_thread(void *data)
 436 {
 437         struct rxrpc_connection *conn;
 438         struct sk_buff_head rx_queue;
 439         struct rxrpc_local *local = data;
 440         struct rxrpc_call *call;
 441         struct sk_buff *skb;
 442 #ifdef CONFIG_AF_RXRPC_INJECT_RX_DELAY
 443         ktime_t now;
 444 #endif
 445         bool should_stop;
 446         LIST_HEAD(conn_attend_q);
 447         LIST_HEAD(call_attend_q);
 448
 449         complete(&local->io_thread_ready);
 450
 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
 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);
 466                 }
 467 #endif
 468
 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);
 473                         trace_rxrpc_iothread_rx(local, skb_queue_len(&rx_queue));
 474                 }
 475
 476                 /* Distribute packets and errors. */
 477                 while ((skb = __skb_dequeue(&rx_queue))) {
 478                         struct rxrpc_skb_priv *sp = rxrpc_skb(skb);
 479                         switch (skb->mark) {
 480                         case RXRPC_SKB_MARK_PACKET:
 481                                 skb->priority = 0;
 482                                 if (!rxrpc_input_packet(local, &skb))
 483                                         rxrpc_reject_packet(local, skb);
 484                                 trace_rxrpc_rx_done(skb->mark, skb->priority);
 485                                 rxrpc_free_skb(skb, rxrpc_skb_put_input);
 486                                 break;
 487                         case RXRPC_SKB_MARK_ERROR:
 488                                 rxrpc_input_error(local, skb);
 489                                 rxrpc_free_skb(skb, rxrpc_skb_put_error_report);
 490                                 break;
 491                         case RXRPC_SKB_MARK_SERVICE_CONN_SECURED:
 492                                 rxrpc_input_conn_event(sp->poke_conn, skb);
 493                                 rxrpc_put_connection(sp->poke_conn, rxrpc_conn_put_poke);
 494                                 rxrpc_free_skb(skb, rxrpc_skb_put_conn_secured);
 495                                 break;
 496                         default:
 497                                 WARN_ON_ONCE(1);
 498                                 rxrpc_free_skb(skb, rxrpc_skb_put_unknown);
 499                                 break;
 500                         }
 501                 }
 502
 503                 /* Deal with connections that want immediate attention. */
 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                 }
 509
 510                 while ((conn = list_first_entry_or_null(&conn_attend_q,
 511                                                         struct rxrpc_connection,
 512                                                         attend_link))) {
 513                         spin_lock_irq(&local->lock);
 514                         list_del_init(&conn->attend_link);
 515                         spin_unlock_irq(&local->lock);
 516                         rxrpc_input_conn_event(conn, NULL);
 517                         rxrpc_put_connection(conn, rxrpc_conn_put_poke);
 518                 }
 519
 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. */
 525                 spin_lock_irq(&local->lock);
 526                 list_splice_tail_init(&local->call_attend_q, &call_attend_q);
 527                 spin_unlock_irq(&local->lock);
 528
 529                 while ((call = list_first_entry_or_null(&call_attend_q,
 530                                                         struct rxrpc_call,
 531                                                         attend_link))) {
 532                         spin_lock_irq(&local->lock);
 533                         list_del_init(&call->attend_link);
 534                         spin_unlock_irq(&local->lock);
 535                         trace_rxrpc_call_poked(call);
 536                         rxrpc_input_call_event(call);
 537                         rxrpc_put_call(call, rxrpc_call_put_poke);
 538                 }
 539
 540                 if (!list_empty(&local->new_client_calls))
 541                         rxrpc_connect_client_calls(local);
 542
 543                 set_current_state(TASK_INTERRUPTIBLE);
 544                 should_stop = kthread_should_stop();
 545                 if (!skb_queue_empty(&local->rx_queue) ||
 546                     !list_empty(&local->call_attend_q) ||
 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)) {
 551                         __set_current_state(TASK_RUNNING);
 552                         continue;
 553                 }
 554
 555                 if (should_stop)
 556                         break;
 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);
 572                         timeout = umax(timeout, 1);
 573                         schedule_timeout(timeout);
 574                         __set_current_state(TASK_RUNNING);
 575                         continue;
 576                 }
 577 #endif
 578
 579                 schedule();
 580         }
 581
 582         __set_current_state(TASK_RUNNING);
 583         rxrpc_see_local(local, rxrpc_local_stop);
 584         rxrpc_destroy_local(local);
 585         WRITE_ONCE(local->io_thread, NULL);
 586         rxrpc_see_local(local, rxrpc_local_stopped);
 587         return 0;
 588 }