3 files changed, 116 insertions, 90 deletions

diff --git a/net/tipc/core.c b/net/tipc/core.c
index 6d6aa5a3c240..3d97386af095 100644
--- a/net/tipc/core.c
+++ b/net/tipc/core.c
@@ -164,8 +164,7 @@ int tipc_core_start(void)
         tipc_mode = TIPC_NODE_MODE;
 
         if ((res = tipc_handler_start()) ||
-            (res = tipc_ref_table_init(tipc_max_ports + tipc_max_subscriptions,
-                                       tipc_random)) ||
+            (res = tipc_ref_table_init(tipc_max_ports, tipc_random)) ||
             (res = tipc_reg_start()) ||
             (res = tipc_nametbl_init()) ||
             (res = tipc_k_signal((Handler)tipc_subscr_start, 0)) ||
diff --git a/net/tipc/subscr.c b/net/tipc/subscr.c
index dde23f1e7542..7c62791eb0cc 100644
--- a/net/tipc/subscr.c
+++ b/net/tipc/subscr.c
@@ -38,23 +38,22 @@
 #include "dbg.h"
 #include "subscr.h"
 #include "name_table.h"
+#include "port.h"
 #include "ref.h"
 
 /**
  * struct subscriber - TIPC network topology subscriber
- * @ref: object reference to subscriber object itself
- * @lock: pointer to spinlock controlling access to subscriber object
+ * @port_ref: object reference to server port connecting to subscriber
+ * @lock: pointer to spinlock controlling access to subscriber's server port
  * @subscriber_list: adjacent subscribers in top. server's list of subscribers
  * @subscription_list: list of subscription objects for this subscriber
- * @port_ref: object reference to port used to communicate with subscriber
  */
 
 struct subscriber {
-        u32 ref;
+        u32 port_ref;
         spinlock_t *lock;
         struct list_head subscriber_list;
         struct list_head subscription_list;
-        u32 port_ref;
 };
 
 /**
@@ -91,6 +90,9 @@ static u32 htohl(u32 in, int swap)
 
 /**
  * subscr_send_event - send a message containing a tipc_event to the subscriber
+ *
+ * Note: Must not hold subscriber's server port lock, since tipc_send() will
+ *       try to take the lock if the message is rejected and returned!
  */
 
 static void subscr_send_event(struct subscription *sub,
@@ -110,7 +112,7 @@ static void subscr_send_event(struct subscription *sub,
         sub->evt.found_upper = htohl(found_upper, sub->swap);
         sub->evt.port.ref = htohl(port_ref, sub->swap);
         sub->evt.port.node = htohl(node, sub->swap);
-        tipc_send(sub->owner->port_ref, 1, &msg_sect);
+        tipc_send(sub->server_ref, 1, &msg_sect);
 }
 
 /**
@@ -163,20 +165,18 @@ void tipc_subscr_report_overlap(struct subscription *sub,
 
 static void subscr_timeout(struct subscription *sub)
 {
-        struct subscriber *subscriber;
-        u32 subscriber_ref;
+        struct port *server_port;
 
-        /* Validate subscriber reference (in case subscriber is terminating) */
+        /* Validate server port reference (in case subscriber is terminating) */
 
-        subscriber_ref = sub->owner->ref;
-        subscriber = (struct subscriber *)tipc_ref_lock(subscriber_ref);
-        if (subscriber == NULL)
+        server_port = tipc_port_lock(sub->server_ref);
+        if (server_port == NULL)
                 return;
 
         /* Validate timeout (in case subscription is being cancelled) */
 
         if (sub->timeout == TIPC_WAIT_FOREVER) {
-                tipc_ref_unlock(subscriber_ref);
+                tipc_port_unlock(server_port);
                 return;
         }
 
@@ -184,19 +184,21 @@ static void subscr_timeout(struct subscription *sub)
 
         tipc_nametbl_unsubscribe(sub);
 
-        /* Notify subscriber of timeout, then unlink subscription */
+        /* Unlink subscription from subscriber */
 
-        subscr_send_event(sub,
-                          sub->evt.s.seq.lower,
-                          sub->evt.s.seq.upper,
-                          TIPC_SUBSCR_TIMEOUT,
-                          0,
-                          0);
         list_del(&sub->subscription_list);
 
+        /* Release subscriber's server port */
+
+        tipc_port_unlock(server_port);
+
+        /* Notify subscriber of timeout */
+
+        subscr_send_event(sub, sub->evt.s.seq.lower, sub->evt.s.seq.upper,
+                          TIPC_SUBSCR_TIMEOUT, 0, 0);
+
         /* Now destroy subscription */
 
-        tipc_ref_unlock(subscriber_ref);
         k_term_timer(&sub->timer);
         kfree(sub);
         atomic_dec(&topsrv.subscription_count);
@@ -205,7 +207,7 @@ static void subscr_timeout(struct subscription *sub)
 /**
  * subscr_del - delete a subscription within a subscription list
  *
- * Called with subscriber locked.
+ * Called with subscriber port locked.
  */
 
 static void subscr_del(struct subscription *sub)
@@ -219,7 +221,7 @@ static void subscr_del(struct subscription *sub)
 /**
  * subscr_terminate - terminate communication with a subscriber
  *
- * Called with subscriber locked.  Routine must temporarily release this lock
+ * Called with subscriber port locked.  Routine must temporarily release lock
  * to enable subscription timeout routine(s) to finish without deadlocking;
  * the lock is then reclaimed to allow caller to release it upon return.
  * (This should work even in the unlikely event some other thread creates
@@ -229,14 +231,21 @@ static void subscr_del(struct subscription *sub)
 
 static void subscr_terminate(struct subscriber *subscriber)
 {
+        u32 port_ref;
         struct subscription *sub;
         struct subscription *sub_temp;
 
         /* Invalidate subscriber reference */
 
-        tipc_ref_discard(subscriber->ref);
+        port_ref = subscriber->port_ref;
+        subscriber->port_ref = 0;
         spin_unlock_bh(subscriber->lock);
 
+        /* Sever connection to subscriber */
+
+        tipc_shutdown(port_ref);
+        tipc_deleteport(port_ref);
+
         /* Destroy any existing subscriptions for subscriber */
 
         list_for_each_entry_safe(sub, sub_temp, &subscriber->subscription_list,
@@ -250,27 +259,25 @@ static void subscr_terminate(struct subscriber *subscriber)
                 subscr_del(sub);
         }
 
-        /* Sever connection to subscriber */
-
-        tipc_shutdown(subscriber->port_ref);
-        tipc_deleteport(subscriber->port_ref);
-
         /* Remove subscriber from topology server's subscriber list */
 
         spin_lock_bh(&topsrv.lock);
         list_del(&subscriber->subscriber_list);
         spin_unlock_bh(&topsrv.lock);
 
-        /* Now destroy subscriber */
+        /* Reclaim subscriber lock */
 
         spin_lock_bh(subscriber->lock);
+
+        /* Now destroy subscriber */
+
         kfree(subscriber);
 }
 
 /**
  * subscr_cancel - handle subscription cancellation request
  *
- * Called with subscriber locked.  Routine must temporarily release this lock
+ * Called with subscriber port locked.  Routine must temporarily release lock
  * to enable the subscription timeout routine to finish without deadlocking;
  * the lock is then reclaimed to allow caller to release it upon return.
  *
@@ -313,11 +320,11 @@ static void subscr_cancel(struct tipc_subscr *s,
 /**
  * subscr_subscribe - create subscription for subscriber
  *
- * Called with subscriber locked
+ * Called with subscriber port locked.
  */
 
-static void subscr_subscribe(struct tipc_subscr *s,
-                             struct subscriber *subscriber)
+static struct subscription *subscr_subscribe(struct tipc_subscr *s,
+                                             struct subscriber *subscriber)
 {
         struct subscription *sub;
         int swap;
@@ -331,7 +338,7 @@ static void subscr_subscribe(struct tipc_subscr *s,
         if (s->filter & htohl(TIPC_SUB_CANCEL, swap)) {
                 s->filter &= ~htohl(TIPC_SUB_CANCEL, swap);
                 subscr_cancel(s, subscriber);
-                return;
+                return NULL;
         }
 
         /* Refuse subscription if global limit exceeded */
@@ -340,16 +347,16 @@ static void subscr_subscribe(struct tipc_subscr *s,
                 warn("Subscription rejected, subscription limit reached (%u)\n",
                      tipc_max_subscriptions);
                 subscr_terminate(subscriber);
-                return;
+                return NULL;
         }
 
         /* Allocate subscription object */
 
-        sub = kzalloc(sizeof(*sub), GFP_ATOMIC);
+        sub = kmalloc(sizeof(*sub), GFP_ATOMIC);
         if (!sub) {
                 warn("Subscription rejected, no memory\n");
                 subscr_terminate(subscriber);
-                return;
+                return NULL;
         }
 
         /* Initialize subscription object */
@@ -365,40 +372,41 @@ static void subscr_subscribe(struct tipc_subscr *s,
                 warn("Subscription rejected, illegal request\n");
                 kfree(sub);
                 subscr_terminate(subscriber);
-                return;
+                return NULL;
         }
         sub->event_cb = subscr_send_event;
-        memcpy(&sub->evt.s, s, sizeof(struct tipc_subscr));
-        INIT_LIST_HEAD(&sub->subscription_list);
         INIT_LIST_HEAD(&sub->nameseq_list);
         list_add(&sub->subscription_list, &subscriber->subscription_list);
+        sub->server_ref = subscriber->port_ref;
         sub->swap = swap;
+        memcpy(&sub->evt.s, s, sizeof(struct tipc_subscr));
         atomic_inc(&topsrv.subscription_count);
         if (sub->timeout != TIPC_WAIT_FOREVER) {
                 k_init_timer(&sub->timer,
                              (Handler)subscr_timeout, (unsigned long)sub);
                 k_start_timer(&sub->timer, sub->timeout);
         }
-        sub->owner = subscriber;
-        tipc_nametbl_subscribe(sub);
+
+        return sub;
 }
 
 /**
  * subscr_conn_shutdown_event - handle termination request from subscriber
+ *
+ * Called with subscriber's server port unlocked.
  */
 
 static void subscr_conn_shutdown_event(void *usr_handle,
-                                       u32 portref,
+                                       u32 port_ref,
                                        struct sk_buff **buf,
                                        unsigned char const *data,
                                        unsigned int size,
                                        int reason)
 {
-        struct subscriber *subscriber;
+        struct subscriber *subscriber = usr_handle;
         spinlock_t *subscriber_lock;
 
-        subscriber = tipc_ref_lock((u32)(unsigned long)usr_handle);
-        if (subscriber == NULL)
+        if (tipc_port_lock(port_ref) == NULL)
                 return;
 
         subscriber_lock = subscriber->lock;
@@ -408,6 +416,8 @@ static void subscr_conn_shutdown_event(void *usr_handle,
 
 /**
  * subscr_conn_msg_event - handle new subscription request from subscriber
+ *
+ * Called with subscriber's server port unlocked.
  */
 
 static void subscr_conn_msg_event(void *usr_handle,
@@ -416,20 +426,46 @@ static void subscr_conn_msg_event(void *usr_handle,
                                   const unchar *data,
                                   u32 size)
 {
-        struct subscriber *subscriber;
+        struct subscriber *subscriber = usr_handle;
         spinlock_t *subscriber_lock;
+        struct subscription *sub;
+
+        /*
+         * Lock subscriber's server port (& make a local copy of lock pointer,
+         * in case subscriber is deleted while processing subscription request)
+         */
 
-        subscriber = tipc_ref_lock((u32)(unsigned long)usr_handle);
-        if (subscriber == NULL)
+        if (tipc_port_lock(port_ref) == NULL)
                 return;
 
         subscriber_lock = subscriber->lock;
-        if (size != sizeof(struct tipc_subscr))
-                subscr_terminate(subscriber);
-        else
-                subscr_subscribe((struct tipc_subscr *)data, subscriber);
 
-        spin_unlock_bh(subscriber_lock);
+        if (size != sizeof(struct tipc_subscr)) {
+                subscr_terminate(subscriber);
+                spin_unlock_bh(subscriber_lock);
+        } else {
+                sub = subscr_subscribe((struct tipc_subscr *)data, subscriber);
+                spin_unlock_bh(subscriber_lock);
+                if (sub != NULL) {
+
+                        /*
+                         * We must release the server port lock before adding a
+                         * subscription to the name table since TIPC needs to be
+                         * able to (re)acquire the port lock if an event message
+                         * issued by the subscription process is rejected and
+                         * returned.  The subscription cannot be deleted while
+                         * it is being added to the name table because:
+                         * a) the single-threading of the native API port code
+                         *    ensures the subscription cannot be cancelled and
+                         *    the subscriber connection cannot be broken, and
+                         * b) the name table lock ensures the subscription
+                         *    timeout code cannot delete the subscription,
+                         * so the subscription object is still protected.
+                         */
+
+                        tipc_nametbl_subscribe(sub);
+                }
+        }
 }
 
 /**
@@ -445,16 +481,10 @@ static void subscr_named_msg_event(void *usr_handle,
                                    struct tipc_portid const *orig,
                                    struct tipc_name_seq const *dest)
 {
-        struct subscriber *subscriber;
-        struct iovec msg_sect = {NULL, 0};
-        spinlock_t *subscriber_lock;
+        static struct iovec msg_sect = {NULL, 0};
 
-        dbg("subscr_named_msg_event: orig = %x own = %x,\n",
-            orig->node, tipc_own_addr);
-        if (size && (size != sizeof(struct tipc_subscr))) {
-                warn("Subscriber rejected, invalid subscription size\n");
-                return;
-        }
+        struct subscriber *subscriber;
+        u32 server_port_ref;
 
         /* Create subscriber object */
 
@@ -465,18 +495,11 @@ static void subscr_named_msg_event(void *usr_handle,
         }
         INIT_LIST_HEAD(&subscriber->subscription_list);
         INIT_LIST_HEAD(&subscriber->subscriber_list);
-        subscriber->ref = tipc_ref_acquire(subscriber, &subscriber->lock);
-        if (subscriber->ref == 0) {
-                warn("Subscriber rejected, reference table exhausted\n");
-                kfree(subscriber);
-                return;
-        }
-        spin_unlock_bh(subscriber->lock);
 
-        /* Establish a connection to subscriber */
+        /* Create server port & establish connection to subscriber */
 
         tipc_createport(topsrv.user_ref,
-                        (void *)(unsigned long)subscriber->ref,
+                        subscriber,
                         importance,
                         NULL,
                         NULL,
@@ -488,32 +511,36 @@ static void subscr_named_msg_event(void *usr_handle,
                         &subscriber->port_ref);
         if (subscriber->port_ref == 0) {
                 warn("Subscriber rejected, unable to create port\n");
-                tipc_ref_discard(subscriber->ref);
                 kfree(subscriber);
                 return;
         }
         tipc_connect2port(subscriber->port_ref, orig);
 
+        /* Lock server port (& save lock address for future use) */
+
+        subscriber->lock = tipc_port_lock(subscriber->port_ref)->publ.lock;
 
         /* Add subscriber to topology server's subscriber list */
 
-        tipc_ref_lock(subscriber->ref);
         spin_lock_bh(&topsrv.lock);
         list_add(&subscriber->subscriber_list, &topsrv.subscriber_list);
         spin_unlock_bh(&topsrv.lock);
 
-        /*
-         * Subscribe now if message contains a subscription,
-         * otherwise send an empty response to complete connection handshaking
-         */
+        /* Unlock server port */
 
-        subscriber_lock = subscriber->lock;
-        if (size)
-                subscr_subscribe((struct tipc_subscr *)data, subscriber);
-        else
-                tipc_send(subscriber->port_ref, 1, &msg_sect);
+        server_port_ref = subscriber->port_ref;
+        spin_unlock_bh(subscriber->lock);
 
-        spin_unlock_bh(subscriber_lock);
+        /* Send an ACK- to complete connection handshaking */
+
+        tipc_send(server_port_ref, 1, &msg_sect);
+
+        /* Handle optional subscription request */
+
+        if (size != 0) {
+                subscr_conn_msg_event(subscriber, server_port_ref,
+                                      buf, data, size);
+        }
 }
 
 int tipc_subscr_start(void)
@@ -572,8 +599,8 @@ void tipc_subscr_stop(void)
                 list_for_each_entry_safe(subscriber, subscriber_temp,
                                          &topsrv.subscriber_list,
                                          subscriber_list) {
-                        tipc_ref_lock(subscriber->ref);
                         subscriber_lock = subscriber->lock;
+                        spin_lock_bh(subscriber_lock);
                         subscr_terminate(subscriber);
                         spin_unlock_bh(subscriber_lock);
                 }
diff --git a/net/tipc/subscr.h b/net/tipc/subscr.h
index 3e3e0265146e..b9af687b6368 100644
--- a/net/tipc/subscr.h
+++ b/net/tipc/subscr.h
@@ -49,12 +49,12 @@ typedef void (*tipc_subscr_event) (struct subscription *sub,
  * @timeout: duration of subscription (in ms)
  * @filter: event filtering to be done for subscription
  * @event_cb: routine invoked when a subscription event is detected
+ * @server_ref: object reference of server port associated with subscription
  * @swap: indicates if subscriber uses opposite endianness in its messages
  * @evt: template for events generated by subscription
  * @subscription_list: adjacent subscriptions in subscriber's subscription list
  * @nameseq_list: adjacent subscriptions in name sequence's subscription list
  * @timer_ref: reference to timer governing subscription duration (may be NULL)
- * @owner: pointer to subscriber object associated with this subscription
  */
 
 struct subscription {
@@ -62,12 +62,12 @@ struct subscription {
         u32 timeout;
         u32 filter;
         tipc_subscr_event event_cb;
+        u32 server_ref;
         int swap;
         struct tipc_event evt;
         struct list_head subscription_list;
         struct list_head nameseq_list;
         struct timer_list timer;
-        struct subscriber *owner;
 };
 
 int tipc_subscr_overlap(struct subscription * sub,