Commit 04d7f1fb authored by David S. Miller's avatar David S. Miller

Merge branch 'tipc-topology-fixes'

Parthasarathy Bhuvaragan says:

====================
tipc: topology server fixes for nametable soft lockup

In this series, we revert the commit 333f7962 ("tipc: fix a
race condition leading to subscriber refcnt bug") and provide an
alternate solution to fix the race conditions in commits 2-4.

We have to do this as the above commit introduced a nametbl soft
lockup at module exit as described by patch#4.
====================
Signed-off-by: default avatarDavid S. Miller <davem@davemloft.net>
parents 294628c1 35e22e49
...@@ -263,6 +263,11 @@ static void tipc_node_write_lock(struct tipc_node *n) ...@@ -263,6 +263,11 @@ static void tipc_node_write_lock(struct tipc_node *n)
write_lock_bh(&n->lock); write_lock_bh(&n->lock);
} }
static void tipc_node_write_unlock_fast(struct tipc_node *n)
{
write_unlock_bh(&n->lock);
}
static void tipc_node_write_unlock(struct tipc_node *n) static void tipc_node_write_unlock(struct tipc_node *n)
{ {
struct net *net = n->net; struct net *net = n->net;
...@@ -417,7 +422,7 @@ void tipc_node_subscribe(struct net *net, struct list_head *subscr, u32 addr) ...@@ -417,7 +422,7 @@ void tipc_node_subscribe(struct net *net, struct list_head *subscr, u32 addr)
} }
tipc_node_write_lock(n); tipc_node_write_lock(n);
list_add_tail(subscr, &n->publ_list); list_add_tail(subscr, &n->publ_list);
tipc_node_write_unlock(n); tipc_node_write_unlock_fast(n);
tipc_node_put(n); tipc_node_put(n);
} }
...@@ -435,7 +440,7 @@ void tipc_node_unsubscribe(struct net *net, struct list_head *subscr, u32 addr) ...@@ -435,7 +440,7 @@ void tipc_node_unsubscribe(struct net *net, struct list_head *subscr, u32 addr)
} }
tipc_node_write_lock(n); tipc_node_write_lock(n);
list_del_init(subscr); list_del_init(subscr);
tipc_node_write_unlock(n); tipc_node_write_unlock_fast(n);
tipc_node_put(n); tipc_node_put(n);
} }
......
...@@ -86,12 +86,12 @@ struct outqueue_entry { ...@@ -86,12 +86,12 @@ struct outqueue_entry {
static void tipc_recv_work(struct work_struct *work); static void tipc_recv_work(struct work_struct *work);
static void tipc_send_work(struct work_struct *work); static void tipc_send_work(struct work_struct *work);
static void tipc_clean_outqueues(struct tipc_conn *con); static void tipc_clean_outqueues(struct tipc_conn *con);
static void tipc_sock_release(struct tipc_conn *con);
static void tipc_conn_kref_release(struct kref *kref) static void tipc_conn_kref_release(struct kref *kref)
{ {
struct tipc_conn *con = container_of(kref, struct tipc_conn, kref); struct tipc_conn *con = container_of(kref, struct tipc_conn, kref);
struct sockaddr_tipc *saddr = con->server->saddr; struct tipc_server *s = con->server;
struct sockaddr_tipc *saddr = s->saddr;
struct socket *sock = con->sock; struct socket *sock = con->sock;
struct sock *sk; struct sock *sk;
...@@ -103,9 +103,13 @@ static void tipc_conn_kref_release(struct kref *kref) ...@@ -103,9 +103,13 @@ static void tipc_conn_kref_release(struct kref *kref)
} }
saddr->scope = -TIPC_NODE_SCOPE; saddr->scope = -TIPC_NODE_SCOPE;
kernel_bind(sock, (struct sockaddr *)saddr, sizeof(*saddr)); kernel_bind(sock, (struct sockaddr *)saddr, sizeof(*saddr));
tipc_sock_release(con);
sock_release(sock); sock_release(sock);
con->sock = NULL; con->sock = NULL;
spin_lock_bh(&s->idr_lock);
idr_remove(&s->conn_idr, con->conid);
s->idr_in_use--;
spin_unlock_bh(&s->idr_lock);
} }
tipc_clean_outqueues(con); tipc_clean_outqueues(con);
...@@ -128,8 +132,10 @@ static struct tipc_conn *tipc_conn_lookup(struct tipc_server *s, int conid) ...@@ -128,8 +132,10 @@ static struct tipc_conn *tipc_conn_lookup(struct tipc_server *s, int conid)
spin_lock_bh(&s->idr_lock); spin_lock_bh(&s->idr_lock);
con = idr_find(&s->conn_idr, conid); con = idr_find(&s->conn_idr, conid);
if (con) if (con && test_bit(CF_CONNECTED, &con->flags))
conn_get(con); conn_get(con);
else
con = NULL;
spin_unlock_bh(&s->idr_lock); spin_unlock_bh(&s->idr_lock);
return con; return con;
} }
...@@ -186,26 +192,15 @@ static void tipc_unregister_callbacks(struct tipc_conn *con) ...@@ -186,26 +192,15 @@ static void tipc_unregister_callbacks(struct tipc_conn *con)
write_unlock_bh(&sk->sk_callback_lock); write_unlock_bh(&sk->sk_callback_lock);
} }
static void tipc_sock_release(struct tipc_conn *con)
{
struct tipc_server *s = con->server;
if (con->conid)
s->tipc_conn_release(con->conid, con->usr_data);
tipc_unregister_callbacks(con);
}
static void tipc_close_conn(struct tipc_conn *con) static void tipc_close_conn(struct tipc_conn *con)
{ {
struct tipc_server *s = con->server; struct tipc_server *s = con->server;
if (test_and_clear_bit(CF_CONNECTED, &con->flags)) { if (test_and_clear_bit(CF_CONNECTED, &con->flags)) {
tipc_unregister_callbacks(con);
spin_lock_bh(&s->idr_lock); if (con->conid)
idr_remove(&s->conn_idr, con->conid); s->tipc_conn_release(con->conid, con->usr_data);
s->idr_in_use--;
spin_unlock_bh(&s->idr_lock);
/* We shouldn't flush pending works as we may be in the /* We shouldn't flush pending works as we may be in the
* thread. In fact the races with pending rx/tx work structs * thread. In fact the races with pending rx/tx work structs
...@@ -458,6 +453,11 @@ int tipc_conn_sendmsg(struct tipc_server *s, int conid, ...@@ -458,6 +453,11 @@ int tipc_conn_sendmsg(struct tipc_server *s, int conid,
if (!con) if (!con)
return -EINVAL; return -EINVAL;
if (!test_bit(CF_CONNECTED, &con->flags)) {
conn_put(con);
return 0;
}
e = tipc_alloc_entry(data, len); e = tipc_alloc_entry(data, len);
if (!e) { if (!e) {
conn_put(con); conn_put(con);
...@@ -471,12 +471,8 @@ int tipc_conn_sendmsg(struct tipc_server *s, int conid, ...@@ -471,12 +471,8 @@ int tipc_conn_sendmsg(struct tipc_server *s, int conid,
list_add_tail(&e->list, &con->outqueue); list_add_tail(&e->list, &con->outqueue);
spin_unlock_bh(&con->outqueue_lock); spin_unlock_bh(&con->outqueue_lock);
if (test_bit(CF_CONNECTED, &con->flags)) {
if (!queue_work(s->send_wq, &con->swork)) if (!queue_work(s->send_wq, &con->swork))
conn_put(con); conn_put(con);
} else {
conn_put(con);
}
return 0; return 0;
} }
...@@ -500,7 +496,7 @@ static void tipc_send_to_sock(struct tipc_conn *con) ...@@ -500,7 +496,7 @@ static void tipc_send_to_sock(struct tipc_conn *con)
int ret; int ret;
spin_lock_bh(&con->outqueue_lock); spin_lock_bh(&con->outqueue_lock);
while (1) { while (test_bit(CF_CONNECTED, &con->flags)) {
e = list_entry(con->outqueue.next, struct outqueue_entry, e = list_entry(con->outqueue.next, struct outqueue_entry,
list); list);
if ((struct list_head *) e == &con->outqueue) if ((struct list_head *) e == &con->outqueue)
...@@ -623,14 +619,12 @@ int tipc_server_start(struct tipc_server *s) ...@@ -623,14 +619,12 @@ int tipc_server_start(struct tipc_server *s)
void tipc_server_stop(struct tipc_server *s) void tipc_server_stop(struct tipc_server *s)
{ {
struct tipc_conn *con; struct tipc_conn *con;
int total = 0;
int id; int id;
spin_lock_bh(&s->idr_lock); spin_lock_bh(&s->idr_lock);
for (id = 0; total < s->idr_in_use; id++) { for (id = 0; s->idr_in_use; id++) {
con = idr_find(&s->conn_idr, id); con = idr_find(&s->conn_idr, id);
if (con) { if (con) {
total++;
spin_unlock_bh(&s->idr_lock); spin_unlock_bh(&s->idr_lock);
tipc_close_conn(con); tipc_close_conn(con);
spin_lock_bh(&s->idr_lock); spin_lock_bh(&s->idr_lock);
......
...@@ -54,6 +54,8 @@ struct tipc_subscriber { ...@@ -54,6 +54,8 @@ struct tipc_subscriber {
static void tipc_subscrp_delete(struct tipc_subscription *sub); static void tipc_subscrp_delete(struct tipc_subscription *sub);
static void tipc_subscrb_put(struct tipc_subscriber *subscriber); static void tipc_subscrb_put(struct tipc_subscriber *subscriber);
static void tipc_subscrp_put(struct tipc_subscription *subscription);
static void tipc_subscrp_get(struct tipc_subscription *subscription);
/** /**
* htohl - convert value to endianness used by destination * htohl - convert value to endianness used by destination
...@@ -123,6 +125,7 @@ void tipc_subscrp_report_overlap(struct tipc_subscription *sub, u32 found_lower, ...@@ -123,6 +125,7 @@ void tipc_subscrp_report_overlap(struct tipc_subscription *sub, u32 found_lower,
{ {
struct tipc_name_seq seq; struct tipc_name_seq seq;
tipc_subscrp_get(sub);
tipc_subscrp_convert_seq(&sub->evt.s.seq, sub->swap, &seq); tipc_subscrp_convert_seq(&sub->evt.s.seq, sub->swap, &seq);
if (!tipc_subscrp_check_overlap(&seq, found_lower, found_upper)) if (!tipc_subscrp_check_overlap(&seq, found_lower, found_upper))
return; return;
...@@ -132,30 +135,23 @@ void tipc_subscrp_report_overlap(struct tipc_subscription *sub, u32 found_lower, ...@@ -132,30 +135,23 @@ void tipc_subscrp_report_overlap(struct tipc_subscription *sub, u32 found_lower,
tipc_subscrp_send_event(sub, found_lower, found_upper, event, port_ref, tipc_subscrp_send_event(sub, found_lower, found_upper, event, port_ref,
node); node);
tipc_subscrp_put(sub);
} }
static void tipc_subscrp_timeout(unsigned long data) static void tipc_subscrp_timeout(unsigned long data)
{ {
struct tipc_subscription *sub = (struct tipc_subscription *)data; struct tipc_subscription *sub = (struct tipc_subscription *)data;
struct tipc_subscriber *subscriber = sub->subscriber;
/* Notify subscriber of timeout */ /* Notify subscriber of timeout */
tipc_subscrp_send_event(sub, sub->evt.s.seq.lower, sub->evt.s.seq.upper, tipc_subscrp_send_event(sub, sub->evt.s.seq.lower, sub->evt.s.seq.upper,
TIPC_SUBSCR_TIMEOUT, 0, 0); TIPC_SUBSCR_TIMEOUT, 0, 0);
spin_lock_bh(&subscriber->lock); tipc_subscrp_put(sub);
tipc_subscrp_delete(sub);
spin_unlock_bh(&subscriber->lock);
tipc_subscrb_put(subscriber);
} }
static void tipc_subscrb_kref_release(struct kref *kref) static void tipc_subscrb_kref_release(struct kref *kref)
{ {
struct tipc_subscriber *subcriber = container_of(kref, kfree(container_of(kref,struct tipc_subscriber, kref));
struct tipc_subscriber, kref);
kfree(subcriber);
} }
static void tipc_subscrb_put(struct tipc_subscriber *subscriber) static void tipc_subscrb_put(struct tipc_subscriber *subscriber)
...@@ -168,6 +164,59 @@ static void tipc_subscrb_get(struct tipc_subscriber *subscriber) ...@@ -168,6 +164,59 @@ static void tipc_subscrb_get(struct tipc_subscriber *subscriber)
kref_get(&subscriber->kref); kref_get(&subscriber->kref);
} }
static void tipc_subscrp_kref_release(struct kref *kref)
{
struct tipc_subscription *sub = container_of(kref,
struct tipc_subscription,
kref);
struct tipc_net *tn = net_generic(sub->net, tipc_net_id);
struct tipc_subscriber *subscriber = sub->subscriber;
spin_lock_bh(&subscriber->lock);
tipc_nametbl_unsubscribe(sub);
list_del(&sub->subscrp_list);
atomic_dec(&tn->subscription_count);
spin_unlock_bh(&subscriber->lock);
kfree(sub);
tipc_subscrb_put(subscriber);
}
static void tipc_subscrp_put(struct tipc_subscription *subscription)
{
kref_put(&subscription->kref, tipc_subscrp_kref_release);
}
static void tipc_subscrp_get(struct tipc_subscription *subscription)
{
kref_get(&subscription->kref);
}
/* tipc_subscrb_subscrp_delete - delete a specific subscription or all
* subscriptions for a given subscriber.
*/
static void tipc_subscrb_subscrp_delete(struct tipc_subscriber *subscriber,
struct tipc_subscr *s)
{
struct list_head *subscription_list = &subscriber->subscrp_list;
struct tipc_subscription *sub, *temp;
spin_lock_bh(&subscriber->lock);
list_for_each_entry_safe(sub, temp, subscription_list, subscrp_list) {
if (s && memcmp(s, &sub->evt.s, sizeof(struct tipc_subscr)))
continue;
tipc_subscrp_get(sub);
spin_unlock_bh(&subscriber->lock);
tipc_subscrp_delete(sub);
tipc_subscrp_put(sub);
spin_lock_bh(&subscriber->lock);
if (s)
break;
}
spin_unlock_bh(&subscriber->lock);
}
static struct tipc_subscriber *tipc_subscrb_create(int conid) static struct tipc_subscriber *tipc_subscrb_create(int conid)
{ {
struct tipc_subscriber *subscriber; struct tipc_subscriber *subscriber;
...@@ -177,8 +226,8 @@ static struct tipc_subscriber *tipc_subscrb_create(int conid) ...@@ -177,8 +226,8 @@ static struct tipc_subscriber *tipc_subscrb_create(int conid)
pr_warn("Subscriber rejected, no memory\n"); pr_warn("Subscriber rejected, no memory\n");
return NULL; return NULL;
} }
kref_init(&subscriber->kref);
INIT_LIST_HEAD(&subscriber->subscrp_list); INIT_LIST_HEAD(&subscriber->subscrp_list);
kref_init(&subscriber->kref);
subscriber->conid = conid; subscriber->conid = conid;
spin_lock_init(&subscriber->lock); spin_lock_init(&subscriber->lock);
...@@ -187,55 +236,22 @@ static struct tipc_subscriber *tipc_subscrb_create(int conid) ...@@ -187,55 +236,22 @@ static struct tipc_subscriber *tipc_subscrb_create(int conid)
static void tipc_subscrb_delete(struct tipc_subscriber *subscriber) static void tipc_subscrb_delete(struct tipc_subscriber *subscriber)
{ {
struct tipc_subscription *sub, *temp; tipc_subscrb_subscrp_delete(subscriber, NULL);
u32 timeout;
spin_lock_bh(&subscriber->lock);
/* Destroy any existing subscriptions for subscriber */
list_for_each_entry_safe(sub, temp, &subscriber->subscrp_list,
subscrp_list) {
timeout = htohl(sub->evt.s.timeout, sub->swap);
if ((timeout == TIPC_WAIT_FOREVER) || del_timer(&sub->timer)) {
tipc_subscrp_delete(sub);
tipc_subscrb_put(subscriber);
}
}
spin_unlock_bh(&subscriber->lock);
tipc_subscrb_put(subscriber); tipc_subscrb_put(subscriber);
} }
static void tipc_subscrp_delete(struct tipc_subscription *sub) static void tipc_subscrp_delete(struct tipc_subscription *sub)
{ {
struct tipc_net *tn = net_generic(sub->net, tipc_net_id); u32 timeout = htohl(sub->evt.s.timeout, sub->swap);
tipc_nametbl_unsubscribe(sub); if (timeout == TIPC_WAIT_FOREVER || del_timer(&sub->timer))
list_del(&sub->subscrp_list); tipc_subscrp_put(sub);
kfree(sub);
atomic_dec(&tn->subscription_count);
} }
static void tipc_subscrp_cancel(struct tipc_subscr *s, static void tipc_subscrp_cancel(struct tipc_subscr *s,
struct tipc_subscriber *subscriber) struct tipc_subscriber *subscriber)
{ {
struct tipc_subscription *sub, *temp; tipc_subscrb_subscrp_delete(subscriber, s);
u32 timeout;
spin_lock_bh(&subscriber->lock);
/* Find first matching subscription, exit if not found */
list_for_each_entry_safe(sub, temp, &subscriber->subscrp_list,
subscrp_list) {
if (!memcmp(s, &sub->evt.s, sizeof(struct tipc_subscr))) {
timeout = htohl(sub->evt.s.timeout, sub->swap);
if ((timeout == TIPC_WAIT_FOREVER) ||
del_timer(&sub->timer)) {
tipc_subscrp_delete(sub);
tipc_subscrb_put(subscriber);
}
break;
}
}
spin_unlock_bh(&subscriber->lock);
} }
static struct tipc_subscription *tipc_subscrp_create(struct net *net, static struct tipc_subscription *tipc_subscrp_create(struct net *net,
...@@ -272,6 +288,7 @@ static struct tipc_subscription *tipc_subscrp_create(struct net *net, ...@@ -272,6 +288,7 @@ static struct tipc_subscription *tipc_subscrp_create(struct net *net,
sub->swap = swap; sub->swap = swap;
memcpy(&sub->evt.s, s, sizeof(*s)); memcpy(&sub->evt.s, s, sizeof(*s));
atomic_inc(&tn->subscription_count); atomic_inc(&tn->subscription_count);
kref_init(&sub->kref);
return sub; return sub;
} }
...@@ -288,16 +305,15 @@ static void tipc_subscrp_subscribe(struct net *net, struct tipc_subscr *s, ...@@ -288,16 +305,15 @@ static void tipc_subscrp_subscribe(struct net *net, struct tipc_subscr *s,
spin_lock_bh(&subscriber->lock); spin_lock_bh(&subscriber->lock);
list_add(&sub->subscrp_list, &subscriber->subscrp_list); list_add(&sub->subscrp_list, &subscriber->subscrp_list);
tipc_subscrb_get(subscriber);
sub->subscriber = subscriber; sub->subscriber = subscriber;
tipc_nametbl_subscribe(sub); tipc_nametbl_subscribe(sub);
tipc_subscrb_get(subscriber);
spin_unlock_bh(&subscriber->lock); spin_unlock_bh(&subscriber->lock);
setup_timer(&sub->timer, tipc_subscrp_timeout, (unsigned long)sub);
timeout = htohl(sub->evt.s.timeout, swap); timeout = htohl(sub->evt.s.timeout, swap);
if (timeout == TIPC_WAIT_FOREVER)
return;
setup_timer(&sub->timer, tipc_subscrp_timeout, (unsigned long)sub); if (timeout != TIPC_WAIT_FOREVER)
mod_timer(&sub->timer, jiffies + msecs_to_jiffies(timeout)); mod_timer(&sub->timer, jiffies + msecs_to_jiffies(timeout));
} }
......
...@@ -57,6 +57,7 @@ struct tipc_subscriber; ...@@ -57,6 +57,7 @@ struct tipc_subscriber;
* @evt: template for events generated by subscription * @evt: template for events generated by subscription
*/ */
struct tipc_subscription { struct tipc_subscription {
struct kref kref;
struct tipc_subscriber *subscriber; struct tipc_subscriber *subscriber;
struct net *net; struct net *net;
struct timer_list timer; struct timer_list timer;
......
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