Commit 7ad32bcb authored by Jon Maloy's avatar Jon Maloy Committed by David S. Miller
Browse files

tipc: create group member event messages when they are needed 



In the current implementation, a group socket receiving topology
events about other members just converts the topology event message
into a group event message and stores it until it reaches the right
state to issue it to the user. This complicates the code unnecessarily,
and becomes impractical when we in the coming commits will need to
create and issue membership events independently.

In this commit, we change this so that we just notice the type and
origin of the incoming topology event, and then drop the buffer. Only
when it is time to actually send a group event to the user do we
explicitly create a new message and send it upwards.

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 0233493a

net/tipc/group.c

+54 −41
Original line number Original line Diff line number Diff line
@@ -64,7 +64,6 @@ struct tipc_member { 
	struct rb_node tree_node;

	struct rb_node tree_node;

	struct list_head list;

	struct list_head list;

	struct list_head small_win;

	struct list_head small_win;

	struct sk_buff *event_msg;

	struct sk_buff_head deferredq;

	struct sk_buff_head deferredq;

	struct tipc_group *group;

	struct tipc_group *group;

	u32 node;

	u32 node;
@@ -632,6 +631,40 @@ void tipc_group_update_rcv_win(struct tipc_group *grp, int blks, u32 node, 
	}

	}

}

}





static void tipc_group_create_event(struct tipc_group *grp,

				    struct tipc_member *m,

				    u32 event, u16 seqno,

				    struct sk_buff_head *inputq)

{	u32 dnode = tipc_own_addr(grp->net);

	struct tipc_event evt;

	struct sk_buff *skb;

	struct tipc_msg *hdr;



	evt.event = event;

	evt.found_lower = m->instance;

	evt.found_upper = m->instance;

	evt.port.ref = m->port;

	evt.port.node = m->node;

	evt.s.seq.type = grp->type;

	evt.s.seq.lower = m->instance;

	evt.s.seq.upper = m->instance;



	skb = tipc_msg_create(TIPC_CRITICAL_IMPORTANCE, TIPC_GRP_MEMBER_EVT,

			      GROUP_H_SIZE, sizeof(evt), dnode, m->node,

			      grp->portid, m->port, 0);

	if (!skb)

		return;



	hdr = buf_msg(skb);

	msg_set_nametype(hdr, grp->type);

	msg_set_grp_evt(hdr, event);

	msg_set_dest_droppable(hdr, true);

	msg_set_grp_bc_seqno(hdr, seqno);

	memcpy(msg_data(hdr), &evt, sizeof(evt));

	TIPC_SKB_CB(skb)->orig_member = m->instance;

	__skb_queue_tail(inputq, skb);

}



static void tipc_group_proto_xmit(struct tipc_group *grp, struct tipc_member *m,

static void tipc_group_proto_xmit(struct tipc_group *grp, struct tipc_member *m,

				  int mtyp, struct sk_buff_head *xmitq)

				  int mtyp, struct sk_buff_head *xmitq)

{

{
@@ -677,7 +710,6 @@ void tipc_group_proto_rcv(struct tipc_group *grp, bool *usr_wakeup, 
	u32 node = msg_orignode(hdr);

	u32 node = msg_orignode(hdr);

	u32 port = msg_origport(hdr);

	u32 port = msg_origport(hdr);

	struct tipc_member *m, *pm;

	struct tipc_member *m, *pm;

	struct tipc_msg *ehdr;

	u16 remitted, in_flight;

	u16 remitted, in_flight;





	if (!grp)

	if (!grp)
@@ -704,9 +736,8 @@ void tipc_group_proto_rcv(struct tipc_group *grp, bool *usr_wakeup, 
			*usr_wakeup = true;

			*usr_wakeup = true;

			m->usr_pending = false;

			m->usr_pending = false;

			tipc_group_proto_xmit(grp, m, GRP_ADV_MSG, xmitq);

			tipc_group_proto_xmit(grp, m, GRP_ADV_MSG, xmitq);

			ehdr = buf_msg(m->event_msg);

			tipc_group_create_event(grp, m, TIPC_PUBLISHED,

			msg_set_grp_bc_seqno(ehdr, m->bc_syncpt);

						m->bc_syncpt, inputq);

			__skb_queue_tail(inputq, m->event_msg);

		}

		}

		list_del_init(&m->small_win);

		list_del_init(&m->small_win);

		tipc_group_update_member(m, 0);

		tipc_group_update_member(m, 0);
@@ -725,10 +756,9 @@ void tipc_group_proto_rcv(struct tipc_group *grp, bool *usr_wakeup, 
			m->state = MBR_LEAVING;

			m->state = MBR_LEAVING;

			return;

			return;

		}

		}

		/* Otherwise deliver already received WITHDRAW event */

		/* Otherwise deliver member WITHDRAW event */

		ehdr = buf_msg(m->event_msg);

		tipc_group_create_event(grp, m, TIPC_WITHDRAWN,

		msg_set_grp_bc_seqno(ehdr, m->bc_syncpt);

					m->bc_syncpt, inputq);

		__skb_queue_tail(inputq, m->event_msg);

		return;

		return;

	case GRP_ADV_MSG:

	case GRP_ADV_MSG:

		if (!m)

		if (!m)
@@ -797,11 +827,10 @@ void tipc_group_proto_rcv(struct tipc_group *grp, bool *usr_wakeup, 
void tipc_group_member_evt(struct tipc_group *grp,

void tipc_group_member_evt(struct tipc_group *grp,

			   bool *usr_wakeup,

			   bool *usr_wakeup,

			   int *sk_rcvbuf,

			   int *sk_rcvbuf,

			   struct sk_buff *skb,

			   struct tipc_msg *hdr,

			   struct sk_buff_head *inputq,

			   struct sk_buff_head *inputq,

			   struct sk_buff_head *xmitq)

			   struct sk_buff_head *xmitq)

{

{

	struct tipc_msg *hdr = buf_msg(skb);

	struct tipc_event *evt = (void *)msg_data(hdr);

	struct tipc_event *evt = (void *)msg_data(hdr);

	u32 instance = evt->found_lower;

	u32 instance = evt->found_lower;

	u32 node = evt->port.node;

	u32 node = evt->port.node;
@@ -813,21 +842,12 @@ void tipc_group_member_evt(struct tipc_group *grp, 
	u32 self;

	u32 self;





	if (!grp)

	if (!grp)

		goto drop;

		return;





	net = grp->net;

	net = grp->net;

	self = tipc_own_addr(net);

	self = tipc_own_addr(net);

	if (!grp->loopback && node == self && port == grp->portid)

	if (!grp->loopback && node == self && port == grp->portid)

		goto drop;

		return;



	/* Convert message before delivery to user */

	msg_set_hdr_sz(hdr, GROUP_H_SIZE);

	msg_set_user(hdr, TIPC_CRITICAL_IMPORTANCE);

	msg_set_type(hdr, TIPC_GRP_MEMBER_EVT);

	msg_set_origport(hdr, port);

	msg_set_orignode(hdr, node);

	msg_set_nametype(hdr, grp->type);

	msg_set_grp_evt(hdr, event);





	m = tipc_group_find_member(grp, node, port);

	m = tipc_group_find_member(grp, node, port);
@@ -836,59 +856,52 @@ void tipc_group_member_evt(struct tipc_group *grp, 
			m = tipc_group_create_member(grp, node, port,

			m = tipc_group_create_member(grp, node, port,

						     MBR_DISCOVERED);

						     MBR_DISCOVERED);

		if (!m)

		if (!m)

			goto drop;

			return;



		m->instance = instance;





		/* Hold back event if JOIN message not yet received */

		/* Hold back event if JOIN message not yet received */

		if (m->state == MBR_DISCOVERED) {

		if (m->state == MBR_DISCOVERED) {

			m->event_msg = skb;

			m->state = MBR_PUBLISHED;

			m->state = MBR_PUBLISHED;

		} else {

		} else {

			msg_set_grp_bc_seqno(hdr, m->bc_syncpt);

			tipc_group_create_event(grp, m, TIPC_PUBLISHED,

			__skb_queue_tail(inputq, skb);

						m->bc_syncpt, inputq);

			m->state = MBR_JOINED;

			m->state = MBR_JOINED;

			*usr_wakeup = true;

			*usr_wakeup = true;

			m->usr_pending = false;

			m->usr_pending = false;

		}

		}

		m->instance = instance;

		TIPC_SKB_CB(skb)->orig_member = m->instance;

		tipc_group_proto_xmit(grp, m, GRP_JOIN_MSG, xmitq);

		tipc_group_proto_xmit(grp, m, GRP_JOIN_MSG, xmitq);

		tipc_group_update_member(m, 0);

		tipc_group_update_member(m, 0);

	} else if (event == TIPC_WITHDRAWN) {

	} else if (event == TIPC_WITHDRAWN) {

		if (!m)

		if (!m)

			goto drop;

			return;



		TIPC_SKB_CB(skb)->orig_member = m->instance;





		*usr_wakeup = true;

		*usr_wakeup = true;

		m->usr_pending = false;

		m->usr_pending = false;

		node_up = tipc_node_is_up(net, node);

		node_up = tipc_node_is_up(net, node);

		m->event_msg = NULL;





		if (node_up) {

		if (node_up) {

			/* Hold back event if a LEAVE msg should be expected */

			/* Hold back event if a LEAVE msg should be expected */

			if (m->state != MBR_LEAVING) {

			if (m->state != MBR_LEAVING) {

				m->event_msg = skb;

				tipc_group_decr_active(grp, m);

				tipc_group_decr_active(grp, m);

				m->state = MBR_LEAVING;

				m->state = MBR_LEAVING;

			} else {

			} else {

				msg_set_grp_bc_seqno(hdr, m->bc_syncpt);

				tipc_group_create_event(grp, m, TIPC_WITHDRAWN,

				__skb_queue_tail(inputq, skb);

							m->bc_syncpt, inputq);

			}

			}

		} else {

		} else {

			if (m->state != MBR_LEAVING) {

			if (m->state != MBR_LEAVING) {

				tipc_group_decr_active(grp, m);

				tipc_group_decr_active(grp, m);

				m->state = MBR_LEAVING;

				m->state = MBR_LEAVING;

				msg_set_grp_bc_seqno(hdr, m->bc_rcv_nxt);

				tipc_group_create_event(grp, m, TIPC_WITHDRAWN,

							m->bc_rcv_nxt, inputq);

			} else {

			} else {

				msg_set_grp_bc_seqno(hdr, m->bc_syncpt);

				tipc_group_create_event(grp, m, TIPC_WITHDRAWN,

							m->bc_syncpt, inputq);

			}

			}

			__skb_queue_tail(inputq, skb);

		}

		}

		list_del_init(&m->list);

		list_del_init(&m->list);

		list_del_init(&m->small_win);

		list_del_init(&m->small_win);

	}

	}

	*sk_rcvbuf = tipc_group_rcvbuf_limit(grp);

	*sk_rcvbuf = tipc_group_rcvbuf_limit(grp);

	return;

drop:

	kfree_skb(skb);

}
 
}

net/tipc/group.h

+1 −1
Original line number Original line Diff line number Diff line
@@ -54,7 +54,7 @@ void tipc_group_filter_msg(struct tipc_group *grp, 
			   struct sk_buff_head *inputq,

			   struct sk_buff_head *inputq,

			   struct sk_buff_head *xmitq);

			   struct sk_buff_head *xmitq);

void tipc_group_member_evt(struct tipc_group *grp, bool *wakeup,

void tipc_group_member_evt(struct tipc_group *grp, bool *wakeup,

			   int *sk_rcvbuf, struct sk_buff *skb,

			   int *sk_rcvbuf, struct tipc_msg *hdr,

			   struct sk_buff_head *inputq,

			   struct sk_buff_head *inputq,

			   struct sk_buff_head *xmitq);

			   struct sk_buff_head *xmitq);

void tipc_group_proto_rcv(struct tipc_group *grp, bool *wakeup,

void tipc_group_proto_rcv(struct tipc_group *grp, bool *wakeup,

net/tipc/socket.c

+1 −2
Original line number Original line Diff line number Diff line
@@ -1933,8 +1933,7 @@ static void tipc_sk_proto_rcv(struct sock *sk, 
		break;

		break;

	case TOP_SRV:

	case TOP_SRV:

		tipc_group_member_evt(tsk->group, &wakeup, &sk->sk_rcvbuf,

		tipc_group_member_evt(tsk->group, &wakeup, &sk->sk_rcvbuf,

				      skb, inputq, xmitq);

				      hdr, inputq, xmitq);

		skb = NULL;

		break;

		break;

	default:

	default:

		break;

		break;

CRA Git | Maintained and supported by SUSTech CRA and CCSE