FEX-Emu/linux
1
0
Fork 0
mirror of https://github.com/FEX-Emu/linux.git synced 2024-12-27 03:47:43 +00:00
Code Issues Actions Packages Projects Releases Wiki Activity
a1399f8bb0
linux/net/rxrpc/conn_object.c
David Howells a1399f8bb0 rxrpc: Call channels should have separate call number spaces 
Each channel on a connection has a separate, independent number space from
which to allocate callNumber values.  It is entirely possible, for example,
to have a connection with four active calls, each with call number 1.

Note that the callNumber values for any particular channel don't have to
start at 1, but they are supposed to increment monotonically for that
channel from a client's perspective and may not be reused once the call
number is transmitted (until the epoch cycles all the way back round).

Currently, however, call numbers are allocated on a per-connection basis
and, further, are held in an rb-tree.  The rb-tree is redundant as the four
channel pointers in the rxrpc_connection struct are entirely capable of
pointing to all the calls currently in progress on a connection.

To this end, make the following changes:

 (1) Handle call number allocation independently per channel.

 (2) Get rid of the conn->calls rb-tree.  This is overkill as a connection
     may have a maximum of four calls in progress at any one time.  Use the
     pointers in the channels[] array instead, indexed by the channel
     number from the packet.

 (3) For each channel, save the result of the last call that was in
     progress on that channel in conn->channels[] so that the final ACK or
     ABORT packet can be replayed if necessary.  Any call earlier than that
     is just ignored.  If we've seen the next call number in a packet, the
     last one is most definitely defunct.

 (4) When generating a RESPONSE packet for a connection, the call number
     counter for each channel must be included in it.

 (5) When parsing a RESPONSE packet for a connection, the call number
     counters contained therein should be used to set the minimum expected
     call numbers on each channel.

To do in future commits:

 (1) Replay terminal packets based on the last call stored in
     conn->channels[].

 (2) Connections should be retired before the callNumber space on any
     channel runs out.

 (3) A server is expected to disregard or reject any new incoming call that
     has a call number less than the current call number counter.  The call
     number counter for that channel must be advanced to the new call
     number.

     Note that the server cannot just require that the next call that it
     sees on a channel be exactly the call number counter + 1 because then
     there's a scenario that could cause a problem: The client transmits a
     packet to initiate a connection, the network goes out, the server
     sends an ACK (which gets lost), the client sends an ABORT (which also
     gets lost); the network then reconnects, the client then reuses the
     call number for the next call (it doesn't know the server already saw
     the call number), but the server thinks it already has the first
     packet of this call (it doesn't know that the client doesn't know that
     it saw the call number the first time).

Signed-off-by: David Howells <dhowells@redhat.com>
2016-07-06 10:43:52 +01:00

706 lines
17 KiB
C
Raw Blame History

/* RxRPC virtual connection handler
*
* Copyright (C) 2007 Red Hat, Inc. All Rights Reserved.
* Written by David Howells (dhowells@redhat.com)
*
* This program is free software; you can redistribute it and/or
* modify it under the terms of the GNU General Public License
* as published by the Free Software Foundation; either version
* 2 of the License, or (at your option) any later version.
*/
#define pr_fmt(fmt) KBUILD_MODNAME ": " fmt
#include <linux/module.h>
#include <linux/slab.h>
#include <linux/net.h>
#include <linux/skbuff.h>
#include <linux/crypto.h>
#include <net/sock.h>
#include <net/af_rxrpc.h>
#include "ar-internal.h"
/*
* Time till a connection expires after last use (in seconds).
*/
unsigned int rxrpc_connection_expiry = 10 * 60;
static void rxrpc_connection_reaper(struct work_struct *work);
LIST_HEAD(rxrpc_connections);
DEFINE_RWLOCK(rxrpc_connection_lock);
static DECLARE_DELAYED_WORK(rxrpc_connection_reap, rxrpc_connection_reaper);
/*
* allocate a new connection
*/
static struct rxrpc_connection *rxrpc_alloc_connection(gfp_t gfp)
{
struct rxrpc_connection *conn;
_enter("");
conn = kzalloc(sizeof(struct rxrpc_connection), gfp);
if (conn) {
spin_lock_init(&conn->channel_lock);
init_waitqueue_head(&conn->channel_wq);
INIT_WORK(&conn->processor, &rxrpc_process_connection);
INIT_LIST_HEAD(&conn->link);
skb_queue_head_init(&conn->rx_queue);
conn->security = &rxrpc_no_security;
spin_lock_init(&conn->state_lock);
atomic_set(&conn->usage, 1);
conn->debug_id = atomic_inc_return(&rxrpc_debug_id);
atomic_set(&conn->avail_chans, RXRPC_MAXCALLS);
conn->size_align = 4;
conn->header_size = sizeof(struct rxrpc_wire_header);
}
_leave(" = %p{%d}", conn, conn ? conn->debug_id : 0);
return conn;
}
/*
* Allocate a client connection. The caller must take care to clear any
* padding bytes in *cp.
*/
static struct rxrpc_connection *
rxrpc_alloc_client_connection(struct rxrpc_conn_parameters *cp, gfp_t gfp)
{
struct rxrpc_connection *conn;
int ret;
_enter("");
conn = rxrpc_alloc_connection(gfp);
if (!conn) {
_leave(" = -ENOMEM");
return ERR_PTR(-ENOMEM);
}
conn->params = *cp;
conn->proto.local = cp->local;
conn->proto.epoch = rxrpc_epoch;
conn->proto.cid = 0;
conn->proto.in_clientflag = 0;
conn->proto.family = cp->peer->srx.transport.family;
conn->out_clientflag = RXRPC_CLIENT_INITIATED;
conn->state = RXRPC_CONN_CLIENT;
switch (conn->proto.family) {
case AF_INET:
conn->proto.addr_size = sizeof(conn->proto.ipv4_addr);
conn->proto.ipv4_addr = cp->peer->srx.transport.sin.sin_addr;
conn->proto.port = cp->peer->srx.transport.sin.sin_port;
break;
}
ret = rxrpc_get_client_connection_id(conn, gfp);
if (ret < 0)
goto error_0;
ret = rxrpc_init_client_conn_security(conn);
if (ret < 0)
goto error_1;
ret = conn->security->prime_packet_security(conn);
if (ret < 0)
goto error_2;
write_lock(&rxrpc_connection_lock);
list_add_tail(&conn->link, &rxrpc_connections);
write_unlock(&rxrpc_connection_lock);
/* We steal the caller's peer ref. */
cp->peer = NULL;
rxrpc_get_local(conn->params.local);
key_get(conn->params.key);
_leave(" = %p", conn);
return conn;
error_2:
conn->security->clear(conn);
error_1:
rxrpc_put_client_connection_id(conn);
error_0:
kfree(conn);
_leave(" = %d", ret);
return ERR_PTR(ret);
}
/*
* find a connection for a call
* - called in process context with IRQs enabled
*/
int rxrpc_connect_call(struct rxrpc_call *call,
struct rxrpc_conn_parameters *cp,
struct sockaddr_rxrpc *srx,
gfp_t gfp)
{
struct rxrpc_connection *conn, *candidate = NULL;
struct rxrpc_local *local = cp->local;
struct rb_node *p, **pp, *parent;
long diff;
int chan;
DECLARE_WAITQUEUE(myself, current);
_enter("{%d,%lx},", call->debug_id, call->user_call_ID);
cp->peer = rxrpc_lookup_peer(cp->local, srx, gfp);
if (!cp->peer)
return -ENOMEM;
if (!cp->exclusive) {
/* Search for a existing client connection unless this is going
* to be a connection that's used exclusively for a single call.
*/
_debug("search 1");
spin_lock(&local->client_conns_lock);
p = local->client_conns.rb_node;
while (p) {
conn = rb_entry(p, struct rxrpc_connection, client_node);
#define cmp(X) ((long)conn->params.X - (long)cp->X)
diff = (cmp(peer) ?:
cmp(key) ?:
cmp(security_level));
if (diff < 0)
p = p->rb_left;
else if (diff > 0)
p = p->rb_right;
else
goto found_extant_conn;
}
spin_unlock(&local->client_conns_lock);
}
/* We didn't find a connection or we want an exclusive one. */
_debug("get new conn");
candidate = rxrpc_alloc_client_connection(cp, gfp);
if (!candidate) {
_leave(" = -ENOMEM");
return -ENOMEM;
}
if (cp->exclusive) {
/* Assign the call on an exclusive connection to channel 0 and
* don't add the connection to the endpoint's shareable conn
* lookup tree.
*/
_debug("exclusive chan 0");
conn = candidate;
atomic_set(&conn->avail_chans, RXRPC_MAXCALLS - 1);
spin_lock(&conn->channel_lock);
chan = 0;
goto found_channel;
}
/* We need to redo the search before attempting to add a new connection
* lest we race with someone else adding a conflicting instance.
*/
_debug("search 2");
spin_lock(&local->client_conns_lock);
pp = &local->client_conns.rb_node;
parent = NULL;
while (*pp) {
parent = *pp;
conn = rb_entry(parent, struct rxrpc_connection, client_node);
diff = (cmp(peer) ?:
cmp(key) ?:
cmp(security_level));
if (diff < 0)
pp = &(*pp)->rb_left;
else if (diff > 0)
pp = &(*pp)->rb_right;
else
goto found_extant_conn;
}
/* The second search also failed; simply add the new connection with
* the new call in channel 0. Note that we need to take the channel
* lock before dropping the client conn lock.
*/
_debug("new conn");
conn = candidate;
candidate = NULL;
rb_link_node(&conn->client_node, parent, pp);
rb_insert_color(&conn->client_node, &local->client_conns);
atomic_set(&conn->avail_chans, RXRPC_MAXCALLS - 1);
spin_lock(&conn->channel_lock);
spin_unlock(&local->client_conns_lock);
chan = 0;
found_channel:
_debug("found chan");
call->conn = conn;
call->channel = chan;
call->epoch = conn->proto.epoch;
call->cid = conn->proto.cid | chan;
call->call_id = ++conn->channels[chan].call_counter;
conn->channels[chan].call_id = call->call_id;
rcu_assign_pointer(conn->channels[chan].call, call);
_net("CONNECT call %d on conn %d", call->debug_id, conn->debug_id);
spin_unlock(&conn->channel_lock);
rxrpc_put_peer(cp->peer);
cp->peer = NULL;
_leave(" = %p {u=%d}", conn, atomic_read(&conn->usage));
return 0;
/* We found a suitable connection already in existence. Discard any
* candidate we may have allocated, and try to get a channel on this
* one.
*/
found_extant_conn:
_debug("found conn");
rxrpc_get_connection(conn);
spin_unlock(&local->client_conns_lock);
rxrpc_put_connection(candidate);
if (!atomic_add_unless(&conn->avail_chans, -1, 0)) {
if (!gfpflags_allow_blocking(gfp)) {
rxrpc_put_connection(conn);
_leave(" = -EAGAIN");
return -EAGAIN;
}
add_wait_queue(&conn->channel_wq, &myself);
for (;;) {
set_current_state(TASK_INTERRUPTIBLE);
if (atomic_add_unless(&conn->avail_chans, -1, 0))
break;
if (signal_pending(current))
goto interrupted;
schedule();
}
remove_wait_queue(&conn->channel_wq, &myself);
__set_current_state(TASK_RUNNING);
}
/* The connection allegedly now has a free channel and we can now
* attach the call to it.
*/
spin_lock(&conn->channel_lock);
for (chan = 0; chan < RXRPC_MAXCALLS; chan++)
if (!conn->channels[chan].call)
goto found_channel;
BUG();
interrupted:
remove_wait_queue(&conn->channel_wq, &myself);
__set_current_state(TASK_RUNNING);
rxrpc_put_connection(conn);
rxrpc_put_peer(cp->peer);
cp->peer = NULL;
_leave(" = -ERESTARTSYS");
return -ERESTARTSYS;
}
/*
* get a record of an incoming connection
*/
struct rxrpc_connection *rxrpc_incoming_connection(struct rxrpc_local *local,
struct rxrpc_peer *peer,
struct sk_buff *skb)
{
struct rxrpc_connection *conn, *candidate = NULL;
struct rxrpc_skb_priv *sp = rxrpc_skb(skb);
struct rb_node *p, **pp;
const char *new = "old";
u32 epoch, cid;
_enter("");
ASSERT(sp->hdr.flags & RXRPC_CLIENT_INITIATED);
epoch = sp->hdr.epoch;
cid = sp->hdr.cid & RXRPC_CIDMASK;
/* search the connection list first */
read_lock_bh(&peer->conn_lock);
p = peer->service_conns.rb_node;
while (p) {
conn = rb_entry(p, struct rxrpc_connection, service_node);
_debug("maybe %x", conn->proto.cid);
if (epoch < conn->proto.epoch)
p = p->rb_left;
else if (epoch > conn->proto.epoch)
p = p->rb_right;
else if (cid < conn->proto.cid)
p = p->rb_left;
else if (cid > conn->proto.cid)
p = p->rb_right;
else
goto found_extant_connection;
}
read_unlock_bh(&peer->conn_lock);
/* not yet present - create a candidate for a new record and then
* redo the search */
candidate = rxrpc_alloc_connection(GFP_NOIO);
if (!candidate) {
_leave(" = -ENOMEM");
return ERR_PTR(-ENOMEM);
}
candidate->proto.local = local;
candidate->proto.epoch = sp->hdr.epoch;
candidate->proto.cid = sp->hdr.cid & RXRPC_CIDMASK;
candidate->proto.in_clientflag = RXRPC_CLIENT_INITIATED;
candidate->params.local = local;
candidate->params.peer = peer;
candidate->params.service_id = sp->hdr.serviceId;
candidate->security_ix = sp->hdr.securityIndex;
candidate->out_clientflag = 0;
candidate->state = RXRPC_CONN_SERVICE;
if (candidate->params.service_id)
candidate->state = RXRPC_CONN_SERVICE_UNSECURED;
write_lock_bh(&peer->conn_lock);
pp = &peer->service_conns.rb_node;
p = NULL;
while (*pp) {
p = *pp;
conn = rb_entry(p, struct rxrpc_connection, service_node);
if (epoch < conn->proto.epoch)
pp = &(*pp)->rb_left;
else if (epoch > conn->proto.epoch)
pp = &(*pp)->rb_right;
else if (cid < conn->proto.cid)
pp = &(*pp)->rb_left;
else if (cid > conn->proto.cid)
pp = &(*pp)->rb_right;
else
goto found_extant_second;
}
/* we can now add the new candidate to the list */
conn = candidate;
candidate = NULL;
rb_link_node(&conn->service_node, p, pp);
rb_insert_color(&conn->service_node, &peer->service_conns);
rxrpc_get_peer(peer);
rxrpc_get_local(local);
write_unlock_bh(&peer->conn_lock);
write_lock(&rxrpc_connection_lock);
list_add_tail(&conn->link, &rxrpc_connections);
write_unlock(&rxrpc_connection_lock);
new = "new";
success:
_net("CONNECTION %s %d {%x}", new, conn->debug_id, conn->proto.cid);
_leave(" = %p {u=%d}", conn, atomic_read(&conn->usage));
return conn;
/* we found the connection in the list immediately */
found_extant_connection:
if (sp->hdr.securityIndex != conn->security_ix) {
read_unlock_bh(&peer->conn_lock);
goto security_mismatch;
}
rxrpc_get_connection(conn);
read_unlock_bh(&peer->conn_lock);
goto success;
/* we found the connection on the second time through the list */
found_extant_second:
if (sp->hdr.securityIndex != conn->security_ix) {
write_unlock_bh(&peer->conn_lock);
goto security_mismatch;
}
rxrpc_get_connection(conn);
write_unlock_bh(&peer->conn_lock);
kfree(candidate);
goto success;
security_mismatch:
kfree(candidate);
_leave(" = -EKEYREJECTED");
return ERR_PTR(-EKEYREJECTED);
}
/*
* find a connection based on transport and RxRPC connection ID for an incoming
* packet
*/
struct rxrpc_connection *rxrpc_find_connection(struct rxrpc_local *local,
struct rxrpc_peer *peer,
struct sk_buff *skb)
{
struct rxrpc_connection *conn;
struct rxrpc_skb_priv *sp = rxrpc_skb(skb);
struct rb_node *p;
u32 epoch, cid;
_enter(",{%x,%x}", sp->hdr.cid, sp->hdr.flags);
read_lock_bh(&peer->conn_lock);
cid = sp->hdr.cid & RXRPC_CIDMASK;
epoch = sp->hdr.epoch;
if (sp->hdr.flags & RXRPC_CLIENT_INITIATED) {
p = peer->service_conns.rb_node;
while (p) {
conn = rb_entry(p, struct rxrpc_connection, service_node);
_debug("maybe %x", conn->proto.cid);
if (epoch < conn->proto.epoch)
p = p->rb_left;
else if (epoch > conn->proto.epoch)
p = p->rb_right;
else if (cid < conn->proto.cid)
p = p->rb_left;
else if (cid > conn->proto.cid)
p = p->rb_right;
else
goto found;
}
} else {
conn = idr_find(&rxrpc_client_conn_ids, cid >> RXRPC_CIDSHIFT);
if (conn &&
conn->proto.epoch == epoch &&
conn->params.peer == peer)
goto found;
}
read_unlock_bh(&peer->conn_lock);
_leave(" = NULL");
return NULL;
found:
rxrpc_get_connection(conn);
read_unlock_bh(&peer->conn_lock);
_leave(" = %p", conn);
return conn;
}
/*
* Disconnect a call and clear any channel it occupies when that call
* terminates. The caller must hold the channel_lock and must release the
* call's ref on the connection.
*/
void __rxrpc_disconnect_call(struct rxrpc_call *call)
{
struct rxrpc_connection *conn = call->conn;
struct rxrpc_channel *chan = &conn->channels[call->channel];
_enter("%d,%d", conn->debug_id, call->channel);
if (rcu_access_pointer(chan->call) == call) {
/* Save the result of the call so that we can repeat it if necessary
* through the channel, whilst disposing of the actual call record.
*/
chan->last_result = call->local_abort;
smp_wmb();
chan->last_call = chan->call_id;
chan->call_id = chan->call_counter;
rcu_assign_pointer(chan->call, NULL);
atomic_inc(&conn->avail_chans);
wake_up(&conn->channel_wq);
}
_leave("");
}
/*
* Disconnect a call and clear any channel it occupies when that call
* terminates.
*/
void rxrpc_disconnect_call(struct rxrpc_call *call)
{
struct rxrpc_connection *conn = call->conn;
spin_lock(&conn->channel_lock);
__rxrpc_disconnect_call(call);
spin_unlock(&conn->channel_lock);
call->conn = NULL;
rxrpc_put_connection(conn);
}
/*
* release a virtual connection
*/
void rxrpc_put_connection(struct rxrpc_connection *conn)
{
if (!conn)
return;
_enter("%p{u=%d,d=%d}",
conn, atomic_read(&conn->usage), conn->debug_id);
ASSERTCMP(atomic_read(&conn->usage), >, 0);
conn->put_time = ktime_get_seconds();
if (atomic_dec_and_test(&conn->usage)) {
_debug("zombie");
rxrpc_queue_delayed_work(&rxrpc_connection_reap, 0);
}
_leave("");
}
/*
* destroy a virtual connection
*/
static void rxrpc_destroy_connection(struct rcu_head *rcu)
{
struct rxrpc_connection *conn =
container_of(rcu, struct rxrpc_connection, rcu);
_enter("{%d,u=%d}", conn->debug_id, atomic_read(&conn->usage));
ASSERTCMP(atomic_read(&conn->usage), ==, 0);
_net("DESTROY CONN %d", conn->debug_id);
rxrpc_purge_queue(&conn->rx_queue);
conn->security->clear(conn);
key_put(conn->params.key);
key_put(conn->server_key);
rxrpc_put_peer(conn->params.peer);
rxrpc_put_local(conn->params.local);
kfree(conn);
_leave("");
}
/*
* reap dead connections
*/
static void rxrpc_connection_reaper(struct work_struct *work)
{
struct rxrpc_connection *conn, *_p;
struct rxrpc_peer *peer;
unsigned long now, earliest, reap_time;
LIST_HEAD(graveyard);
_enter("");
now = ktime_get_seconds();
earliest = ULONG_MAX;
write_lock(&rxrpc_connection_lock);
list_for_each_entry_safe(conn, _p, &rxrpc_connections, link) {
_debug("reap CONN %d { u=%d,t=%ld }",
conn->debug_id, atomic_read(&conn->usage),
(long) now - (long) conn->put_time);
if (likely(atomic_read(&conn->usage) > 0))
continue;
if (rxrpc_conn_is_client(conn)) {
struct rxrpc_local *local = conn->params.local;
spin_lock(&local->client_conns_lock);
reap_time = conn->put_time + rxrpc_connection_expiry;
if (atomic_read(&conn->usage) > 0) {
;
} else if (reap_time <= now) {
list_move_tail(&conn->link, &graveyard);
rxrpc_put_client_connection_id(conn);
rb_erase(&conn->client_node,
&local->client_conns);
} else if (reap_time < earliest) {
earliest = reap_time;
}
spin_unlock(&local->client_conns_lock);
} else {
peer = conn->params.peer;
write_lock_bh(&peer->conn_lock);
reap_time = conn->put_time + rxrpc_connection_expiry;
if (atomic_read(&conn->usage) > 0) {
;
} else if (reap_time <= now) {
list_move_tail(&conn->link, &graveyard);
rb_erase(&conn->service_node,
&peer->service_conns);
} else if (reap_time < earliest) {
earliest = reap_time;
}
write_unlock_bh(&peer->conn_lock);
}
}
write_unlock(&rxrpc_connection_lock);
if (earliest != ULONG_MAX) {
_debug("reschedule reaper %ld", (long) earliest - now);
ASSERTCMP(earliest, >, now);
rxrpc_queue_delayed_work(&rxrpc_connection_reap,
(earliest - now) * HZ);
}
/* then destroy all those pulled out */
while (!list_empty(&graveyard)) {
conn = list_entry(graveyard.next, struct rxrpc_connection,
link);
list_del_init(&conn->link);
ASSERTCMP(atomic_read(&conn->usage), ==, 0);
skb_queue_purge(&conn->rx_queue);
call_rcu(&conn->rcu, rxrpc_destroy_connection);
}
_leave("");
}
/*
* preemptively destroy all the connection records rather than waiting for them
* to time out
*/
void __exit rxrpc_destroy_all_connections(void)
{
struct rxrpc_connection *conn, *_p;
bool leak = false;
_enter("");
rxrpc_connection_expiry = 0;
cancel_delayed_work(&rxrpc_connection_reap);
rxrpc_queue_delayed_work(&rxrpc_connection_reap, 0);
flush_workqueue(rxrpc_workqueue);
write_lock(&rxrpc_connection_lock);
list_for_each_entry_safe(conn, _p, &rxrpc_connections, link) {
pr_err("AF_RXRPC: Leaked conn %p {%d}\n",
conn, atomic_read(&conn->usage));
leak = true;
}
write_unlock(&rxrpc_connection_lock);
BUG_ON(leak);
/* Make sure the local and peer records pinned by any dying connections
* are released.
*/
rcu_barrier();
rxrpc_destroy_client_conn_ids();
_leave("");
}
Reference in New Issue View Git Blame Copy Permalink