Thread: [Tipc-discussion] More tipc list updates

SourceForge Headquarters 225 Broadway Suite 1600 San Diego, CA 92101 +1 (858) 422-6466

Jon,
Since I didn't hear from you about my last patch yet, I have added to
it. Take a look at tipc_detach in particular.  The original code didn't
make sense to me.

cvs diff -u bcast.h name_table.c name_table.h name_subscr.c port.c name_subscr.h port.h name_distr.c reg.c
Index: bcast.h
===================================================================
RCS file: /cvsroot/tipc/source/unstable/net/tipc/bcast.h,v
retrieving revision 1.10
diff -u -r1.10 bcast.h

--- bcast.h	16 Apr 2004 04:03:46 -0000	1.10
+++ bcast.h	4 May 2004 22:07:46 -0000
@@ -165,6 +165,10 @@
 void tipc_bcast_start(void);
 void tipc_bcast_stop(void);
 
+void bnode_outqueue_release(int ackno);
+int in_list_node(struct list_head *list, tipc_net_addr_t destnode);
+int in_list(struct list_head *list, uint destport, tipc_net_addr_t destnode);
+
 struct bcastlinkset*  blink_select(int selector);
 
 
Index: name_table.c
===================================================================
RCS file: /cvsroot/tipc/source/unstable/net/tipc/name_table.c,v
retrieving revision 1.14
diff -u -r1.14 name_table.c
--- name_table.c	28 Apr 2004 21:55:26 -0000	1.14
+++ name_table.c	4 May 2004 22:07:46 -0000
@@ -118,8 +118,8 @@
 #include "bcast.h"
 
 
-static void nametbl_dump(void);
-static void nametbl_print(struct print_buf *buf, const char *str);
+void nametbl_dump(void);
+void nametbl_print(struct print_buf *buf, const char *str);
 rwlock_t nametbl_lock = RW_LOCK_UNLOCKED;
 
 /*
@@ -196,14 +196,13 @@
 	struct sub_seq* sseqs;
 	uint alloc;
 	uint first_free;
-	struct name_seq *prev;
-	struct name_seq *next;
-	struct name_subscr *subscriptions;
+	struct hlist_node ns_list;
+	struct list_head subscriptions;
 	spinlock_t lock;
 };
 
 struct name_seq *
-nameseq_create(uint type, struct name_seq *next)
+nameseq_create(uint type, struct hlist_head *seq_head)
 {
 	struct name_seq *this = 
 		(struct name_seq *)k_malloc(sizeof(*this));
@@ -211,12 +210,12 @@
 	this->lock = SPIN_LOCK_UNLOCKED;
 	this->type = type;
 	this->sseqs = subseq_alloc(1);
-	dbg("nameseq_create() this = %x type %u, next =%x, ssseqs %x, ff: %u\n",
-            this,type, next,this->sseqs,this->first_free);
+	dbg("nameseq_create() this = %x type %u, ssseqs %x, ff: %u\n",
+            this,type, this->sseqs,this->first_free);
 	this->alloc = 1;
-	this->next = next;
-	if (next)
-		next->prev = this;
+	INIT_HLIST_NODE(&this->ns_list);
+	INIT_LIST_HEAD(&this->subscriptions);
+	hlist_add_head(&this->ns_list, seq_head);
 	return this;
 }
 
@@ -288,7 +287,6 @@
                                         uint scope,
                                         uint key)
 {
-	struct name_subscr *s = this->subscriptions;
 	struct publication *publ;
 	struct sub_seq *sseq;
 	int created_subseq = 0;
@@ -365,18 +363,21 @@
                 }
         }
 
-	if (!created_subseq)
-		return publ;
+	/* 
+	 * Any subscriptions waiting for notification? 
+	 */
+	if (created_subseq) {
+		struct name_subscr *s, *st;
+        	list_for_each_entry_safe(s, st, 
+					&this->subscriptions, nsub_list) {
 
-	/* Any subscriptions waiting ? */
-        while (s) {
-                namesub_report_overlap(s,
+                	namesub_report_overlap(s,
                                        publ->lower,
                                        publ->upper,
                                        TIPC_PUBLISHED,
                                        publ->ref,
                                        publ->node);
-                s = s->next;
+		}
         }
 	return publ;
 }
@@ -388,11 +389,11 @@
                     tipc_net_addr_t node, 
                     uint key)
 {
-	struct name_subscr *s = this->subscriptions;
 	struct publication *publ;
 	struct publication *prev = 0;        
 	struct sub_seq *sseq = nameseq_find_subseq(this,inst);
 	struct sub_seq *free;
+	struct name_subscr *s, *st;
         assert(this);
         if (!sseq) {
                 int i;
@@ -473,20 +474,22 @@
         if (sseq->node_list || sseq->cluster_list || sseq->zone_list)
                 return publ;
 
-        /* No more publications,contract subseq list: */
-
+        /* 
+	 * No more publications,contract subseq list: 
+	 */
         free = &this->sseqs[this->first_free--];
 	memmove(sseq,sseq+1,(free-(sseq+1))*sizeof(*sseq));
 
-	/* Any subscriptions waiting ? */
-        while (s) {
+	/* 
+	 * Any subscriptions waiting ? 
+	 */
+	list_for_each_entry_safe(s, st, &this->subscriptions, nsub_list) {
                 namesub_report_overlap(s,
                                        publ->lower,
                                        publ->upper,
                                        TIPC_WITHDRAWN,
                                        publ->ref,
                                        publ->node);
-                s = s->next;
         }
 	return publ;
 }
@@ -529,8 +532,7 @@
 {
 	struct sub_seq *sseq = this->sseqs;
 
-	s->next = this->subscriptions;
-	this->subscriptions = s;
+	list_add(&s->nsub_list, &this->subscriptions);
 
         if (!sseq) 
                 return;
@@ -545,10 +547,6 @@
 }
 
 
-
-
-
-
 /*
  * struct name_table: translation table containing all existing
  * port name publications. Consists of 'name_seq' objects
@@ -560,7 +558,7 @@
 #define TABLE_SIZE  (1<<14)
 
 struct name_table {
-	struct name_seq *types[TABLE_SIZE];
+	struct hlist_head types[TABLE_SIZE];
 	uint key;
 	uint local_publ_count;
 };
@@ -576,15 +574,22 @@
 
 static struct name_seq* nametbl_find_seq(uint type)
 {
-	struct name_seq* seq;
-	dbg("find_seq %u,(%u,0x%x) table = %x, hash[type] = %u\n",
-	       type,ntohl(type),type,table,hash(type));
-	seq = table->types[hash(type)];
-	dbg("found %x\n",seq);
-	while (seq && (seq->type != type)) {
-		seq = seq->next;
+	struct hlist_head *seq_head;
+	struct hlist_node *seq_node;
+	struct name_seq *ns;
+
+	dbg("find_seq %u,(%u,0x%x) table = %p, hash[type] = %u\n",
+	       type, ntohl(type), type, table, hash(type));
+
+	seq_head = &table->types[hash(type)];
+	hlist_for_each_entry(ns, seq_node, seq_head, ns_list) {
+		if (ns->type == type) {
+			dbg("found %x\n", ns);
+			return ns;
+		}
 	}
-	return seq;
+
+	return 0;
 };
 
 
@@ -606,8 +611,7 @@
 	}
 	dbg("Publishing <%u,%u,%u> from %x\n", type, lower, upper, node);
 	if (!seq) {
-		struct name_seq *head = table->types[hash(type)];
-		seq = table->types[hash(type)] = nameseq_create(type, head);
+		seq = nameseq_create(type, &table->types[hash(type)]);
 		dbg("nametbl_insert_publ: created %x\n",seq);
 	}
 	assert(seq->type == type);
@@ -626,13 +630,9 @@
 		return 0;
 	dbg("Withdrawing <%u,%u> from %x\n", type, lower, node);
 	publ = nameseq_remove_publ(seq, lower, node, key);
-	if (!seq->first_free && !seq->subscriptions) {
-		if (seq->prev)
-			seq->prev->next = seq->next;
-		else
-			table->types[hash(seq->type)] = seq->next;
-		if (seq->next)
-			seq->next->prev = seq->prev;
+
+	if (!seq->first_free && list_empty(&seq->subscriptions)) {
+		hlist_del_init(&seq->ns_list);
                 kfree(seq->sseqs);
                 kfree(seq);
 	}
@@ -778,7 +778,6 @@
 		   spin_unlock_bh(&seq->lock);
 		   goto not_found;
   }
-  found:
    spin_unlock_bh(&seq->lock);
    read_unlock_bh(&nametbl_lock); 
    return true;
@@ -797,7 +796,6 @@
    struct name_seq* seq; 
    int i = 0; 
    struct  publication* publ;
-   struct publication *publhead;
    struct sub_seq *sseq;   
    int low_seq, high_seq;
    uint destport;
@@ -921,8 +919,7 @@
 	struct name_seq *seq;
         seq = nametbl_find_seq(type);
         if (!seq) {
-                struct name_seq *head = table->types[hash(type)];
-                seq = table->types[hash(type)] = nameseq_create(type, head);
+                seq = nameseq_create(type, &table->types[hash(type)]);
         }
         spin_lock_bh(&seq->lock); 
         dbg("nametbl_subscribe:found %x for <%u,%u,%u>\n",
@@ -936,22 +933,12 @@
 nametbl_unsubscribe(struct name_subscr *s)
 {
 	uint type = s->publ.s.seq.type;
-	struct name_subscr *prev = 0;
 	struct name_subscr *crs;
 	struct name_seq *seq = nametbl_find_seq(type);
         assert(seq);
+
 	spin_lock_bh(&seq->lock); 
-	crs = seq->subscriptions;
-	while ((crs != 0) && (crs != s)) {
-		prev = crs;
-		crs = crs->next;
-	}
-	if (crs){
-		if (!prev)
-			seq->subscriptions = crs->next;
-		else
-			prev->next = crs->next;
-	}
+	list_del_init(&s->nsub_list);
 	spin_unlock_bh(&seq->lock); 
 }
 
@@ -1026,13 +1013,15 @@
 static void
 nametbl_list(struct print_buf *buf, uint type, uint depth)
 {
-	struct name_seq *seq = 0;
+	struct hlist_head *seq_head;
+	struct hlist_node *seq_node;
+	struct name_seq *seq;
+
 	uint i;
 	for (i = 0; i < TABLE_SIZE; i++) {
-                seq = table->types[i];
-                while (seq) {
-                        nameseq_list(seq,buf,type,depth,i);
-                        seq = seq->next;
+                seq_head = &table->types[i];
+		hlist_for_each_entry(seq, seq_node, seq_head, ns_list) {
+                        nameseq_list(seq, buf, type, depth, i);
                 }
         }
 }
@@ -1081,11 +1070,15 @@
 nametbl_stop(void)
 {
 	uint i;
+	struct hlist_head *seq_head;
+	struct hlist_node *seq_node, *tmp;
+	struct name_seq *seq;
+
 	write_lock_bh(&nametbl_lock);
 	for (i = 0; i < TABLE_SIZE; i++) {
-		struct name_seq *seq = table->types[i];
-                while (seq) {
-			struct name_seq *next_seq = seq->next;
+		seq_head = &table->types[i];
+		hlist_for_each_entry_safe(seq, seq_node, 
+						tmp, seq_head, ns_list) {
 			struct sub_seq *sseq = seq->sseqs;
 
 			for(;sseq != &seq->sseqs[seq->first_free];sseq++){
@@ -1098,9 +1091,9 @@
 				}
 				while(publ != sseq->zone_list);
 			}
-                        seq = next_seq;
                 }
         }
 	kfree(table);
+	table = 0;
 	write_unlock_bh(&nametbl_lock);
 }
Index: name_table.h
===================================================================
RCS file: /cvsroot/tipc/source/unstable/net/tipc/name_table.h,v
retrieving revision 1.2
diff -u -r1.2 name_table.h
--- name_table.h	3 Feb 2004 23:27:35 -0000	1.2
+++ name_table.h	4 May 2004 22:07:46 -0000
@@ -81,10 +81,7 @@
 	tipc_net_addr_t node;
 	uint key;
 	uint scope;
-        struct {
-                struct publication *prev;
-                struct publication *next;
-        }local_list;
+        struct list_head local_list;
         struct {
                 struct publication *next;
         }port_list;
Index: name_subscr.c
===================================================================
RCS file: /cvsroot/tipc/source/unstable/net/tipc/name_subscr.c,v
retrieving revision 1.6
diff -u -r1.6 name_subscr.c
--- name_subscr.c	11 Mar 2004 01:27:14 -0000	1.6
+++ name_subscr.c	4 May 2004 22:07:46 -0000
@@ -238,6 +238,8 @@
 	}
         this = (struct name_subscr*)k_malloc(sizeof(*this));
         memset(this,0,sizeof(*this));
+	INIT_LIST_HEAD(&this->nsub_list);
+	INIT_LIST_HEAD(&this->reg.reg_list);
 	s->ref = ref_lock_acquire(this,&this->lock);
 	if (!s->ref){
 		warn("Obtained no ref for subsription\n");
Index: port.c
===================================================================
RCS file: /cvsroot/tipc/source/unstable/net/tipc/port.c,v
retrieving revision 1.16
diff -u -r1.16 port.c
--- port.c	30 Apr 2004 21:23:17 -0000	1.16
+++ port.c	4 May 2004 22:07:46 -0000
@@ -935,6 +935,7 @@
 	p->named_msg_cb = named_msg_cb;
 	p->conn_msg_cb = conn_msg_cb;
 	p->continue_event_cb = continue_event_cb;
+	INIT_LIST_HEAD(&p->uport_list);
 	reg_add_port(user_ref,p);
         *portref = this->publ.ref;
         dbg(" tipc_createport: %x with ref %u\n",this,this->publ.ref);        
Index: name_subscr.h
===================================================================
RCS file: /cvsroot/tipc/source/unstable/net/tipc/name_subscr.h,v
retrieving revision 1.3
diff -u -r1.3 name_subscr.h
--- name_subscr.h	16 Feb 2004 23:00:02 -0000	1.3
+++ name_subscr.h	4 May 2004 22:07:46 -0000
@@ -89,14 +89,13 @@
 
 struct name_subscr {
 	struct tipc_name_subscr publ;
-	struct name_subscr *next;
+	struct list_head nsub_list;
 	void (*event_handler) (struct tipc_name_event *);
 	long unsigned int timer_ref;
 	int expired;
         spinlock_t *lock;
         struct {
-                struct name_subscr *next;
-                struct name_subscr *prev;
+		struct list_head reg_list;
                 tipc_ref_t ref;
         }reg;
 };
Index: port.h
===================================================================
RCS file: /cvsroot/tipc/source/unstable/net/tipc/port.h,v
retrieving revision 1.7
diff -u -r1.7 port.h
--- port.h	30 Apr 2004 21:23:17 -0000	1.7
+++ port.h	4 May 2004 22:07:46 -0000
@@ -135,8 +135,7 @@
 	tipc_named_msg_event named_msg_cb; 
 	tipc_conn_msg_event conn_msg_cb; 
 	tipc_continue_event continue_event_cb;
-	struct user_port* next;
-	struct user_port* prev;
+	struct list_head uport_list;
 };
 
 struct port {
Index: name_distr.c
===================================================================
RCS file: /cvsroot/tipc/source/unstable/net/tipc/name_distr.c,v
retrieving revision 1.6
diff -u -r1.6 name_distr.c
--- name_distr.c	23 Apr 2004 15:12:06 -0000	1.6
+++ name_distr.c	4 May 2004 22:07:46 -0000
@@ -77,7 +77,7 @@
 
 #define ITEM_SIZE sizeof(struct distr_item)
 
-static struct publication *publ_root = 0;
+static LIST_HEAD(publ_root);
 static uint publ_cnt = 0;
 
 struct distr_item {
@@ -109,30 +109,33 @@
 
 void named_node_up(tipc_net_addr_t node)
 {
-	uint rest;
 	struct publication* publ;
+	struct distr_item* item = 0;
+	struct sk_buff* buf = 0;
+	uint left = 0;
+	uint rest;
+	uint max_item_buf;
+
 	assert(in_own_cluster(node));
 	read_lock_bh(&nametbl_lock); 
-	publ = publ_root;
+	max_item_buf = TIPC_MAX_MSG_SIZE / ITEM_SIZE;
+	max_item_buf *= ITEM_SIZE;
 	rest =  publ_cnt*ITEM_SIZE;
-	if (publ_cnt) {
-		while (rest > 0){
-			struct sk_buff* buf;
-			struct distr_item* item;
-			uint size = rest <= TIPC_MAX_MSG_SIZE?rest:TIPC_MAX_MSG_SIZE;
-			uint left = size;
-			buf = named_prepare_buf(PUBLICATION,size,node);	
+
+	list_for_each_entry(publ, &publ_root, local_list) {
+		if (!buf) {
+			left = (rest <= max_item_buf) ?rest :max_item_buf;
+			rest -= left;
+			buf = named_prepare_buf(PUBLICATION, left, node);	
 			item = (struct distr_item*)msg_data(buf_msg(buf));
-			do {
-				publ_to_item(item,publ);
-				publ = publ->local_list.next;
-				item++;
-				left -= ITEM_SIZE;
-			}
-			while(left > 0);
-                        msg_set_link_selector(buf_msg(buf),node);
+		}
+		publ_to_item(item,publ);
+		item++;
+		left -= ITEM_SIZE;
+		if (!left) {
+			msg_set_link_selector(buf_msg(buf),node);
 			link_send(buf,node,node);
-			rest-=size;
+			buf = 0;
 		}
 	}
 	read_unlock_bh(&nametbl_lock); 
@@ -142,11 +145,7 @@
 {
 	struct sk_buff* buf = named_prepare_buf(PUBLICATION,ITEM_SIZE,0);
  	struct distr_item *item = (struct distr_item*) msg_data(buf_msg(buf));
-	if (publ_root)
-		publ_root->local_list.prev = publ;
-	publ->local_list.next = publ_root;
-	publ->local_list.prev = 0;
-	publ_root = publ;
+	list_add(&publ->local_list, &publ_root);
 	publ_cnt++;
         publ_to_item(item,publ);
 	cluster_broadcast(buf);
@@ -156,12 +155,7 @@
 {
 	struct sk_buff* buf = named_prepare_buf(WITHDRAWAL,ITEM_SIZE,0);
 	struct distr_item *item = (struct distr_item*) msg_data(buf_msg(buf));
-	if (publ_root == publ)
-		publ_root = publ->local_list.next;	
-	if (publ->local_list.prev)
-		publ->local_list.prev->local_list.next = publ->local_list.next;
-	if (publ->local_list.next)
-		publ->local_list.next->local_list.prev = publ->local_list.prev;
+	list_del_init(&publ->local_list);
 	publ_cnt--;
         publ_to_item(item,publ);
 	cluster_broadcast(buf);
Index: reg.c
===================================================================
RCS file: /cvsroot/tipc/source/unstable/net/tipc/reg.c,v
retrieving revision 1.6
diff -u -r1.6 reg.c
--- reg.c	16 Feb 2004 23:00:02 -0000	1.6
+++ reg.c	4 May 2004 22:07:46 -0000
@@ -182,8 +182,8 @@
 	void *usr_handle;
 	uint next;
 	tipc_started_event callback;
-	struct user_port *ports;
-	struct name_subscr *subs;
+	struct list_head ports;
+	struct list_head subs;
 };
 
 #define MAX_USERID 64
@@ -204,11 +204,7 @@
         spin_lock_bh(&reg_lock);
 	user = &users[ref];
         p->reg_ref = ref;
-        p->prev = 0;
-        p->next = user->ports;
-        if (p->next)
-                p->next->prev = p;
-        user->ports = p;
+	list_add(&p->uport_list, &user->ports);
         spin_unlock_bh(&reg_lock);
         return TIPC_OK;
 }
@@ -224,12 +220,7 @@
                 return TIPC_OK;
         spin_lock_bh(&reg_lock);
 	user = &users[p->reg_ref];
-        if (p->prev)
-                p->prev->next = p->next;
-        else 
-                user->ports = p->next;
-        if (p->next)
-                p->next->prev = p->prev;
+	list_del_init(&p->uport_list);
         spin_unlock_bh(&reg_lock);
         return TIPC_OK;
 }
@@ -247,18 +238,13 @@
         spin_lock_bh(&reg_lock);
 	user = &users[s->reg.ref];
         s->reg.ref = ref;
-        s->reg.prev = 0;
-        s->reg.next = user->subs;
-        if (s->reg.next)
-                s->reg.next->reg.prev = s;
-        user->subs = s;
+	list_add(&s->reg.reg_list, &user->subs);
         spin_unlock_bh(&reg_lock);
         return TIPC_OK;
 }
 
 int reg_remove_subscr(struct name_subscr* s)
 {
-	struct tipc_user *user;
 	if (!tipc_started || !users)
       		return TIPC_FAILURE;
 	if (s->reg.ref > MAX_USERID)
@@ -266,27 +252,24 @@
 	if (s->reg.ref == 0)
                 return TIPC_OK;
         spin_lock_bh(&reg_lock);
-	user = &users[s->reg.ref];
-        if (s->reg.prev)
-                s->reg.prev->reg.next = s->reg.next;
-        else 
-                user->subs = s->reg.next;
-        if (s->next)
-                s->reg.next->reg.prev = s->reg.prev;
+	list_del_init(&s->reg.reg_list);
         spin_unlock_bh(&reg_lock);
         return TIPC_OK;
 }
 
 static void reg_init(void)
 {
-	uint i = 1;
+	uint i;
+
 	if (users)
 		return;
         spin_lock_bh(&reg_lock);
         users = (struct tipc_user*)k_malloc(USER_LIST_SIZE);
-	memset(users, 0,USER_LIST_SIZE);
+	memset(users, 0, USER_LIST_SIZE);
 	for (i = 1; i <= MAX_USERID; i++) {
 		users[i].next = i - 1;
+		INIT_LIST_HEAD(&users[i].subs);
+		INIT_LIST_HEAD(&users[i].ports);
 	}
 	next_free_user = MAX_USERID;
         spin_unlock_bh(&reg_lock);
@@ -324,7 +307,7 @@
 reg_stop(void)
 {		
 	uint id;
-	for (id = 0; id <= MAX_USERID; id++) {
+	for (id = 1; id <= MAX_USERID; id++) {
 		if (users[id].next == 0)
 			tipc_detach(id);
 	}
@@ -359,24 +342,20 @@
 void
 tipc_detach(tipc_ref_t userid)
 {
-        struct name_subscr* s = 0;
-        struct user_port* p = 0;
+        struct name_subscr *s, *st;
+        struct user_port *p, *pt;
 	struct tipc_user *user;
         spin_lock_bh(&reg_lock);
         user = &users[userid];        
 	if (!user->next){
-                s = user->subs;
-                p = user->ports;
-                user->subs = 0;
-                user->ports = 0;
                 user->next = next_free_user;
                 next_free_user = userid;
-                spin_unlock_bh(&reg_lock);
         }
-        while (s) {
+	spin_unlock_bh(&reg_lock);
+	list_for_each_entry_safe(s, st, &user->subs, reg.reg_list) {
                 tipc_unsubscribe(s->publ.s.ref);
         }
-        while (user->ports) {
+	list_for_each_entry_safe(p, pt, &user->ports, uport_list) {
                 tipc_deleteport(p->ref);
         }
 }

-- 
Mark Haverkamp <ma...@os...>





Thread: [Tipc-discussion] More tipc list updates

Cluster wide IPC providing datagram, connection, and bus messaging

tipc-discussion