summaryrefslogtreecommitdiff
path: root/net/tipc
diff options
context:
space:
mode:
authorJon Maloy <jon.maloy@ericsson.com>2017-10-13 11:04:30 +0200
committerDavid S. Miller <davem@davemloft.net>2017-10-13 08:46:01 -0700
commitb87a5ea31c935a7f7e11ca85df2ec7917921e96d (patch)
tree6d8d1235da6271bea5506b6564205583a4772305 /net/tipc
parent5b8dddb63769587badc50725ec9857caaeba4de0 (diff)
tipc: guarantee group unicast doesn't bypass group broadcast
Group unicast messages don't follow the same path as broadcast messages, and there is a high risk that unicasts sent from a socket might bypass previously sent broadcasts from the same socket. We fix this by letting all unicast messages carry the sequence number of the next sent broadcast from the same node, but without updating this number at the receiver. This way, a receiver can check and if necessary re-order such messages before they are added to the socket receive buffer. Signed-off-by: Jon Maloy <jon.maloy@ericsson.com> Acked-by: Ying Xue <ying.xue@windriver.com> Signed-off-by: David S. Miller <davem@davemloft.net>
Diffstat (limited to 'net/tipc')
-rw-r--r--net/tipc/group.c87
-rw-r--r--net/tipc/socket.c2
2 files changed, 74 insertions, 15 deletions
diff --git a/net/tipc/group.c b/net/tipc/group.c
index ffac2f33fce2..985e0ce32e8e 100644
--- a/net/tipc/group.c
+++ b/net/tipc/group.c
@@ -62,6 +62,7 @@ struct tipc_member {
struct list_head list;
struct list_head congested;
struct sk_buff *event_msg;
+ struct sk_buff_head deferredq;
struct tipc_group *group;
u32 node;
u32 port;
@@ -253,6 +254,7 @@ static struct tipc_member *tipc_group_create_member(struct tipc_group *grp,
return NULL;
INIT_LIST_HEAD(&m->list);
INIT_LIST_HEAD(&m->congested);
+ __skb_queue_head_init(&m->deferredq);
m->group = grp;
m->node = node;
m->port = port;
@@ -380,29 +382,54 @@ bool tipc_group_bc_cong(struct tipc_group *grp, int len)
return tipc_group_cong(grp, m->node, m->port, len, &m);
}
+/* tipc_group_sort_msg() - sort msg into queue by bcast sequence number
+ */
+static void tipc_group_sort_msg(struct sk_buff *skb, struct sk_buff_head *defq)
+{
+ struct tipc_msg *_hdr, *hdr = buf_msg(skb);
+ u16 bc_seqno = msg_grp_bc_seqno(hdr);
+ struct sk_buff *_skb, *tmp;
+ int mtyp = msg_type(hdr);
+
+ /* Bcast may be bypassed by unicast, - sort it in */
+ if (mtyp == TIPC_GRP_BCAST_MSG || mtyp == TIPC_GRP_MCAST_MSG) {
+ skb_queue_walk_safe(defq, _skb, tmp) {
+ _hdr = buf_msg(_skb);
+ if (!less(bc_seqno, msg_grp_bc_seqno(_hdr)))
+ continue;
+ __skb_queue_before(defq, _skb, skb);
+ return;
+ }
+ /* Bcast was not bypassed, - add to tail */
+ }
+ /* Unicasts are never bypassed, - always add to tail */
+ __skb_queue_tail(defq, skb);
+}
+
/* tipc_group_filter_msg() - determine if we should accept arriving message
*/
void tipc_group_filter_msg(struct tipc_group *grp, struct sk_buff_head *inputq,
struct sk_buff_head *xmitq)
{
struct sk_buff *skb = __skb_dequeue(inputq);
+ struct sk_buff_head *defq;
struct tipc_member *m;
struct tipc_msg *hdr;
+ bool deliver, update;
u32 node, port;
- int mtyp;
+ int mtyp, blks;
if (!skb)
return;
hdr = buf_msg(skb);
- mtyp = msg_type(hdr);
node = msg_orignode(hdr);
port = msg_origport(hdr);
if (!msg_in_group(hdr))
goto drop;
- if (mtyp == TIPC_GRP_MEMBER_EVT) {
+ if (msg_is_grp_evt(hdr)) {
if (!grp->events)
goto drop;
__skb_queue_tail(inputq, skb);
@@ -413,22 +440,52 @@ void tipc_group_filter_msg(struct tipc_group *grp, struct sk_buff_head *inputq,
if (!tipc_group_is_receiver(m))
goto drop;
- m->bc_rcv_nxt = msg_grp_bc_seqno(hdr) + 1;
+ if (less(msg_grp_bc_seqno(hdr), m->bc_rcv_nxt))
+ goto drop;
- /* Drop multicast here if not for this member */
- if (mtyp == TIPC_GRP_MCAST_MSG) {
- if (msg_nameinst(hdr) != grp->instance) {
- m->bc_rcv_nxt = msg_grp_bc_seqno(hdr) + 1;
- tipc_group_update_rcv_win(grp, msg_blocks(hdr),
- node, port, xmitq);
- kfree_skb(skb);
- return;
+ TIPC_SKB_CB(skb)->orig_member = m->instance;
+ defq = &m->deferredq;
+ tipc_group_sort_msg(skb, defq);
+
+ while ((skb = skb_peek(defq))) {
+ hdr = buf_msg(skb);
+ mtyp = msg_type(hdr);
+ deliver = true;
+ update = false;
+
+ if (more(msg_grp_bc_seqno(hdr), m->bc_rcv_nxt))
+ break;
+
+ /* Decide what to do with message */
+ switch (mtyp) {
+ case TIPC_GRP_MCAST_MSG:
+ if (msg_nameinst(hdr) != grp->instance) {
+ update = true;
+ deliver = false;
+ }
+ /* Fall thru */
+ case TIPC_GRP_BCAST_MSG:
+ m->bc_rcv_nxt++;
+ break;
+ case TIPC_GRP_UCAST_MSG:
+ break;
+ default:
+ break;
}
- }
- TIPC_SKB_CB(skb)->orig_member = m->instance;
- __skb_queue_tail(inputq, skb);
+ /* Execute decisions */
+ __skb_dequeue(defq);
+ if (deliver)
+ __skb_queue_tail(inputq, skb);
+ else
+ kfree_skb(skb);
+
+ if (!update)
+ continue;
+ blks = msg_blocks(hdr);
+ tipc_group_update_rcv_win(grp, blks, node, port, xmitq);
+ }
return;
drop:
kfree_skb(skb);
diff --git a/net/tipc/socket.c b/net/tipc/socket.c
index 8fdd969e12bd..3276b7a0d445 100644
--- a/net/tipc/socket.c
+++ b/net/tipc/socket.c
@@ -830,6 +830,7 @@ static int tipc_send_group_msg(struct net *net, struct tipc_sock *tsk,
struct msghdr *m, struct tipc_member *mb,
u32 dnode, u32 dport, int dlen)
{
+ u16 bc_snd_nxt = tipc_group_bc_snd_nxt(tsk->group);
int blks = tsk_blocks(GROUP_H_SIZE + dlen);
struct tipc_msg *hdr = &tsk->phdr;
struct sk_buff_head pkts;
@@ -840,6 +841,7 @@ static int tipc_send_group_msg(struct net *net, struct tipc_sock *tsk,
msg_set_hdr_sz(hdr, GROUP_H_SIZE);
msg_set_destport(hdr, dport);
msg_set_destnode(hdr, dnode);
+ msg_set_grp_bc_seqno(hdr, bc_snd_nxt);
/* Build message as chain of buffers */
skb_queue_head_init(&pkts);