1 /* RxRPC virtual connection handler
3 * Copyright (C) 2007 Red Hat, Inc. All Rights Reserved.
4 * Written by David Howells (dhowells@redhat.com)
6 * This program is free software; you can redistribute it and/or
7 * modify it under the terms of the GNU General Public License
8 * as published by the Free Software Foundation; either version
9 * 2 of the License, or (at your option) any later version.
12 #define pr_fmt(fmt) KBUILD_MODNAME ": " fmt
14 #include <linux/module.h>
15 #include <linux/slab.h>
16 #include <linux/net.h>
17 #include <linux/skbuff.h>
18 #include <linux/crypto.h>
20 #include <net/af_rxrpc.h>
21 #include "ar-internal.h"
24 * Time till a connection expires after last use (in seconds).
26 unsigned int rxrpc_connection_expiry = 10 * 60;
28 static void rxrpc_connection_reaper(struct work_struct *work);
30 LIST_HEAD(rxrpc_connections);
31 DEFINE_RWLOCK(rxrpc_connection_lock);
32 static DECLARE_DELAYED_WORK(rxrpc_connection_reap, rxrpc_connection_reaper);
35 * allocate a new client connection bundle
37 static struct rxrpc_conn_bundle *rxrpc_alloc_bundle(gfp_t gfp)
39 struct rxrpc_conn_bundle *bundle;
43 bundle = kzalloc(sizeof(struct rxrpc_conn_bundle), gfp);
45 INIT_LIST_HEAD(&bundle->unused_conns);
46 INIT_LIST_HEAD(&bundle->avail_conns);
47 INIT_LIST_HEAD(&bundle->busy_conns);
48 init_waitqueue_head(&bundle->chanwait);
49 atomic_set(&bundle->usage, 1);
52 _leave(" = %p", bundle);
57 * compare bundle parameters with what we're looking for
58 * - return -ve, 0 or +ve
61 int rxrpc_cmp_bundle(const struct rxrpc_conn_bundle *bundle,
62 struct key *key, u16 service_id)
64 return (bundle->service_id - service_id) ?:
65 ((unsigned long)bundle->key - (unsigned long)key);
69 * get bundle of client connections that a client socket can make use of
71 struct rxrpc_conn_bundle *rxrpc_get_bundle(struct rxrpc_sock *rx,
72 struct rxrpc_transport *trans,
77 struct rxrpc_conn_bundle *bundle, *candidate;
78 struct rb_node *p, *parent, **pp;
80 _enter("%p{%x},%x,%hx,",
81 rx, key_serial(key), trans->debug_id, service_id);
83 /* search the extant bundles first for one that matches the specified
85 spin_lock(&trans->client_lock);
87 p = trans->bundles.rb_node;
89 bundle = rb_entry(p, struct rxrpc_conn_bundle, node);
91 if (rxrpc_cmp_bundle(bundle, key, service_id) < 0)
93 else if (rxrpc_cmp_bundle(bundle, key, service_id) > 0)
96 goto found_extant_bundle;
99 spin_unlock(&trans->client_lock);
101 /* not yet present - create a candidate for a new record and then
103 candidate = rxrpc_alloc_bundle(gfp);
105 _leave(" = -ENOMEM");
106 return ERR_PTR(-ENOMEM);
109 candidate->key = key_get(key);
110 candidate->service_id = service_id;
112 spin_lock(&trans->client_lock);
114 pp = &trans->bundles.rb_node;
118 bundle = rb_entry(parent, struct rxrpc_conn_bundle, node);
120 if (rxrpc_cmp_bundle(bundle, key, service_id) < 0)
121 pp = &(*pp)->rb_left;
122 else if (rxrpc_cmp_bundle(bundle, key, service_id) > 0)
123 pp = &(*pp)->rb_right;
125 goto found_extant_second;
128 /* second search also failed; add the new bundle */
132 rb_link_node(&bundle->node, parent, pp);
133 rb_insert_color(&bundle->node, &trans->bundles);
134 spin_unlock(&trans->client_lock);
135 _net("BUNDLE new on trans %d", trans->debug_id);
136 _leave(" = %p [new]", bundle);
139 /* we found the bundle in the list immediately */
141 atomic_inc(&bundle->usage);
142 spin_unlock(&trans->client_lock);
143 _net("BUNDLE old on trans %d", trans->debug_id);
144 _leave(" = %p [extant %d]", bundle, atomic_read(&bundle->usage));
147 /* we found the bundle on the second time through the list */
149 atomic_inc(&bundle->usage);
150 spin_unlock(&trans->client_lock);
152 _net("BUNDLE old2 on trans %d", trans->debug_id);
153 _leave(" = %p [second %d]", bundle, atomic_read(&bundle->usage));
160 void rxrpc_put_bundle(struct rxrpc_transport *trans,
161 struct rxrpc_conn_bundle *bundle)
163 _enter("%p,%p{%d}",trans, bundle, atomic_read(&bundle->usage));
165 if (atomic_dec_and_lock(&bundle->usage, &trans->client_lock)) {
166 _debug("Destroy bundle");
167 rb_erase(&bundle->node, &trans->bundles);
168 spin_unlock(&trans->client_lock);
169 ASSERT(list_empty(&bundle->unused_conns));
170 ASSERT(list_empty(&bundle->avail_conns));
171 ASSERT(list_empty(&bundle->busy_conns));
172 ASSERTCMP(bundle->num_conns, ==, 0);
173 key_put(bundle->key);
181 * allocate a new connection
183 static struct rxrpc_connection *rxrpc_alloc_connection(gfp_t gfp)
185 struct rxrpc_connection *conn;
189 conn = kzalloc(sizeof(struct rxrpc_connection), gfp);
191 INIT_WORK(&conn->processor, &rxrpc_process_connection);
192 INIT_LIST_HEAD(&conn->bundle_link);
193 conn->calls = RB_ROOT;
194 skb_queue_head_init(&conn->rx_queue);
195 conn->security = &rxrpc_no_security;
196 rwlock_init(&conn->lock);
197 spin_lock_init(&conn->state_lock);
198 atomic_set(&conn->usage, 1);
199 conn->debug_id = atomic_inc_return(&rxrpc_debug_id);
200 conn->avail_calls = RXRPC_MAXCALLS;
201 conn->size_align = 4;
202 conn->header_size = sizeof(struct rxrpc_wire_header);
205 _leave(" = %p{%d}", conn, conn ? conn->debug_id : 0);
210 * assign a connection ID to a connection and add it to the transport's
211 * connection lookup tree
212 * - called with transport client lock held
214 static void rxrpc_assign_connection_id(struct rxrpc_connection *conn)
216 struct rxrpc_connection *xconn;
217 struct rb_node *parent, **p;
223 epoch = conn->proto.epoch;
225 write_lock_bh(&conn->trans->conn_lock);
227 conn->trans->conn_idcounter += RXRPC_CID_INC;
228 if (conn->trans->conn_idcounter < RXRPC_CID_INC)
229 conn->trans->conn_idcounter = RXRPC_CID_INC;
230 cid = conn->trans->conn_idcounter;
234 p = &conn->trans->client_conns.rb_node;
238 xconn = rb_entry(parent, struct rxrpc_connection, node);
240 if (epoch < xconn->proto.epoch)
242 else if (epoch > xconn->proto.epoch)
244 else if (cid < xconn->proto.cid)
246 else if (cid > xconn->proto.cid)
252 /* we've found a suitable hole - arrange for this connection to occupy
254 rb_link_node(&conn->node, parent, p);
255 rb_insert_color(&conn->node, &conn->trans->client_conns);
257 conn->proto.cid = cid;
258 write_unlock_bh(&conn->trans->conn_lock);
259 _leave(" [CID %x]", cid);
262 /* we found a connection with the proposed ID - walk the tree from that
263 * point looking for the next unused ID */
266 cid += RXRPC_CID_INC;
267 if (cid < RXRPC_CID_INC) {
269 conn->trans->conn_idcounter = cid;
270 goto attempt_insertion;
273 parent = rb_next(parent);
275 goto attempt_insertion;
277 xconn = rb_entry(parent, struct rxrpc_connection, node);
278 if (epoch < xconn->proto.epoch ||
279 cid < xconn->proto.cid)
280 goto attempt_insertion;
285 * add a call to a connection's call-by-ID tree
287 static void rxrpc_add_call_ID_to_conn(struct rxrpc_connection *conn,
288 struct rxrpc_call *call)
290 struct rxrpc_call *xcall;
291 struct rb_node *parent, **p;
294 write_lock_bh(&conn->lock);
296 call_id = call->call_id;
297 p = &conn->calls.rb_node;
301 xcall = rb_entry(parent, struct rxrpc_call, conn_node);
303 if (call_id < xcall->call_id)
305 else if (call_id > xcall->call_id)
311 rb_link_node(&call->conn_node, parent, p);
312 rb_insert_color(&call->conn_node, &conn->calls);
314 write_unlock_bh(&conn->lock);
318 * connect a call on an exclusive connection
320 static int rxrpc_connect_exclusive(struct rxrpc_sock *rx,
321 struct rxrpc_conn_parameters *cp,
322 struct rxrpc_transport *trans,
323 struct rxrpc_call *call,
326 struct rxrpc_connection *conn;
331 conn = rxrpc_alloc_connection(gfp);
333 _leave(" = -ENOMEM");
340 conn->proto.local = cp->local;
341 conn->proto.epoch = rxrpc_epoch;
343 conn->proto.in_clientflag = 0;
344 conn->proto.family = cp->peer->srx.transport.family;
345 conn->out_clientflag = RXRPC_CLIENT_INITIATED;
346 conn->state = RXRPC_CONN_CLIENT;
347 conn->avail_calls = RXRPC_MAXCALLS - 1;
349 key_get(conn->params.key);
351 ret = rxrpc_init_client_conn_security(conn);
353 key_put(conn->params.key);
355 _leave(" = %d [key]", ret);
359 write_lock_bh(&rxrpc_connection_lock);
360 list_add_tail(&conn->link, &rxrpc_connections);
361 write_unlock_bh(&rxrpc_connection_lock);
363 spin_lock(&trans->client_lock);
364 atomic_inc(&trans->usage);
366 _net("CONNECT EXCL new %d on TRANS %d",
367 conn->debug_id, conn->trans->debug_id);
369 rxrpc_assign_connection_id(conn);
371 /* Since no one else can use the connection, we just use the first
375 atomic_inc(&conn->usage);
376 conn->channels[chan] = call;
377 conn->call_counter = 1;
379 call->channel = chan;
380 call->cid = conn->proto.cid | chan;
383 _net("CONNECT client on conn %d chan %d as call %x",
384 conn->debug_id, chan, call->call_id);
386 spin_unlock(&trans->client_lock);
388 rxrpc_add_call_ID_to_conn(conn, call);
394 * find a connection for a call
395 * - called in process context with IRQs enabled
397 int rxrpc_connect_call(struct rxrpc_sock *rx,
398 struct rxrpc_conn_parameters *cp,
399 struct rxrpc_transport *trans,
400 struct rxrpc_conn_bundle *bundle,
401 struct rxrpc_call *call,
404 struct rxrpc_connection *conn, *candidate;
407 DECLARE_WAITQUEUE(myself, current);
409 _enter("%p,%lx,", rx, call->user_call_ID);
412 return rxrpc_connect_exclusive(rx, cp, trans, call, gfp);
414 spin_lock(&trans->client_lock);
416 /* see if the bundle has a call slot available */
417 if (!list_empty(&bundle->avail_conns)) {
419 conn = list_entry(bundle->avail_conns.next,
420 struct rxrpc_connection,
422 if (conn->state >= RXRPC_CONN_REMOTELY_ABORTED) {
423 list_del_init(&conn->bundle_link);
427 if (--conn->avail_calls == 0)
428 list_move(&conn->bundle_link,
429 &bundle->busy_conns);
430 ASSERTCMP(conn->avail_calls, <, RXRPC_MAXCALLS);
431 ASSERT(conn->channels[0] == NULL ||
432 conn->channels[1] == NULL ||
433 conn->channels[2] == NULL ||
434 conn->channels[3] == NULL);
435 atomic_inc(&conn->usage);
439 if (!list_empty(&bundle->unused_conns)) {
441 conn = list_entry(bundle->unused_conns.next,
442 struct rxrpc_connection,
444 if (conn->state >= RXRPC_CONN_REMOTELY_ABORTED) {
445 list_del_init(&conn->bundle_link);
449 ASSERTCMP(conn->avail_calls, ==, RXRPC_MAXCALLS);
450 conn->avail_calls = RXRPC_MAXCALLS - 1;
451 ASSERT(conn->channels[0] == NULL &&
452 conn->channels[1] == NULL &&
453 conn->channels[2] == NULL &&
454 conn->channels[3] == NULL);
455 atomic_inc(&conn->usage);
456 list_move(&conn->bundle_link, &bundle->avail_conns);
460 /* need to allocate a new connection */
461 _debug("get new conn [%d]", bundle->num_conns);
463 spin_unlock(&trans->client_lock);
465 if (signal_pending(current))
468 if (bundle->num_conns >= 20) {
469 _debug("too many conns");
471 if (!gfpflags_allow_blocking(gfp)) {
472 _leave(" = -EAGAIN");
476 add_wait_queue(&bundle->chanwait, &myself);
478 set_current_state(TASK_INTERRUPTIBLE);
479 if (bundle->num_conns < 20 ||
480 !list_empty(&bundle->unused_conns) ||
481 !list_empty(&bundle->avail_conns))
483 if (signal_pending(current))
484 goto interrupted_dequeue;
487 remove_wait_queue(&bundle->chanwait, &myself);
488 __set_current_state(TASK_RUNNING);
489 spin_lock(&trans->client_lock);
493 /* not yet present - create a candidate for a new connection and then
495 candidate = rxrpc_alloc_connection(gfp);
497 _leave(" = -ENOMEM");
501 candidate->trans = trans;
502 candidate->bundle = bundle;
503 candidate->params = *cp;
504 candidate->proto.local = cp->local;
505 candidate->proto.epoch = rxrpc_epoch;
506 candidate->proto.cid = 0;
507 candidate->proto.in_clientflag = 0;
508 candidate->proto.family = cp->peer->srx.transport.family;
509 candidate->out_clientflag = RXRPC_CLIENT_INITIATED;
510 candidate->state = RXRPC_CONN_CLIENT;
511 candidate->avail_calls = RXRPC_MAXCALLS;
513 key_get(candidate->params.key);
515 ret = rxrpc_init_client_conn_security(candidate);
517 key_put(candidate->params.key);
519 _leave(" = %d [key]", ret);
523 write_lock_bh(&rxrpc_connection_lock);
524 list_add_tail(&candidate->link, &rxrpc_connections);
525 write_unlock_bh(&rxrpc_connection_lock);
527 spin_lock(&trans->client_lock);
529 list_add(&candidate->bundle_link, &bundle->unused_conns);
531 atomic_inc(&bundle->usage);
532 atomic_inc(&trans->usage);
534 _net("CONNECT new %d on TRANS %d",
535 candidate->debug_id, candidate->trans->debug_id);
537 rxrpc_assign_connection_id(candidate);
538 candidate->security->prime_packet_security(candidate);
540 /* leave the candidate lurking in zombie mode attached to the
541 * bundle until we're ready for it */
542 rxrpc_put_connection(candidate);
546 /* we've got a connection with a free channel and we can now attach the
548 * - we're holding the transport's client lock
549 * - we're holding a reference on the connection
550 * - we're holding a reference on the bundle
552 for (chan = 0; chan < RXRPC_MAXCALLS; chan++)
553 if (!conn->channels[chan])
555 ASSERT(conn->channels[0] == NULL ||
556 conn->channels[1] == NULL ||
557 conn->channels[2] == NULL ||
558 conn->channels[3] == NULL);
562 conn->channels[chan] = call;
564 call->channel = chan;
565 call->cid = conn->proto.cid | chan;
566 call->call_id = ++conn->call_counter;
568 _net("CONNECT client on conn %d chan %d as call %x",
569 conn->debug_id, chan, call->call_id);
571 ASSERTCMP(conn->avail_calls, <, RXRPC_MAXCALLS);
572 spin_unlock(&trans->client_lock);
574 rxrpc_add_call_ID_to_conn(conn, call);
580 remove_wait_queue(&bundle->chanwait, &myself);
581 __set_current_state(TASK_RUNNING);
583 _leave(" = -ERESTARTSYS");
588 * get a record of an incoming connection
590 struct rxrpc_connection *
591 rxrpc_incoming_connection(struct rxrpc_transport *trans, struct sk_buff *skb)
593 struct rxrpc_connection *conn, *candidate = NULL;
594 struct rxrpc_skb_priv *sp = rxrpc_skb(skb);
595 struct rb_node *p, **pp;
596 const char *new = "old";
602 ASSERT(sp->hdr.flags & RXRPC_CLIENT_INITIATED);
604 epoch = sp->hdr.epoch;
605 cid = sp->hdr.cid & RXRPC_CIDMASK;
607 /* search the connection list first */
608 read_lock_bh(&trans->conn_lock);
610 p = trans->server_conns.rb_node;
612 conn = rb_entry(p, struct rxrpc_connection, node);
614 _debug("maybe %x", conn->proto.cid);
616 if (epoch < conn->proto.epoch)
618 else if (epoch > conn->proto.epoch)
620 else if (cid < conn->proto.cid)
622 else if (cid > conn->proto.cid)
625 goto found_extant_connection;
627 read_unlock_bh(&trans->conn_lock);
629 /* not yet present - create a candidate for a new record and then
631 candidate = rxrpc_alloc_connection(GFP_NOIO);
633 _leave(" = -ENOMEM");
634 return ERR_PTR(-ENOMEM);
637 candidate->trans = trans;
638 candidate->proto.local = trans->local;
639 candidate->proto.epoch = sp->hdr.epoch;
640 candidate->proto.cid = sp->hdr.cid & RXRPC_CIDMASK;
641 candidate->proto.in_clientflag = RXRPC_CLIENT_INITIATED;
642 candidate->params.local = trans->local;
643 candidate->params.peer = trans->peer;
644 candidate->params.service_id = sp->hdr.serviceId;
645 candidate->security_ix = sp->hdr.securityIndex;
646 candidate->out_clientflag = 0;
647 candidate->state = RXRPC_CONN_SERVER;
648 if (candidate->params.service_id)
649 candidate->state = RXRPC_CONN_SERVER_UNSECURED;
651 write_lock_bh(&trans->conn_lock);
653 pp = &trans->server_conns.rb_node;
657 conn = rb_entry(p, struct rxrpc_connection, node);
659 if (epoch < conn->proto.epoch)
660 pp = &(*pp)->rb_left;
661 else if (epoch > conn->proto.epoch)
662 pp = &(*pp)->rb_right;
663 else if (cid < conn->proto.cid)
664 pp = &(*pp)->rb_left;
665 else if (cid > conn->proto.cid)
666 pp = &(*pp)->rb_right;
668 goto found_extant_second;
671 /* we can now add the new candidate to the list */
674 rb_link_node(&conn->node, p, pp);
675 rb_insert_color(&conn->node, &trans->server_conns);
676 atomic_inc(&conn->trans->usage);
678 write_unlock_bh(&trans->conn_lock);
680 write_lock_bh(&rxrpc_connection_lock);
681 list_add_tail(&conn->link, &rxrpc_connections);
682 write_unlock_bh(&rxrpc_connection_lock);
687 _net("CONNECTION %s %d {%x}", new, conn->debug_id, conn->proto.cid);
689 _leave(" = %p {u=%d}", conn, atomic_read(&conn->usage));
692 /* we found the connection in the list immediately */
693 found_extant_connection:
694 if (sp->hdr.securityIndex != conn->security_ix) {
695 read_unlock_bh(&trans->conn_lock);
696 goto security_mismatch;
698 atomic_inc(&conn->usage);
699 read_unlock_bh(&trans->conn_lock);
702 /* we found the connection on the second time through the list */
704 if (sp->hdr.securityIndex != conn->security_ix) {
705 write_unlock_bh(&trans->conn_lock);
706 goto security_mismatch;
708 atomic_inc(&conn->usage);
709 write_unlock_bh(&trans->conn_lock);
715 _leave(" = -EKEYREJECTED");
716 return ERR_PTR(-EKEYREJECTED);
720 * find a connection based on transport and RxRPC connection ID for an incoming
723 struct rxrpc_connection *rxrpc_find_connection(struct rxrpc_transport *trans,
726 struct rxrpc_connection *conn;
727 struct rxrpc_skb_priv *sp = rxrpc_skb(skb);
731 _enter(",{%x,%x}", sp->hdr.cid, sp->hdr.flags);
733 read_lock_bh(&trans->conn_lock);
735 cid = sp->hdr.cid & RXRPC_CIDMASK;
736 epoch = sp->hdr.epoch;
738 if (sp->hdr.flags & RXRPC_CLIENT_INITIATED)
739 p = trans->server_conns.rb_node;
741 p = trans->client_conns.rb_node;
744 conn = rb_entry(p, struct rxrpc_connection, node);
746 _debug("maybe %x", conn->proto.cid);
748 if (epoch < conn->proto.epoch)
750 else if (epoch > conn->proto.epoch)
752 else if (cid < conn->proto.cid)
754 else if (cid > conn->proto.cid)
760 read_unlock_bh(&trans->conn_lock);
765 atomic_inc(&conn->usage);
766 read_unlock_bh(&trans->conn_lock);
767 _leave(" = %p", conn);
772 * release a virtual connection
774 void rxrpc_put_connection(struct rxrpc_connection *conn)
776 _enter("%p{u=%d,d=%d}",
777 conn, atomic_read(&conn->usage), conn->debug_id);
779 ASSERTCMP(atomic_read(&conn->usage), >, 0);
781 conn->put_time = ktime_get_seconds();
782 if (atomic_dec_and_test(&conn->usage)) {
784 rxrpc_queue_delayed_work(&rxrpc_connection_reap, 0);
791 * destroy a virtual connection
793 static void rxrpc_destroy_connection(struct rxrpc_connection *conn)
795 _enter("%p{%d}", conn, atomic_read(&conn->usage));
797 ASSERTCMP(atomic_read(&conn->usage), ==, 0);
799 _net("DESTROY CONN %d", conn->debug_id);
802 rxrpc_put_bundle(conn->trans, conn->bundle);
804 ASSERT(RB_EMPTY_ROOT(&conn->calls));
805 rxrpc_purge_queue(&conn->rx_queue);
807 conn->security->clear(conn);
808 key_put(conn->params.key);
809 key_put(conn->server_key);
811 rxrpc_put_transport(conn->trans);
817 * reap dead connections
819 static void rxrpc_connection_reaper(struct work_struct *work)
821 struct rxrpc_connection *conn, *_p;
822 unsigned long now, earliest, reap_time;
824 LIST_HEAD(graveyard);
828 now = ktime_get_seconds();
829 earliest = ULONG_MAX;
831 write_lock_bh(&rxrpc_connection_lock);
832 list_for_each_entry_safe(conn, _p, &rxrpc_connections, link) {
833 _debug("reap CONN %d { u=%d,t=%ld }",
834 conn->debug_id, atomic_read(&conn->usage),
835 (long) now - (long) conn->put_time);
837 if (likely(atomic_read(&conn->usage) > 0))
840 spin_lock(&conn->trans->client_lock);
841 write_lock(&conn->trans->conn_lock);
842 reap_time = conn->put_time + rxrpc_connection_expiry;
844 if (atomic_read(&conn->usage) > 0) {
846 } else if (reap_time <= now) {
847 list_move_tail(&conn->link, &graveyard);
848 if (conn->out_clientflag)
849 rb_erase(&conn->node,
850 &conn->trans->client_conns);
852 rb_erase(&conn->node,
853 &conn->trans->server_conns);
855 list_del_init(&conn->bundle_link);
856 conn->bundle->num_conns--;
859 } else if (reap_time < earliest) {
860 earliest = reap_time;
863 write_unlock(&conn->trans->conn_lock);
864 spin_unlock(&conn->trans->client_lock);
866 write_unlock_bh(&rxrpc_connection_lock);
868 if (earliest != ULONG_MAX) {
869 _debug("reschedule reaper %ld", (long) earliest - now);
870 ASSERTCMP(earliest, >, now);
871 rxrpc_queue_delayed_work(&rxrpc_connection_reap,
872 (earliest - now) * HZ);
875 /* then destroy all those pulled out */
876 while (!list_empty(&graveyard)) {
877 conn = list_entry(graveyard.next, struct rxrpc_connection,
879 list_del_init(&conn->link);
881 ASSERTCMP(atomic_read(&conn->usage), ==, 0);
882 rxrpc_destroy_connection(conn);
889 * preemptively destroy all the connection records rather than waiting for them
892 void __exit rxrpc_destroy_all_connections(void)
896 rxrpc_connection_expiry = 0;
897 cancel_delayed_work(&rxrpc_connection_reap);
898 rxrpc_queue_delayed_work(&rxrpc_connection_reap, 0);