After the design policy of subscription timeout is adjusted,
subscription is still present in name table after it's expired, and
hence the subscriber will receive new events for this subscription
until the subscriber terminates.
Now when subscription is expired, it will be first deleted from name
table and its subscriber, and then schedule its work. Finally the
subscription will be destroyed in workqueue asynchronously.
Signed-off-by: Ying Xue <ying....@windriver.com>
---
net/tipc/server.h | 2 ++
net/tipc/subscr.c | 32 ++++++++++++++++++++++++++++++++
net/tipc/subscr.h | 1 +
3 files changed, 35 insertions(+)
diff --git a/net/tipc/server.h b/net/tipc/server.h
index 34f8055..7d5588f 100644
--- a/net/tipc/server.h
+++ b/net/tipc/server.h
@@ -59,6 +59,7 @@
* @name: server name
* @imp: message importance
* @type: socket type
+ * @wq: workqueue
*/
struct tipc_server {
struct idr conn_idr;
@@ -78,6 +79,7 @@ struct tipc_server {
char name[TIPC_SERVER_NAME_LEN];
int imp;
int type;
+ struct workqueue_struct *wq;
};
int tipc_conn_sendmsg(struct tipc_server *s, int conid,
diff --git a/net/tipc/subscr.c b/net/tipc/subscr.c
index ea67cce..3dcb009 100644
--- a/net/tipc/subscr.c
+++ b/net/tipc/subscr.c
@@ -137,14 +137,34 @@ void tipc_subscrp_report_overlap(struct tipc_subscription
*sub, u32 found_lower,
static void tipc_subscrp_timeout(unsigned long data)
{
struct tipc_subscription *sub = (struct tipc_subscription *)data;
+ struct tipc_subscriber *subscriber = sub->subscriber;
+ struct tipc_net *tn = net_generic(sub->net, tipc_net_id);
/* Notify subscriber of timeout */
tipc_subscrp_send_event(sub, sub->evt.s.seq.lower, sub->evt.s.seq.upper,
TIPC_SUBSCR_TIMEOUT, 0, 0);
+ spin_lock_bh(&subscriber->lock);
+ list_del(&sub->subscrp_list);
+ tipc_nametbl_unsubscribe(sub);
+ spin_unlock_bh(&subscriber->lock);
+
+ queue_work(tn->topsrv->wq, &sub->work);
+
tipc_subscrp_put(sub);
}
+static void tipc_subscrp_work(struct work_struct *work)
+{
+ struct tipc_subscription *sub = container_of(work,
+ struct tipc_subscription, work);
+ struct tipc_subscriber *subscriber = sub->subscriber;
+
+ spin_lock_bh(&subscriber->lock);
+ tipc_subscrp_delete(sub);
+ spin_unlock_bh(&subscriber->lock);
+}
+
static void tipc_subscrb_kref_release(struct kref *kref)
{
kfree(container_of(kref,struct tipc_subscriber, kref));
@@ -282,6 +302,9 @@ static struct tipc_subscription *tipc_subscrp_create(struct
net *net,
memcpy(&sub->evt.s, s, sizeof(*s));
atomic_inc(&tn->subscription_count);
kref_init(&sub->kref);
+
+ INIT_WORK(&sub->work, tipc_subscrp_work);
+
return sub;
}
@@ -379,6 +402,13 @@ int tipc_topsrv_start(struct net *net)
topsrv->tipc_conn_release = tipc_subscrb_release_cb;
strncpy(topsrv->name, name, strlen(name) + 1);
+ topsrv->wq = create_singlethread_workqueue(topsrv->name);
+ if (!topsrv->wq) {
+ kfree(saddr);
+ kfree(topsrv);
+ return -ENOMEM;
+ }
+
tn->topsrv = topsrv;
atomic_set(&tn->subscription_count, 0);
@@ -391,6 +421,8 @@ void tipc_topsrv_stop(struct net *net)
struct tipc_server *topsrv = tn->topsrv;
tipc_server_stop(topsrv);
+ flush_workqueue(topsrv->wq);
+ destroy_workqueue(topsrv->wq);
kfree(topsrv->saddr);
kfree(topsrv);
}
diff --git a/net/tipc/subscr.h b/net/tipc/subscr.h
index ee52957..50fc64b 100644
--- a/net/tipc/subscr.h
+++ b/net/tipc/subscr.h
@@ -65,6 +65,7 @@ struct tipc_subscription {
struct list_head subscrp_list;
int swap;
struct tipc_event evt;
+ struct work_struct work;
};
int tipc_subscrp_check_overlap(struct tipc_name_seq *seq, u32 found_lower,
--
2.7.4
------------------------------------------------------------------------------
Announcing the Oxford Dictionaries API! The API offers world-renowned
dictionary content that is easy and intuitive to access. Sign up for an
account today to start using our lexical data to power your apps and
projects. Get started today and enter our developer competition.
http://sdm.link/oxford
_______________________________________________
tipc-discussion mailing list
tipc-discussion@lists.sourceforge.net
https://lists.sourceforge.net/lists/listinfo/tipc-discussion
