Commit 232d07b7 authored by Jon Maloy's avatar Jon Maloy Committed by David S. Miller
Browse files

tipc: improve groupcast scope handling 



When a member joins a group, it also indicates a binding scope. This
makes it possible to create both node local groups, invisible to other
nodes, as well as cluster global groups, visible everywhere.

In order to avoid that different members end up having permanently
differing views of group size and memberhip, we must inhibit locally
and globally bound members from joining the same group.

We do this by using the binding scope as an additional separator between
groups. I.e., a member must ignore all membership events from sockets
using a different scope than itself, and all lookups for message
destinations must require an exact match between the message's lookup
scope and the potential target's binding scope.

Apart from making it possible to create local groups using the same
identity on different nodes, a side effect of this is that it now also
becomes possible to create a cluster global group with the same identity
across the same nodes, without interfering with the local groups.

Acked-by: default avatarYing Xue <ying.xue@windriver.com>
Signed-off-by: default avatarJon Maloy <jon.maloy@ericsson.com>
Signed-off-by: default avatarDavid S. Miller <davem@davemloft.net>
parent 8348500f

include/uapi/linux/tipc.h

+3 −4
Original line number Diff line number Diff line
@@ -117,7 +117,6 @@ static inline unsigned int tipc_node(__u32 addr) 
/*

 * Publication scopes when binding port names and port name sequences

 */



#define TIPC_ZONE_SCOPE         1

#define TIPC_CLUSTER_SCOPE      2

#define TIPC_NODE_SCOPE         3

net/tipc/group.c

+8 −5
Original line number Diff line number Diff line
@@ -87,7 +87,6 @@ struct tipc_group { 
	int subid;

	u32 type;

	u32 instance;

	u32 domain;

	u32 scope;

	u32 portid;

	u16 member_cnt;
@@ -158,6 +157,8 @@ int tipc_group_size(struct tipc_group *grp) 
struct tipc_group *tipc_group_create(struct net *net, u32 portid,

				     struct tipc_group_req *mreq)

{

	u32 filter = TIPC_SUB_PORTS | TIPC_SUB_NO_STATUS;

	bool global = mreq->scope != TIPC_NODE_SCOPE;

	struct tipc_group *grp;

	u32 type = mreq->type;
@@ -171,15 +172,14 @@ struct tipc_group *tipc_group_create(struct net *net, u32 portid, 
	grp->members = RB_ROOT;

	grp->net = net;

	grp->portid = portid;

	grp->domain = addr_domain(net, mreq->scope);

	grp->type = type;

	grp->instance = mreq->instance;

	grp->scope = mreq->scope;

	grp->loopback = mreq->flags & TIPC_GROUP_LOOPBACK;

	grp->events = mreq->flags & TIPC_GROUP_MEMBER_EVTS;

	if (tipc_topsrv_kern_subscr(net, portid, type,

				    TIPC_SUB_PORTS | TIPC_SUB_NO_STATUS,

				    0, ~0, &grp->subid))

	filter |= global ? TIPC_SUB_CLUSTER_SCOPE : TIPC_SUB_NODE_SCOPE;

	if (tipc_topsrv_kern_subscr(net, portid, type, 0, ~0,

				    filter, &grp->subid))

		return grp;

	kfree(grp);

	return NULL;
@@ -732,6 +732,9 @@ void tipc_group_proto_rcv(struct tipc_group *grp, bool *usr_wakeup, 
	if (!grp)

		return;



	if (grp->scope == TIPC_NODE_SCOPE && node != tipc_own_addr(grp->net))

		return;



	m = tipc_group_find_member(grp, node, port);



	switch (msg_type(hdr)) {

net/tipc/name_table.c

+19 −21
Original line number Diff line number Diff line
@@ -328,7 +328,8 @@ static struct publication *tipc_nameseq_insert_publ(struct net *net, 
	list_for_each_entry_safe(s, st, &nseq->subscriptions, nameseq_list) {

		tipc_subscrp_report_overlap(s, publ->lower, publ->upper,

					    TIPC_PUBLISHED, publ->ref,

					    publ->node, created_subseq);

					    publ->node, publ->scope,

					    created_subseq);

	}

	return publ;

}
@@ -398,7 +399,8 @@ found: 
	list_for_each_entry_safe(s, st, &nseq->subscriptions, nameseq_list) {

		tipc_subscrp_report_overlap(s, publ->lower, publ->upper,

					    TIPC_WITHDRAWN, publ->ref,

					    publ->node, removed_subseq);

					    publ->node, publ->scope,

					    removed_subseq);

	}



	return publ;
@@ -435,6 +437,7 @@ static void tipc_nameseq_subscribe(struct name_seq *nseq, 
							    sseq->upper,

							    TIPC_PUBLISHED,

							    crs->ref, crs->node,

							    crs->scope,

							    must_report);

				must_report = 0;

			}
@@ -598,7 +601,7 @@ not_found: 
	return ref;

}



bool tipc_nametbl_lookup(struct net *net, u32 type, u32 instance, u32 domain,

bool tipc_nametbl_lookup(struct net *net, u32 type, u32 instance, u32 scope,

			 struct list_head *dsts, int *dstcnt, u32 exclude,

			 bool all)

{
@@ -608,9 +611,6 @@ bool tipc_nametbl_lookup(struct net *net, u32 type, u32 instance, u32 domain, 
	struct name_seq *seq;

	struct sub_seq *sseq;



	if (!tipc_in_scope(domain, self))

		return false;



	*dstcnt = 0;

	rcu_read_lock();

	seq = nametbl_find_seq(net, type);
@@ -621,7 +621,7 @@ bool tipc_nametbl_lookup(struct net *net, u32 type, u32 instance, u32 domain, 
	if (likely(sseq)) {

		info = sseq->info;

		list_for_each_entry(publ, &info->zone_list, zone_list) {

			if (!tipc_in_scope(domain, publ->node))

			if (publ->scope != scope)

				continue;

			if (publ->ref == exclude && publ->node == self)

				continue;
@@ -639,13 +639,14 @@ exit: 
	return !list_empty(dsts);

}



int tipc_nametbl_mc_translate(struct net *net, u32 type, u32 lower, u32 upper,

			      u32 limit, struct list_head *dports)

int tipc_nametbl_mc_lookup(struct net *net, u32 type, u32 lower, u32 upper,

			   u32 scope, bool exact, struct list_head *dports)

{

	struct name_seq *seq;

	struct sub_seq *sseq;

	struct sub_seq *sseq_stop;

	struct name_info *info;

	struct publication *p;

	struct name_seq *seq;

	struct sub_seq *sseq;

	int res = 0;



	rcu_read_lock();
@@ -657,15 +658,12 @@ int tipc_nametbl_mc_translate(struct net *net, u32 type, u32 lower, u32 upper, 
	sseq = seq->sseqs + nameseq_locate_subseq(seq, lower);

	sseq_stop = seq->sseqs + seq->first_free;

	for (; sseq != sseq_stop; sseq++) {

		struct publication *publ;



		if (sseq->lower > upper)

			break;



		info = sseq->info;

		list_for_each_entry(publ, &info->node_list, node_list) {

			if (publ->scope <= limit)

				tipc_dest_push(dports, 0, publ->ref);

		list_for_each_entry(p, &info->node_list, node_list) {

			if (p->scope == scope || (!exact && p->scope < scope))

				tipc_dest_push(dports, 0, p->ref);

		}



		if (info->cluster_list_size != info->node_list_size)
@@ -682,7 +680,7 @@ exit: 
 * - Determines if any node local ports overlap

 */

void tipc_nametbl_lookup_dst_nodes(struct net *net, u32 type, u32 lower,

				   u32 upper, u32 domain,

				   u32 upper, u32 scope,

				   struct tipc_nlist *nodes)

{

	struct sub_seq *sseq, *stop;
@@ -701,7 +699,7 @@ void tipc_nametbl_lookup_dst_nodes(struct net *net, u32 type, u32 lower, 
	for (; sseq != stop && sseq->lower <= upper; sseq++) {

		info = sseq->info;

		list_for_each_entry(publ, &info->zone_list, zone_list) {

			if (tipc_in_scope(domain, publ->node))

			if (publ->scope == scope)

				tipc_nlist_add(nodes, publ->node);

		}

	}
@@ -713,7 +711,7 @@ exit: 
/* tipc_nametbl_build_group - build list of communication group members

 */

void tipc_nametbl_build_group(struct net *net, struct tipc_group *grp,

			      u32 type, u32 domain)

			      u32 type, u32 scope)

{

	struct sub_seq *sseq, *stop;

	struct name_info *info;
@@ -731,7 +729,7 @@ void tipc_nametbl_build_group(struct net *net, struct tipc_group *grp, 
	for (; sseq != stop; sseq++) {

		info = sseq->info;

		list_for_each_entry(p, &info->zone_list, zone_list) {

			if (!tipc_in_scope(domain, p->node))

			if (p->scope != scope)

				continue;

			tipc_group_add_member(grp, p->node, p->ref, p->lower);

		}

net/tipc/name_table.h

+2 −2
Original line number Diff line number Diff line
@@ -100,8 +100,8 @@ struct name_table { 
int tipc_nl_name_table_dump(struct sk_buff *skb, struct netlink_callback *cb);



u32 tipc_nametbl_translate(struct net *net, u32 type, u32 instance, u32 *node);

int tipc_nametbl_mc_translate(struct net *net, u32 type, u32 lower, u32 upper,

			      u32 limit, struct list_head *dports);

int tipc_nametbl_mc_lookup(struct net *net, u32 type, u32 lower, u32 upper,

			   u32 scope, bool exact, struct list_head *dports);

void tipc_nametbl_build_group(struct net *net, struct tipc_group *grp,

			      u32 type, u32 domain);

void tipc_nametbl_lookup_dst_nodes(struct net *net, u32 type, u32 lower,

net/tipc/server.c

+2 −2
Original line number Diff line number Diff line
@@ -489,8 +489,8 @@ void tipc_conn_terminate(struct tipc_server *s, int conid) 
	}

}



bool tipc_topsrv_kern_subscr(struct net *net, u32 port, u32 type,

			     u32 filter, u32 lower, u32 upper, int *conid)

bool tipc_topsrv_kern_subscr(struct net *net, u32 port, u32 type, u32 lower,

			     u32 upper, u32 filter, int *conid)

{

	struct tipc_subscriber *scbr;

	struct tipc_subscr sub;

CRA Git | Maintained and supported by SUSTech CRA and CCSE