Commit 28353e7f authored by Allan Stephens's avatar Allan Stephens Committed by David S. Miller

tipc: Consolidate subscriber & subscriber port references

This patch modifies TIPC's network topology service so that it
only requires a single reference table entry per subscriber
connection, rather than two.  This is achieved by letting the
reference to the server port communicating with the subscriber
act as the reference to the subscriber object itself.  (Since
the subscriber cannot exist without its port, and vice versa,
this dual role for the reference is perfectly natural.)  This
consolidation reduces the size of the reference table by 50%
in the default configuration.
Signed-off-by: default avatarAllan Stephens <allan.stephens@windriver.com>
Signed-off-by: default avatarDavid S. Miller <davem@davemloft.net>
parent fc5ad582
...@@ -164,8 +164,7 @@ int tipc_core_start(void) ...@@ -164,8 +164,7 @@ int tipc_core_start(void)
tipc_mode = TIPC_NODE_MODE; tipc_mode = TIPC_NODE_MODE;
if ((res = tipc_handler_start()) || if ((res = tipc_handler_start()) ||
(res = tipc_ref_table_init(tipc_max_ports + tipc_max_subscriptions, (res = tipc_ref_table_init(tipc_max_ports, tipc_random)) ||
tipc_random)) ||
(res = tipc_reg_start()) || (res = tipc_reg_start()) ||
(res = tipc_nametbl_init()) || (res = tipc_nametbl_init()) ||
(res = tipc_k_signal((Handler)tipc_subscr_start, 0)) || (res = tipc_k_signal((Handler)tipc_subscr_start, 0)) ||
......
...@@ -38,23 +38,22 @@ ...@@ -38,23 +38,22 @@
#include "dbg.h" #include "dbg.h"
#include "subscr.h" #include "subscr.h"
#include "name_table.h" #include "name_table.h"
#include "port.h"
#include "ref.h" #include "ref.h"
/** /**
* struct subscriber - TIPC network topology subscriber * struct subscriber - TIPC network topology subscriber
* @ref: object reference to subscriber object itself * @port_ref: object reference to server port connecting to subscriber
* @lock: pointer to spinlock controlling access to subscriber object * @lock: pointer to spinlock controlling access to subscriber's server port
* @subscriber_list: adjacent subscribers in top. server's list of subscribers * @subscriber_list: adjacent subscribers in top. server's list of subscribers
* @subscription_list: list of subscription objects for this subscriber * @subscription_list: list of subscription objects for this subscriber
* @port_ref: object reference to port used to communicate with subscriber
*/ */
struct subscriber { struct subscriber {
u32 ref; u32 port_ref;
spinlock_t *lock; spinlock_t *lock;
struct list_head subscriber_list; struct list_head subscriber_list;
struct list_head subscription_list; struct list_head subscription_list;
u32 port_ref;
}; };
/** /**
...@@ -91,6 +90,9 @@ static u32 htohl(u32 in, int swap) ...@@ -91,6 +90,9 @@ static u32 htohl(u32 in, int swap)
/** /**
* subscr_send_event - send a message containing a tipc_event to the subscriber * 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, static void subscr_send_event(struct subscription *sub,
...@@ -110,7 +112,7 @@ 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.found_upper = htohl(found_upper, sub->swap);
sub->evt.port.ref = htohl(port_ref, sub->swap); sub->evt.port.ref = htohl(port_ref, sub->swap);
sub->evt.port.node = htohl(node, 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, ...@@ -163,20 +165,18 @@ void tipc_subscr_report_overlap(struct subscription *sub,
static void subscr_timeout(struct subscription *sub) static void subscr_timeout(struct subscription *sub)
{ {
struct subscriber *subscriber; struct port *server_port;
u32 subscriber_ref;
/* Validate subscriber reference (in case subscriber is terminating) */ /* Validate server port reference (in case subscriber is terminating) */
subscriber_ref = sub->owner->ref; server_port = tipc_port_lock(sub->server_ref);
subscriber = (struct subscriber *)tipc_ref_lock(subscriber_ref); if (server_port == NULL)
if (subscriber == NULL)
return; return;
/* Validate timeout (in case subscription is being cancelled) */ /* Validate timeout (in case subscription is being cancelled) */
if (sub->timeout == TIPC_WAIT_FOREVER) { if (sub->timeout == TIPC_WAIT_FOREVER) {
tipc_ref_unlock(subscriber_ref); tipc_port_unlock(server_port);
return; return;
} }
...@@ -184,19 +184,21 @@ static void subscr_timeout(struct subscription *sub) ...@@ -184,19 +184,21 @@ static void subscr_timeout(struct subscription *sub)
tipc_nametbl_unsubscribe(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); 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 */ /* Now destroy subscription */
tipc_ref_unlock(subscriber_ref);
k_term_timer(&sub->timer); k_term_timer(&sub->timer);
kfree(sub); kfree(sub);
atomic_dec(&topsrv.subscription_count); atomic_dec(&topsrv.subscription_count);
...@@ -205,7 +207,7 @@ static void subscr_timeout(struct subscription *sub) ...@@ -205,7 +207,7 @@ static void subscr_timeout(struct subscription *sub)
/** /**
* subscr_del - delete a subscription within a subscription list * 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) static void subscr_del(struct subscription *sub)
...@@ -219,7 +221,7 @@ 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 * 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; * to enable subscription timeout routine(s) to finish without deadlocking;
* the lock is then reclaimed to allow caller to release it upon return. * 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 * (This should work even in the unlikely event some other thread creates
...@@ -229,14 +231,21 @@ static void subscr_del(struct subscription *sub) ...@@ -229,14 +231,21 @@ static void subscr_del(struct subscription *sub)
static void subscr_terminate(struct subscriber *subscriber) static void subscr_terminate(struct subscriber *subscriber)
{ {
u32 port_ref;
struct subscription *sub; struct subscription *sub;
struct subscription *sub_temp; struct subscription *sub_temp;
/* Invalidate subscriber reference */ /* Invalidate subscriber reference */
tipc_ref_discard(subscriber->ref); port_ref = subscriber->port_ref;
subscriber->port_ref = 0;
spin_unlock_bh(subscriber->lock); spin_unlock_bh(subscriber->lock);
/* Sever connection to subscriber */
tipc_shutdown(port_ref);
tipc_deleteport(port_ref);
/* Destroy any existing subscriptions for subscriber */ /* Destroy any existing subscriptions for subscriber */
list_for_each_entry_safe(sub, sub_temp, &subscriber->subscription_list, list_for_each_entry_safe(sub, sub_temp, &subscriber->subscription_list,
...@@ -250,27 +259,25 @@ static void subscr_terminate(struct subscriber *subscriber) ...@@ -250,27 +259,25 @@ static void subscr_terminate(struct subscriber *subscriber)
subscr_del(sub); 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 */ /* Remove subscriber from topology server's subscriber list */
spin_lock_bh(&topsrv.lock); spin_lock_bh(&topsrv.lock);
list_del(&subscriber->subscriber_list); list_del(&subscriber->subscriber_list);
spin_unlock_bh(&topsrv.lock); spin_unlock_bh(&topsrv.lock);
/* Now destroy subscriber */ /* Reclaim subscriber lock */
spin_lock_bh(subscriber->lock); spin_lock_bh(subscriber->lock);
/* Now destroy subscriber */
kfree(subscriber); kfree(subscriber);
} }
/** /**
* subscr_cancel - handle subscription cancellation request * 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; * to enable the subscription timeout routine to finish without deadlocking;
* the lock is then reclaimed to allow caller to release it upon return. * 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, ...@@ -313,11 +320,11 @@ static void subscr_cancel(struct tipc_subscr *s,
/** /**
* subscr_subscribe - create subscription for subscriber * subscr_subscribe - create subscription for subscriber
* *
* Called with subscriber locked * Called with subscriber port locked.
*/ */
static void subscr_subscribe(struct tipc_subscr *s, static struct subscription *subscr_subscribe(struct tipc_subscr *s,
struct subscriber *subscriber) struct subscriber *subscriber)
{ {
struct subscription *sub; struct subscription *sub;
int swap; int swap;
...@@ -331,7 +338,7 @@ static void subscr_subscribe(struct tipc_subscr *s, ...@@ -331,7 +338,7 @@ static void subscr_subscribe(struct tipc_subscr *s,
if (s->filter & htohl(TIPC_SUB_CANCEL, swap)) { if (s->filter & htohl(TIPC_SUB_CANCEL, swap)) {
s->filter &= ~htohl(TIPC_SUB_CANCEL, swap); s->filter &= ~htohl(TIPC_SUB_CANCEL, swap);
subscr_cancel(s, subscriber); subscr_cancel(s, subscriber);
return; return NULL;
} }
/* Refuse subscription if global limit exceeded */ /* Refuse subscription if global limit exceeded */
...@@ -340,16 +347,16 @@ static void subscr_subscribe(struct tipc_subscr *s, ...@@ -340,16 +347,16 @@ static void subscr_subscribe(struct tipc_subscr *s,
warn("Subscription rejected, subscription limit reached (%u)\n", warn("Subscription rejected, subscription limit reached (%u)\n",
tipc_max_subscriptions); tipc_max_subscriptions);
subscr_terminate(subscriber); subscr_terminate(subscriber);
return; return NULL;
} }
/* Allocate subscription object */ /* Allocate subscription object */
sub = kzalloc(sizeof(*sub), GFP_ATOMIC); sub = kmalloc(sizeof(*sub), GFP_ATOMIC);
if (!sub) { if (!sub) {
warn("Subscription rejected, no memory\n"); warn("Subscription rejected, no memory\n");
subscr_terminate(subscriber); subscr_terminate(subscriber);
return; return NULL;
} }
/* Initialize subscription object */ /* Initialize subscription object */
...@@ -365,40 +372,41 @@ static void subscr_subscribe(struct tipc_subscr *s, ...@@ -365,40 +372,41 @@ static void subscr_subscribe(struct tipc_subscr *s,
warn("Subscription rejected, illegal request\n"); warn("Subscription rejected, illegal request\n");
kfree(sub); kfree(sub);
subscr_terminate(subscriber); subscr_terminate(subscriber);
return; return NULL;
} }
sub->event_cb = subscr_send_event; 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); INIT_LIST_HEAD(&sub->nameseq_list);
list_add(&sub->subscription_list, &subscriber->subscription_list); list_add(&sub->subscription_list, &subscriber->subscription_list);
sub->server_ref = subscriber->port_ref;
sub->swap = swap; sub->swap = swap;
memcpy(&sub->evt.s, s, sizeof(struct tipc_subscr));
atomic_inc(&topsrv.subscription_count); atomic_inc(&topsrv.subscription_count);
if (sub->timeout != TIPC_WAIT_FOREVER) { if (sub->timeout != TIPC_WAIT_FOREVER) {
k_init_timer(&sub->timer, k_init_timer(&sub->timer,
(Handler)subscr_timeout, (unsigned long)sub); (Handler)subscr_timeout, (unsigned long)sub);
k_start_timer(&sub->timer, sub->timeout); 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 * 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, static void subscr_conn_shutdown_event(void *usr_handle,
u32 portref, u32 port_ref,
struct sk_buff **buf, struct sk_buff **buf,
unsigned char const *data, unsigned char const *data,
unsigned int size, unsigned int size,
int reason) int reason)
{ {
struct subscriber *subscriber; struct subscriber *subscriber = usr_handle;
spinlock_t *subscriber_lock; spinlock_t *subscriber_lock;
subscriber = tipc_ref_lock((u32)(unsigned long)usr_handle); if (tipc_port_lock(port_ref) == NULL)
if (subscriber == NULL)
return; return;
subscriber_lock = subscriber->lock; subscriber_lock = subscriber->lock;
...@@ -408,6 +416,8 @@ static void subscr_conn_shutdown_event(void *usr_handle, ...@@ -408,6 +416,8 @@ static void subscr_conn_shutdown_event(void *usr_handle,
/** /**
* subscr_conn_msg_event - handle new subscription request from subscriber * 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, static void subscr_conn_msg_event(void *usr_handle,
...@@ -416,20 +426,46 @@ 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, const unchar *data,
u32 size) u32 size)
{ {
struct subscriber *subscriber; struct subscriber *subscriber = usr_handle;
spinlock_t *subscriber_lock; 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 (tipc_port_lock(port_ref) == NULL)
if (subscriber == NULL)
return; return;
subscriber_lock = subscriber->lock; 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, ...@@ -445,16 +481,10 @@ static void subscr_named_msg_event(void *usr_handle,
struct tipc_portid const *orig, struct tipc_portid const *orig,
struct tipc_name_seq const *dest) struct tipc_name_seq const *dest)
{ {
struct subscriber *subscriber; static struct iovec msg_sect = {NULL, 0};
struct iovec msg_sect = {NULL, 0};
spinlock_t *subscriber_lock;
dbg("subscr_named_msg_event: orig = %x own = %x,\n", struct subscriber *subscriber;
orig->node, tipc_own_addr); u32 server_port_ref;
if (size && (size != sizeof(struct tipc_subscr))) {
warn("Subscriber rejected, invalid subscription size\n");
return;
}
/* Create subscriber object */ /* Create subscriber object */
...@@ -465,18 +495,11 @@ static void subscr_named_msg_event(void *usr_handle, ...@@ -465,18 +495,11 @@ static void subscr_named_msg_event(void *usr_handle,
} }
INIT_LIST_HEAD(&subscriber->subscription_list); INIT_LIST_HEAD(&subscriber->subscription_list);
INIT_LIST_HEAD(&subscriber->subscriber_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, tipc_createport(topsrv.user_ref,
(void *)(unsigned long)subscriber->ref, subscriber,
importance, importance,
NULL, NULL,
NULL, NULL,
...@@ -488,32 +511,36 @@ static void subscr_named_msg_event(void *usr_handle, ...@@ -488,32 +511,36 @@ static void subscr_named_msg_event(void *usr_handle,
&subscriber->port_ref); &subscriber->port_ref);
if (subscriber->port_ref == 0) { if (subscriber->port_ref == 0) {
warn("Subscriber rejected, unable to create port\n"); warn("Subscriber rejected, unable to create port\n");
tipc_ref_discard(subscriber->ref);
kfree(subscriber); kfree(subscriber);
return; return;
} }
tipc_connect2port(subscriber->port_ref, orig); 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 */ /* Add subscriber to topology server's subscriber list */
tipc_ref_lock(subscriber->ref);
spin_lock_bh(&topsrv.lock); spin_lock_bh(&topsrv.lock);
list_add(&subscriber->subscriber_list, &topsrv.subscriber_list); list_add(&subscriber->subscriber_list, &topsrv.subscriber_list);
spin_unlock_bh(&topsrv.lock); spin_unlock_bh(&topsrv.lock);
/* /* Unlock server port */
* Subscribe now if message contains a subscription,
* otherwise send an empty response to complete connection handshaking
*/
subscriber_lock = subscriber->lock; server_port_ref = subscriber->port_ref;
if (size) spin_unlock_bh(subscriber->lock);
subscr_subscribe((struct tipc_subscr *)data, subscriber);
else
tipc_send(subscriber->port_ref, 1, &msg_sect);
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) int tipc_subscr_start(void)
...@@ -572,8 +599,8 @@ void tipc_subscr_stop(void) ...@@ -572,8 +599,8 @@ void tipc_subscr_stop(void)
list_for_each_entry_safe(subscriber, subscriber_temp, list_for_each_entry_safe(subscriber, subscriber_temp,
&topsrv.subscriber_list, &topsrv.subscriber_list,
subscriber_list) { subscriber_list) {
tipc_ref_lock(subscriber->ref);
subscriber_lock = subscriber->lock; subscriber_lock = subscriber->lock;
spin_lock_bh(subscriber_lock);
subscr_terminate(subscriber); subscr_terminate(subscriber);
spin_unlock_bh(subscriber_lock); spin_unlock_bh(subscriber_lock);
} }
......
...@@ -49,12 +49,12 @@ typedef void (*tipc_subscr_event) (struct subscription *sub, ...@@ -49,12 +49,12 @@ typedef void (*tipc_subscr_event) (struct subscription *sub,
* @timeout: duration of subscription (in ms) * @timeout: duration of subscription (in ms)
* @filter: event filtering to be done for subscription * @filter: event filtering to be done for subscription
* @event_cb: routine invoked when a subscription event is detected * @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 * @swap: indicates if subscriber uses opposite endianness in its messages
* @evt: template for events generated by subscription * @evt: template for events generated by subscription
* @subscription_list: adjacent subscriptions in subscriber's subscription list * @subscription_list: adjacent subscriptions in subscriber's subscription list
* @nameseq_list: adjacent subscriptions in name sequence's subscription list * @nameseq_list: adjacent subscriptions in name sequence's subscription list
* @timer_ref: reference to timer governing subscription duration (may be NULL) * @timer_ref: reference to timer governing subscription duration (may be NULL)
* @owner: pointer to subscriber object associated with this subscription
*/ */
struct subscription { struct subscription {
...@@ -62,12 +62,12 @@ struct subscription { ...@@ -62,12 +62,12 @@ struct subscription {
u32 timeout; u32 timeout;
u32 filter; u32 filter;
tipc_subscr_event event_cb; tipc_subscr_event event_cb;
u32 server_ref;
int swap; int swap;
struct tipc_event evt; struct tipc_event evt;
struct list_head subscription_list; struct list_head subscription_list;
struct list_head nameseq_list; struct list_head nameseq_list;
struct timer_list timer; struct timer_list timer;
struct subscriber *owner;
}; };
int tipc_subscr_overlap(struct subscription * sub, int tipc_subscr_overlap(struct subscription * sub,
......
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