Skip to content

Commit

Permalink
tipc: introduce group anycast messaging
Browse files Browse the repository at this point in the history
In this commit, we make it possible to send connectionless unicast
messages to any member corresponding to the given member identity,
when there is more than one such member. The sender must use a
TIPC_ADDR_NAME address to achieve this effect.

We also perform load balancing between the destinations, i.e., we
primarily select one which has advertised sufficient send window
to not cause a block/EAGAIN delay, if any. This mechanism is
overlayed on the always present round-robin selection.

Anycast messages are subject to the same start synchronization
and flow control mechanism as group broadcast messages.

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>
  • Loading branch information
Jon Maloy authored and David S. Miller committed Oct 13, 2017
1 parent 27bd9ec commit ee106d7
Show file tree
Hide file tree
Showing 5 changed files with 138 additions and 0 deletions.
7 changes: 7 additions & 0 deletions net/tipc/group.c
Original file line number Diff line number Diff line change
Expand Up @@ -116,6 +116,13 @@ static bool tipc_group_is_receiver(struct tipc_member *m)
return m && m->state >= MBR_JOINED;
}

u32 tipc_group_exclude(struct tipc_group *grp)
{
if (!grp->loopback)
return grp->portid;
return 0;
}

int tipc_group_size(struct tipc_group *grp)
{
return grp->member_cnt;
Expand Down
3 changes: 3 additions & 0 deletions net/tipc/group.h
Original file line number Diff line number Diff line change
Expand Up @@ -49,6 +49,7 @@ void tipc_group_add_member(struct tipc_group *grp, u32 node, u32 port);
struct tipc_nlist *tipc_group_dests(struct tipc_group *grp);
void tipc_group_self(struct tipc_group *grp, struct tipc_name_seq *seq,
int *scope);
u32 tipc_group_exclude(struct tipc_group *grp);
void tipc_group_filter_msg(struct tipc_group *grp,
struct sk_buff_head *inputq,
struct sk_buff_head *xmitq);
Expand All @@ -68,5 +69,7 @@ void tipc_group_update_rcv_win(struct tipc_group *grp, int blks, u32 node,
u32 port, struct sk_buff_head *xmitq);
u16 tipc_group_bc_snd_nxt(struct tipc_group *grp);
void tipc_group_update_member(struct tipc_member *m, int len);
struct tipc_member *tipc_group_find_sender(struct tipc_group *grp,
u32 node, u32 port);
int tipc_group_size(struct tipc_group *grp);
#endif
41 changes: 41 additions & 0 deletions net/tipc/name_table.c
Original file line number Diff line number Diff line change
Expand Up @@ -597,6 +597,47 @@ u32 tipc_nametbl_translate(struct net *net, u32 type, u32 instance,
return ref;
}

bool tipc_nametbl_lookup(struct net *net, u32 type, u32 instance, u32 domain,
struct list_head *dsts, int *dstcnt, u32 exclude,
bool all)
{
u32 self = tipc_own_addr(net);
struct publication *publ;
struct name_info *info;
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);
if (unlikely(!seq))
goto exit;
spin_lock_bh(&seq->lock);
sseq = nameseq_find_subseq(seq, instance);
if (likely(sseq)) {
info = sseq->info;
list_for_each_entry(publ, &info->zone_list, zone_list) {
if (!tipc_in_scope(domain, publ->node))
continue;
if (publ->ref == exclude && publ->node == self)
continue;
tipc_dest_push(dsts, publ->node, publ->ref);
(*dstcnt)++;
if (all)
continue;
list_move_tail(&publ->zone_list, &info->zone_list);
break;
}
}
spin_unlock_bh(&seq->lock);
exit:
rcu_read_unlock();
return !list_empty(dsts);
}

int tipc_nametbl_mc_translate(struct net *net, u32 type, u32 lower, u32 upper,
u32 limit, struct list_head *dports)
{
Expand Down
3 changes: 3 additions & 0 deletions net/tipc/name_table.h
Original file line number Diff line number Diff line change
Expand Up @@ -107,6 +107,9 @@ void tipc_nametbl_build_group(struct net *net, struct tipc_group *grp,
void tipc_nametbl_lookup_dst_nodes(struct net *net, u32 type, u32 lower,
u32 upper, u32 domain,
struct tipc_nlist *nodes);
bool tipc_nametbl_lookup(struct net *net, u32 type, u32 instance, u32 domain,
struct list_head *dsts, int *dstcnt, u32 exclude,
bool all);
struct publication *tipc_nametbl_publish(struct net *net, u32 type, u32 lower,
u32 upper, u32 scope, u32 port_ref,
u32 key);
Expand Down
84 changes: 84 additions & 0 deletions net/tipc/socket.c
Original file line number Diff line number Diff line change
Expand Up @@ -904,6 +904,88 @@ static int tipc_send_group_unicast(struct socket *sock, struct msghdr *m,
return rc ? rc : dlen;
}

/**
* tipc_send_group_anycast - send message to any member with given identity
* @sock: socket structure
* @m: message to send
* @dlen: total length of message data
* @timeout: timeout to wait for wakeup
*
* Called from function tipc_sendmsg(), which has done all sanity checks
* Returns the number of bytes sent on success, or errno
*/
static int tipc_send_group_anycast(struct socket *sock, struct msghdr *m,
int dlen, long timeout)
{
DECLARE_SOCKADDR(struct sockaddr_tipc *, dest, m->msg_name);
struct sock *sk = sock->sk;
struct tipc_sock *tsk = tipc_sk(sk);
struct list_head *cong_links = &tsk->cong_links;
int blks = tsk_blocks(GROUP_H_SIZE + dlen);
struct tipc_group *grp = tsk->group;
struct tipc_member *first = NULL;
struct tipc_member *mbr = NULL;
struct net *net = sock_net(sk);
u32 node, port, exclude;
u32 type, inst, domain;
struct list_head dsts;
int lookups = 0;
int dstcnt, rc;
bool cong;

INIT_LIST_HEAD(&dsts);

type = dest->addr.name.name.type;
inst = dest->addr.name.name.instance;
domain = addr_domain(net, dest->scope);
exclude = tipc_group_exclude(grp);

while (++lookups < 4) {
first = NULL;

/* Look for a non-congested destination member, if any */
while (1) {
if (!tipc_nametbl_lookup(net, type, inst, domain, &dsts,
&dstcnt, exclude, false))
return -EHOSTUNREACH;
tipc_dest_pop(&dsts, &node, &port);
cong = tipc_group_cong(grp, node, port, blks, &mbr);
if (!cong)
break;
if (mbr == first)
break;
if (!first)
first = mbr;
}

/* Start over if destination was not in member list */
if (unlikely(!mbr))
continue;

if (likely(!cong && !tipc_dest_find(cong_links, node, 0)))
break;

/* Block or return if destination link or member is congested */
rc = tipc_wait_for_cond(sock, &timeout,
!tipc_dest_find(cong_links, node, 0) &&
!tipc_group_cong(grp, node, port,
blks, &mbr));
if (unlikely(rc))
return rc;

/* Send, unless destination disappeared while waiting */
if (likely(mbr))
break;
}

if (unlikely(lookups >= 4))
return -EHOSTUNREACH;

rc = tipc_send_group_msg(net, tsk, m, mbr, node, port, dlen);

return rc ? rc : dlen;
}

/**
* tipc_send_group_bcast - send message to all members in communication group
* @sk: socket structure
Expand Down Expand Up @@ -1127,6 +1209,8 @@ static int __tipc_sendmsg(struct socket *sock, struct msghdr *m, size_t dlen)
if (grp) {
if (!dest)
return tipc_send_group_bcast(sock, m, dlen, timeout);
if (dest->addrtype == TIPC_ADDR_NAME)
return tipc_send_group_anycast(sock, m, dlen, timeout);
if (dest->addrtype == TIPC_ADDR_ID)
return tipc_send_group_unicast(sock, m, dlen, timeout);
return -EINVAL;
Expand Down

0 comments on commit ee106d7

Please sign in to comment.