Commit 4f95dd78 authored by David Howells's avatar David Howells

rxrpc: Rework local endpoint management

Rework the local RxRPC endpoint management.

Local endpoint objects are maintained in a flat list as before.  This
should be okay as there shouldn't be more than one per open AF_RXRPC socket
(there can be fewer as local endpoints can be shared if their local service
ID is 0 and they share the same local transport parameters).

Changes:

 (1) Local endpoints may now only be shared if they have local service ID 0
     (ie. they're not being used for listening).

     This prevents a scenario where process A is listening of the Cache
     Manager port and process B contacts a fileserver - which may then
     attempt to send CM requests back to B.  But if A and B are sharing a
     local endpoint, A will get the CM requests meant for B.

 (2) We use a mutex to handle lookups and don't provide RCU-only lookups
     since we only expect to access the list when opening a socket or
     destroying an endpoint.

     The local endpoint object is pointed to by the transport socket's
     sk_user_data for the life of the transport socket - allowing us to
     refer to it directly from the sk_data_ready and sk_error_report
     callbacks.

 (3) atomic_inc_not_zero() now exists and can be used to only share a local
     endpoint if the last reference hasn't yet gone.

 (4) We can remove rxrpc_local_lock - a spinlock that had to be taken with
     BH processing disabled given that we assume sk_user_data won't change
     under us.

 (5) The transport socket is shut down before we clear the sk_user_data
     pointer so that we can be sure that the transport socket's callbacks
     won't be invoked once the RCU destruction is scheduled.

 (6) Local endpoints have a work item that handles both destruction and
     event processing.  The means that destruction doesn't then need to
     wait for event processing.  The event queues can then be cleared after
     the transport socket is shut down.

 (7) Local endpoints are no longer available for resurrection beyond the
     life of the sockets that had them open.  As soon as their last ref
     goes, they are scheduled for destruction and may not have their usage
     count moved from 0.
Signed-off-by: default avatarDavid Howells <dhowells@redhat.com>
parent 87563616
...@@ -102,6 +102,8 @@ static int rxrpc_validate_address(struct rxrpc_sock *rx, ...@@ -102,6 +102,8 @@ static int rxrpc_validate_address(struct rxrpc_sock *rx,
switch (srx->transport.family) { switch (srx->transport.family) {
case AF_INET: case AF_INET:
if (srx->transport_len < sizeof(struct sockaddr_in))
return -EINVAL;
_debug("INET: %x @ %pI4", _debug("INET: %x @ %pI4",
ntohs(srx->transport.sin.sin_port), ntohs(srx->transport.sin.sin_port),
&srx->transport.sin.sin_addr); &srx->transport.sin.sin_addr);
...@@ -835,12 +837,27 @@ static void __exit af_rxrpc_exit(void) ...@@ -835,12 +837,27 @@ static void __exit af_rxrpc_exit(void)
rxrpc_destroy_all_calls(); rxrpc_destroy_all_calls();
rxrpc_destroy_all_connections(); rxrpc_destroy_all_connections();
rxrpc_destroy_all_transports(); rxrpc_destroy_all_transports();
rxrpc_destroy_all_locals();
ASSERTCMP(atomic_read(&rxrpc_n_skbs), ==, 0); ASSERTCMP(atomic_read(&rxrpc_n_skbs), ==, 0);
/* We need to flush the scheduled work twice because the local endpoint
* records involve a work item in their destruction as they can only be
* destroyed from process context. However, a connection may have a
* work item outstanding - and this will pin the local endpoint record
* until the connection goes away.
*
* Peers don't pin locals and calls pin sockets - which prevents the
* module from being unloaded - so we should only need two flushes.
*/
_debug("flush scheduled work"); _debug("flush scheduled work");
flush_workqueue(rxrpc_workqueue); flush_workqueue(rxrpc_workqueue);
_debug("flush scheduled work 2");
flush_workqueue(rxrpc_workqueue);
_debug("synchronise RCU");
rcu_barrier();
_debug("destroy locals");
rxrpc_destroy_all_locals();
remove_proc_entry("rxrpc_conns", init_net.proc_net); remove_proc_entry("rxrpc_conns", init_net.proc_net);
remove_proc_entry("rxrpc_calls", init_net.proc_net); remove_proc_entry("rxrpc_calls", init_net.proc_net);
destroy_workqueue(rxrpc_workqueue); destroy_workqueue(rxrpc_workqueue);
......
...@@ -170,25 +170,26 @@ struct rxrpc_security { ...@@ -170,25 +170,26 @@ struct rxrpc_security {
}; };
/* /*
* RxRPC local transport endpoint definition * RxRPC local transport endpoint description
* - matched by local port, address and protocol type * - owned by a single AF_RXRPC socket
* - pointed to by transport socket struct sk_user_data
*/ */
struct rxrpc_local { struct rxrpc_local {
struct rcu_head rcu;
atomic_t usage;
struct list_head link;
struct socket *socket; /* my UDP socket */ struct socket *socket; /* my UDP socket */
struct work_struct destroyer; /* endpoint destroyer */ struct work_struct processor;
struct work_struct acceptor; /* incoming call processor */
struct work_struct rejecter; /* packet reject writer */
struct work_struct event_processor; /* endpoint event processor */
struct list_head services; /* services listening on this endpoint */ struct list_head services; /* services listening on this endpoint */
struct list_head link; /* link in endpoint list */
struct rw_semaphore defrag_sem; /* control re-enablement of IP DF bit */ struct rw_semaphore defrag_sem; /* control re-enablement of IP DF bit */
struct sk_buff_head accept_queue; /* incoming calls awaiting acceptance */ struct sk_buff_head accept_queue; /* incoming calls awaiting acceptance */
struct sk_buff_head reject_queue; /* packets awaiting rejection */ struct sk_buff_head reject_queue; /* packets awaiting rejection */
struct sk_buff_head event_queue; /* endpoint event packets awaiting processing */ struct sk_buff_head event_queue; /* endpoint event packets awaiting processing */
struct mutex conn_lock; /* Client connection creation lock */
spinlock_t lock; /* access lock */ spinlock_t lock; /* access lock */
rwlock_t services_lock; /* lock for services list */ rwlock_t services_lock; /* lock for services list */
atomic_t usage;
int debug_id; /* debug ID for printks */ int debug_id; /* debug ID for printks */
bool dead;
struct sockaddr_rxrpc srx; /* local address */ struct sockaddr_rxrpc srx; /* local address */
}; };
...@@ -487,7 +488,7 @@ extern struct rxrpc_transport *rxrpc_name_to_transport(struct rxrpc_sock *, ...@@ -487,7 +488,7 @@ extern struct rxrpc_transport *rxrpc_name_to_transport(struct rxrpc_sock *,
/* /*
* call_accept.c * call_accept.c
*/ */
void rxrpc_accept_incoming_calls(struct work_struct *); void rxrpc_accept_incoming_calls(struct rxrpc_local *);
struct rxrpc_call *rxrpc_accept_call(struct rxrpc_sock *, unsigned long); struct rxrpc_call *rxrpc_accept_call(struct rxrpc_sock *, unsigned long);
int rxrpc_reject_call(struct rxrpc_sock *); int rxrpc_reject_call(struct rxrpc_sock *);
...@@ -527,7 +528,7 @@ void __exit rxrpc_destroy_all_calls(void); ...@@ -527,7 +528,7 @@ void __exit rxrpc_destroy_all_calls(void);
*/ */
void rxrpc_process_connection(struct work_struct *); void rxrpc_process_connection(struct work_struct *);
void rxrpc_reject_packet(struct rxrpc_local *, struct sk_buff *); void rxrpc_reject_packet(struct rxrpc_local *, struct sk_buff *);
void rxrpc_reject_packets(struct work_struct *); void rxrpc_reject_packets(struct rxrpc_local *);
/* /*
* conn_object.c * conn_object.c
...@@ -575,17 +576,32 @@ int rxrpc_get_server_data_key(struct rxrpc_connection *, const void *, time_t, ...@@ -575,17 +576,32 @@ int rxrpc_get_server_data_key(struct rxrpc_connection *, const void *, time_t,
/* /*
* local_event.c * local_event.c
*/ */
extern void rxrpc_process_local_events(struct work_struct *); extern void rxrpc_process_local_events(struct rxrpc_local *);
/* /*
* local_object.c * local_object.c
*/ */
extern rwlock_t rxrpc_local_lock; struct rxrpc_local *rxrpc_lookup_local(const struct sockaddr_rxrpc *);
void __rxrpc_put_local(struct rxrpc_local *);
struct rxrpc_local *rxrpc_lookup_local(struct sockaddr_rxrpc *);
void rxrpc_put_local(struct rxrpc_local *);
void __exit rxrpc_destroy_all_locals(void); void __exit rxrpc_destroy_all_locals(void);
static inline void rxrpc_get_local(struct rxrpc_local *local)
{
atomic_inc(&local->usage);
}
static inline
struct rxrpc_local *rxrpc_get_local_maybe(struct rxrpc_local *local)
{
return atomic_inc_not_zero(&local->usage) ? local : NULL;
}
static inline void rxrpc_put_local(struct rxrpc_local *local)
{
if (atomic_dec_and_test(&local->usage))
__rxrpc_put_local(local);
}
/* /*
* misc.c * misc.c
*/ */
...@@ -874,15 +890,6 @@ static inline void rxrpc_purge_queue(struct sk_buff_head *list) ...@@ -874,15 +890,6 @@ static inline void rxrpc_purge_queue(struct sk_buff_head *list)
rxrpc_free_skb(skb); rxrpc_free_skb(skb);
} }
static inline void __rxrpc_get_local(struct rxrpc_local *local, const char *f)
{
CHECK_SLAB_OKAY(&local->usage);
if (atomic_inc_return(&local->usage) == 1)
printk("resurrected (%s)\n", f);
}
#define rxrpc_get_local(LOCAL) __rxrpc_get_local((LOCAL), __func__)
#define rxrpc_get_call(CALL) \ #define rxrpc_get_call(CALL) \
do { \ do { \
CHECK_SLAB_OKAY(&(CALL)->usage); \ CHECK_SLAB_OKAY(&(CALL)->usage); \
......
...@@ -202,10 +202,8 @@ static int rxrpc_accept_incoming_call(struct rxrpc_local *local, ...@@ -202,10 +202,8 @@ static int rxrpc_accept_incoming_call(struct rxrpc_local *local,
* accept incoming calls that need peer, transport and/or connection setting up * accept incoming calls that need peer, transport and/or connection setting up
* - the packets we get are all incoming client DATA packets that have seq == 1 * - the packets we get are all incoming client DATA packets that have seq == 1
*/ */
void rxrpc_accept_incoming_calls(struct work_struct *work) void rxrpc_accept_incoming_calls(struct rxrpc_local *local)
{ {
struct rxrpc_local *local =
container_of(work, struct rxrpc_local, acceptor);
struct rxrpc_skb_priv *sp; struct rxrpc_skb_priv *sp;
struct sockaddr_rxrpc srx; struct sockaddr_rxrpc srx;
struct rxrpc_sock *rx; struct rxrpc_sock *rx;
...@@ -215,21 +213,8 @@ void rxrpc_accept_incoming_calls(struct work_struct *work) ...@@ -215,21 +213,8 @@ void rxrpc_accept_incoming_calls(struct work_struct *work)
_enter("%d", local->debug_id); _enter("%d", local->debug_id);
read_lock_bh(&rxrpc_local_lock);
if (atomic_read(&local->usage) > 0)
rxrpc_get_local(local);
else
local = NULL;
read_unlock_bh(&rxrpc_local_lock);
if (!local) {
_leave(" [local dead]");
return;
}
process_next_packet:
skb = skb_dequeue(&local->accept_queue); skb = skb_dequeue(&local->accept_queue);
if (!skb) { if (!skb) {
rxrpc_put_local(local);
_leave("\n"); _leave("\n");
return; return;
} }
...@@ -292,7 +277,7 @@ void rxrpc_accept_incoming_calls(struct work_struct *work) ...@@ -292,7 +277,7 @@ void rxrpc_accept_incoming_calls(struct work_struct *work)
case -ECONNRESET: /* old calls are ignored */ case -ECONNRESET: /* old calls are ignored */
case -ECONNABORTED: /* aborted calls are reaborted or ignored */ case -ECONNABORTED: /* aborted calls are reaborted or ignored */
case 0: case 0:
goto process_next_packet; return;
case -ECONNREFUSED: case -ECONNREFUSED:
goto invalid_service; goto invalid_service;
case -EBUSY: case -EBUSY:
...@@ -308,18 +293,18 @@ void rxrpc_accept_incoming_calls(struct work_struct *work) ...@@ -308,18 +293,18 @@ void rxrpc_accept_incoming_calls(struct work_struct *work)
busy: busy:
rxrpc_busy(local, &srx, &whdr); rxrpc_busy(local, &srx, &whdr);
rxrpc_free_skb(skb); rxrpc_free_skb(skb);
goto process_next_packet; return;
invalid_service: invalid_service:
skb->priority = RX_INVALID_OPERATION; skb->priority = RX_INVALID_OPERATION;
rxrpc_reject_packet(local, skb); rxrpc_reject_packet(local, skb);
goto process_next_packet; return;
/* can't change connection security type mid-flow */ /* can't change connection security type mid-flow */
security_mismatch: security_mismatch:
skb->priority = RX_PROTOCOL_ERROR; skb->priority = RX_PROTOCOL_ERROR;
rxrpc_reject_packet(local, skb); rxrpc_reject_packet(local, skb);
goto process_next_packet; return;
} }
/* /*
......
...@@ -314,19 +314,14 @@ void rxrpc_reject_packet(struct rxrpc_local *local, struct sk_buff *skb) ...@@ -314,19 +314,14 @@ void rxrpc_reject_packet(struct rxrpc_local *local, struct sk_buff *skb)
{ {
CHECK_SLAB_OKAY(&local->usage); CHECK_SLAB_OKAY(&local->usage);
if (!atomic_inc_not_zero(&local->usage)) {
printk("resurrected on reject\n");
BUG();
}
skb_queue_tail(&local->reject_queue, skb); skb_queue_tail(&local->reject_queue, skb);
rxrpc_queue_work(&local->rejecter); rxrpc_queue_work(&local->processor);
} }
/* /*
* reject packets through the local endpoint * reject packets through the local endpoint
*/ */
void rxrpc_reject_packets(struct work_struct *work) void rxrpc_reject_packets(struct rxrpc_local *local)
{ {
union { union {
struct sockaddr sa; struct sockaddr sa;
...@@ -334,16 +329,12 @@ void rxrpc_reject_packets(struct work_struct *work) ...@@ -334,16 +329,12 @@ void rxrpc_reject_packets(struct work_struct *work)
} sa; } sa;
struct rxrpc_skb_priv *sp; struct rxrpc_skb_priv *sp;
struct rxrpc_wire_header whdr; struct rxrpc_wire_header whdr;
struct rxrpc_local *local;
struct sk_buff *skb; struct sk_buff *skb;
struct msghdr msg; struct msghdr msg;
struct kvec iov[2]; struct kvec iov[2];
size_t size; size_t size;
__be32 code; __be32 code;
local = container_of(work, struct rxrpc_local, rejecter);
rxrpc_get_local(local);
_enter("%d", local->debug_id); _enter("%d", local->debug_id);
iov[0].iov_base = &whdr; iov[0].iov_base = &whdr;
...@@ -395,9 +386,7 @@ void rxrpc_reject_packets(struct work_struct *work) ...@@ -395,9 +386,7 @@ void rxrpc_reject_packets(struct work_struct *work)
} }
rxrpc_free_skb(skb); rxrpc_free_skb(skb);
rxrpc_put_local(local);
} }
rxrpc_put_local(local);
_leave(""); _leave("");
} }
...@@ -594,9 +594,8 @@ static void rxrpc_post_packet_to_local(struct rxrpc_local *local, ...@@ -594,9 +594,8 @@ static void rxrpc_post_packet_to_local(struct rxrpc_local *local,
{ {
_enter("%p,%p", local, skb); _enter("%p,%p", local, skb);
atomic_inc(&local->usage);
skb_queue_tail(&local->event_queue, skb); skb_queue_tail(&local->event_queue, skb);
rxrpc_queue_work(&local->event_processor); rxrpc_queue_work(&local->processor);
} }
/* /*
...@@ -664,11 +663,15 @@ static struct rxrpc_connection *rxrpc_conn_from_local(struct rxrpc_local *local, ...@@ -664,11 +663,15 @@ static struct rxrpc_connection *rxrpc_conn_from_local(struct rxrpc_local *local,
/* /*
* handle data received on the local endpoint * handle data received on the local endpoint
* - may be called in interrupt context * - may be called in interrupt context
*
* The socket is locked by the caller and this prevents the socket from being
* shut down and the local endpoint from going away, thus sk_user_data will not
* be cleared until this function returns.
*/ */
void rxrpc_data_ready(struct sock *sk) void rxrpc_data_ready(struct sock *sk)
{ {
struct rxrpc_skb_priv *sp; struct rxrpc_skb_priv *sp;
struct rxrpc_local *local; struct rxrpc_local *local = sk->sk_user_data;
struct sk_buff *skb; struct sk_buff *skb;
int ret; int ret;
...@@ -676,21 +679,8 @@ void rxrpc_data_ready(struct sock *sk) ...@@ -676,21 +679,8 @@ void rxrpc_data_ready(struct sock *sk)
ASSERT(!irqs_disabled()); ASSERT(!irqs_disabled());
read_lock_bh(&rxrpc_local_lock);
local = sk->sk_user_data;
if (local && atomic_read(&local->usage) > 0)
rxrpc_get_local(local);
else
local = NULL;
read_unlock_bh(&rxrpc_local_lock);
if (!local) {
_leave(" [local dead]");
return;
}
skb = skb_recv_datagram(sk, 0, 1, &ret); skb = skb_recv_datagram(sk, 0, 1, &ret);
if (!skb) { if (!skb) {
rxrpc_put_local(local);
if (ret == -EAGAIN) if (ret == -EAGAIN)
return; return;
_debug("UDP socket error %d", ret); _debug("UDP socket error %d", ret);
...@@ -704,7 +694,6 @@ void rxrpc_data_ready(struct sock *sk) ...@@ -704,7 +694,6 @@ void rxrpc_data_ready(struct sock *sk)
/* we'll probably need to checksum it (didn't call sock_recvmsg) */ /* we'll probably need to checksum it (didn't call sock_recvmsg) */
if (skb_checksum_complete(skb)) { if (skb_checksum_complete(skb)) {
rxrpc_free_skb(skb); rxrpc_free_skb(skb);
rxrpc_put_local(local);
__UDP_INC_STATS(&init_net, UDP_MIB_INERRORS, 0); __UDP_INC_STATS(&init_net, UDP_MIB_INERRORS, 0);
_leave(" [CSUM failed]"); _leave(" [CSUM failed]");
return; return;
...@@ -769,7 +758,6 @@ void rxrpc_data_ready(struct sock *sk) ...@@ -769,7 +758,6 @@ void rxrpc_data_ready(struct sock *sk)
} }
out: out:
rxrpc_put_local(local);
return; return;
cant_route_call: cant_route_call:
...@@ -779,8 +767,7 @@ void rxrpc_data_ready(struct sock *sk) ...@@ -779,8 +767,7 @@ void rxrpc_data_ready(struct sock *sk)
if (sp->hdr.seq == 1) { if (sp->hdr.seq == 1) {
_debug("first packet"); _debug("first packet");
skb_queue_tail(&local->accept_queue, skb); skb_queue_tail(&local->accept_queue, skb);
rxrpc_queue_work(&local->acceptor); rxrpc_queue_work(&local->processor);
rxrpc_put_local(local);
_leave(" [incoming]"); _leave(" [incoming]");
return; return;
} }
...@@ -793,13 +780,11 @@ void rxrpc_data_ready(struct sock *sk) ...@@ -793,13 +780,11 @@ void rxrpc_data_ready(struct sock *sk)
_debug("reject type %d",sp->hdr.type); _debug("reject type %d",sp->hdr.type);
rxrpc_reject_packet(local, skb); rxrpc_reject_packet(local, skb);
} }
rxrpc_put_local(local);
_leave(" [no call]"); _leave(" [no call]");
return; return;
bad_message: bad_message:
skb->priority = RX_PROTOCOL_ERROR; skb->priority = RX_PROTOCOL_ERROR;
rxrpc_reject_packet(local, skb); rxrpc_reject_packet(local, skb);
rxrpc_put_local(local);
_leave(" [badmsg]"); _leave(" [badmsg]");
} }
...@@ -82,17 +82,15 @@ static void rxrpc_send_version_request(struct rxrpc_local *local, ...@@ -82,17 +82,15 @@ static void rxrpc_send_version_request(struct rxrpc_local *local,
/* /*
* Process event packets targetted at a local endpoint. * Process event packets targetted at a local endpoint.
*/ */
void rxrpc_process_local_events(struct work_struct *work) void rxrpc_process_local_events(struct rxrpc_local *local)
{ {
struct rxrpc_local *local = container_of(work, struct rxrpc_local, event_processor);
struct sk_buff *skb; struct sk_buff *skb;
char v; char v;
_enter(""); _enter("");
atomic_inc(&local->usage); skb = skb_dequeue(&local->event_queue);
if (skb) {
while ((skb = skb_dequeue(&local->event_queue))) {
struct rxrpc_skb_priv *sp = rxrpc_skb(skb); struct rxrpc_skb_priv *sp = rxrpc_skb(skb);
_debug("{%d},{%u}", local->debug_id, sp->hdr.type); _debug("{%d},{%u}", local->debug_id, sp->hdr.type);
...@@ -111,10 +109,8 @@ void rxrpc_process_local_events(struct work_struct *work) ...@@ -111,10 +109,8 @@ void rxrpc_process_local_events(struct work_struct *work)
break; break;
} }
rxrpc_put_local(local);
rxrpc_free_skb(skb); rxrpc_free_skb(skb);
} }
rxrpc_put_local(local);
_leave(""); _leave("");
} }
This diff is collapsed.
Markdown is supported
0%
or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment