aboutsummaryrefslogtreecommitdiff
path: root/net/tipc
diff options
context:
space:
mode:
Diffstat (limited to 'net/tipc')
-rw-r--r--net/tipc/Kconfig8
-rw-r--r--net/tipc/Makefile1
-rw-r--r--net/tipc/bcast.c53
-rw-r--r--net/tipc/bearer.c13
-rw-r--r--net/tipc/bearer.h17
-rw-r--r--net/tipc/discover.c10
-rw-r--r--net/tipc/eth_media.c8
-rw-r--r--net/tipc/ib_media.c2
-rw-r--r--net/tipc/link.c362
-rw-r--r--net/tipc/link.h17
-rw-r--r--net/tipc/msg.c119
-rw-r--r--net/tipc/msg.h93
-rw-r--r--net/tipc/name_distr.c2
-rw-r--r--net/tipc/name_table.c4
-rw-r--r--net/tipc/node.c4
-rw-r--r--net/tipc/node.h6
-rw-r--r--net/tipc/server.c47
-rw-r--r--net/tipc/socket.c269
-rw-r--r--net/tipc/socket.h4
-rw-r--r--net/tipc/subscr.c23
-rw-r--r--net/tipc/udp_media.c446
21 files changed, 941 insertions, 567 deletions
diff --git a/net/tipc/Kconfig b/net/tipc/Kconfig
index 91c8a8e031db..c25a3a149dc4 100644
--- a/net/tipc/Kconfig
+++ b/net/tipc/Kconfig
@@ -26,3 +26,11 @@ config TIPC_MEDIA_IB
help
Saying Y here will enable support for running TIPC on
IP-over-InfiniBand devices.
+config TIPC_MEDIA_UDP
+ bool "IP/UDP media type support"
+ depends on TIPC
+ select NET_UDP_TUNNEL
+ help
+ Saying Y here will enable support for running TIPC over IP/UDP
+ bool
+ default y
diff --git a/net/tipc/Makefile b/net/tipc/Makefile
index 599b1a540d2b..57e460be4692 100644
--- a/net/tipc/Makefile
+++ b/net/tipc/Makefile
@@ -10,5 +10,6 @@ tipc-y += addr.o bcast.o bearer.o \
netlink.o netlink_compat.o node.o socket.o eth_media.o \
server.o socket.o
+tipc-$(CONFIG_TIPC_MEDIA_UDP) += udp_media.o
tipc-$(CONFIG_TIPC_MEDIA_IB) += ib_media.o
tipc-$(CONFIG_SYSCTL) += sysctl.o
diff --git a/net/tipc/bcast.c b/net/tipc/bcast.c
index 3e41704832de..5aff0844d4d3 100644
--- a/net/tipc/bcast.c
+++ b/net/tipc/bcast.c
@@ -135,9 +135,10 @@ static void bclink_set_last_sent(struct net *net)
{
struct tipc_net *tn = net_generic(net, tipc_net_id);
struct tipc_link *bcl = tn->bcl;
+ struct sk_buff *skb = skb_peek(&bcl->backlogq);
- if (bcl->next_out)
- bcl->fsm_msg_cnt = mod(buf_seqno(bcl->next_out) - 1);
+ if (skb)
+ bcl->fsm_msg_cnt = mod(buf_seqno(skb) - 1);
else
bcl->fsm_msg_cnt = mod(bcl->next_out_no - 1);
}
@@ -180,7 +181,7 @@ static void bclink_retransmit_pkt(struct tipc_net *tn, u32 after, u32 to)
struct sk_buff *skb;
struct tipc_link *bcl = tn->bcl;
- skb_queue_walk(&bcl->outqueue, skb) {
+ skb_queue_walk(&bcl->transmq, skb) {
if (more(buf_seqno(skb), after)) {
tipc_link_retransmit(bcl, skb, mod(to - after));
break;
@@ -210,14 +211,17 @@ void tipc_bclink_wakeup_users(struct net *net)
void tipc_bclink_acknowledge(struct tipc_node *n_ptr, u32 acked)
{
struct sk_buff *skb, *tmp;
- struct sk_buff *next;
unsigned int released = 0;
struct net *net = n_ptr->net;
struct tipc_net *tn = net_generic(net, tipc_net_id);
+ if (unlikely(!n_ptr->bclink.recv_permitted))
+ return;
+
tipc_bclink_lock(net);
+
/* Bail out if tx queue is empty (no clean up is required) */
- skb = skb_peek(&tn->bcl->outqueue);
+ skb = skb_peek(&tn->bcl->transmq);
if (!skb)
goto exit;
@@ -244,27 +248,19 @@ void tipc_bclink_acknowledge(struct tipc_node *n_ptr, u32 acked)
}
/* Skip over packets that node has previously acknowledged */
- skb_queue_walk(&tn->bcl->outqueue, skb) {
+ skb_queue_walk(&tn->bcl->transmq, skb) {
if (more(buf_seqno(skb), n_ptr->bclink.acked))
break;
}
/* Update packets that node is now acknowledging */
- skb_queue_walk_from_safe(&tn->bcl->outqueue, skb, tmp) {
+ skb_queue_walk_from_safe(&tn->bcl->transmq, skb, tmp) {
if (more(buf_seqno(skb), acked))
break;
-
- next = tipc_skb_queue_next(&tn->bcl->outqueue, skb);
- if (skb != tn->bcl->next_out) {
- bcbuf_decr_acks(skb);
- } else {
- bcbuf_set_acks(skb, 0);
- tn->bcl->next_out = next;
- bclink_set_last_sent(net);
- }
-
+ bcbuf_decr_acks(skb);
+ bclink_set_last_sent(net);
if (bcbuf_acks(skb) == 0) {
- __skb_unlink(skb, &tn->bcl->outqueue);
+ __skb_unlink(skb, &tn->bcl->transmq);
kfree_skb(skb);
released = 1;
}
@@ -272,7 +268,7 @@ void tipc_bclink_acknowledge(struct tipc_node *n_ptr, u32 acked)
n_ptr->bclink.acked = acked;
/* Try resolving broadcast link congestion, if necessary */
- if (unlikely(tn->bcl->next_out)) {
+ if (unlikely(skb_peek(&tn->bcl->backlogq))) {
tipc_link_push_packets(tn->bcl);
bclink_set_last_sent(net);
}
@@ -319,7 +315,7 @@ void tipc_bclink_update_link_state(struct tipc_node *n_ptr,
buf = tipc_buf_acquire(INT_H_SIZE);
if (buf) {
struct tipc_msg *msg = buf_msg(buf);
- struct sk_buff *skb = skb_peek(&n_ptr->bclink.deferred_queue);
+ struct sk_buff *skb = skb_peek(&n_ptr->bclink.deferdq);
u32 to = skb ? buf_seqno(skb) - 1 : n_ptr->bclink.last_sent;
tipc_msg_init(tn->own_addr, msg, BCAST_PROTOCOL, STATE_MSG,
@@ -387,14 +383,13 @@ int tipc_bclink_xmit(struct net *net, struct sk_buff_head *list)
__skb_queue_purge(list);
return -EHOSTUNREACH;
}
-
/* Broadcast to all nodes */
if (likely(bclink)) {
tipc_bclink_lock(net);
if (likely(bclink->bcast_nodes.count)) {
rc = __tipc_link_xmit(net, bcl, list);
if (likely(!rc)) {
- u32 len = skb_queue_len(&bcl->outqueue);
+ u32 len = skb_queue_len(&bcl->transmq);
bclink_set_last_sent(net);
bcl->stats.queue_sz_counts++;
@@ -559,25 +554,25 @@ receive:
if (node->bclink.last_in == node->bclink.last_sent)
goto unlock;
- if (skb_queue_empty(&node->bclink.deferred_queue)) {
+ if (skb_queue_empty(&node->bclink.deferdq)) {
node->bclink.oos_state = 1;
goto unlock;
}
- msg = buf_msg(skb_peek(&node->bclink.deferred_queue));
+ msg = buf_msg(skb_peek(&node->bclink.deferdq));
seqno = msg_seqno(msg);
next_in = mod(next_in + 1);
if (seqno != next_in)
goto unlock;
/* Take in-sequence message from deferred queue & deliver it */
- buf = __skb_dequeue(&node->bclink.deferred_queue);
+ buf = __skb_dequeue(&node->bclink.deferdq);
goto receive;
}
/* Handle out-of-sequence broadcast message */
if (less(next_in, seqno)) {
- deferred = tipc_link_defer_pkt(&node->bclink.deferred_queue,
+ deferred = tipc_link_defer_pkt(&node->bclink.deferdq,
buf);
bclink_update_last_sent(node, seqno);
buf = NULL;
@@ -634,7 +629,6 @@ static int tipc_bcbearer_send(struct net *net, struct sk_buff *buf,
msg_set_non_seq(msg, 1);
msg_set_mc_netid(msg, tn->net_id);
tn->bcl->stats.sent_info++;
-
if (WARN_ON(!bclink->bcast_nodes.count)) {
dump_stack();
return 0;
@@ -913,8 +907,9 @@ int tipc_bclink_init(struct net *net)
sprintf(bcbearer->media.name, "tipc-broadcast");
spin_lock_init(&bclink->lock);
- __skb_queue_head_init(&bcl->outqueue);
- __skb_queue_head_init(&bcl->deferred_queue);
+ __skb_queue_head_init(&bcl->transmq);
+ __skb_queue_head_init(&bcl->backlogq);
+ __skb_queue_head_init(&bcl->deferdq);
skb_queue_head_init(&bcl->wakeupq);
bcl->next_out_no = 1;
spin_lock_init(&bclink->node.lock);
diff --git a/net/tipc/bearer.c b/net/tipc/bearer.c
index 48852c2dcc03..3613e72e858e 100644
--- a/net/tipc/bearer.c
+++ b/net/tipc/bearer.c
@@ -48,6 +48,9 @@ static struct tipc_media * const media_info_array[] = {
#ifdef CONFIG_TIPC_MEDIA_IB
&ib_media_info,
#endif
+#ifdef CONFIG_TIPC_MEDIA_UDP
+ &udp_media_info,
+#endif
NULL
};
@@ -216,7 +219,8 @@ void tipc_bearer_remove_dest(struct net *net, u32 bearer_id, u32 dest)
* tipc_enable_bearer - enable bearer with the given name
*/
static int tipc_enable_bearer(struct net *net, const char *name,
- u32 disc_domain, u32 priority)
+ u32 disc_domain, u32 priority,
+ struct nlattr *attr[])
{
struct tipc_net *tn = net_generic(net, tipc_net_id);
struct tipc_bearer *b_ptr;
@@ -304,7 +308,7 @@ restart:
strcpy(b_ptr->name, name);
b_ptr->media = m_ptr;
- res = m_ptr->enable_media(net, b_ptr);
+ res = m_ptr->enable_media(net, b_ptr, attr);
if (res) {
pr_warn("Bearer <%s> rejected, enable failure (%d)\n",
name, -res);
@@ -372,7 +376,8 @@ static void bearer_disable(struct net *net, struct tipc_bearer *b_ptr,
kfree_rcu(b_ptr, rcu);
}
-int tipc_enable_l2_media(struct net *net, struct tipc_bearer *b)
+int tipc_enable_l2_media(struct net *net, struct tipc_bearer *b,
+ struct nlattr *attr[])
{
struct net_device *dev;
char *driver_name = strchr((const char *)b->name, ':') + 1;
@@ -791,7 +796,7 @@ int tipc_nl_bearer_enable(struct sk_buff *skb, struct genl_info *info)
}
rtnl_lock();
- err = tipc_enable_bearer(net, bearer, domain, prio);
+ err = tipc_enable_bearer(net, bearer, domain, prio, attrs);
if (err) {
rtnl_unlock();
return err;
diff --git a/net/tipc/bearer.h b/net/tipc/bearer.h
index 6b17795ff8bc..5cad243ee8fc 100644
--- a/net/tipc/bearer.h
+++ b/net/tipc/bearer.h
@@ -41,7 +41,7 @@
#include <net/genetlink.h>
#define MAX_BEARERS 2
-#define MAX_MEDIA 2
+#define MAX_MEDIA 3
#define MAX_NODES 4096
#define WSIZE 32
@@ -50,14 +50,16 @@
* - the field's actual content and length is defined per media
* - remaining unused bytes in the field are set to zero
*/
-#define TIPC_MEDIA_ADDR_SIZE 32
+#define TIPC_MEDIA_INFO_SIZE 32
#define TIPC_MEDIA_TYPE_OFFSET 3
+#define TIPC_MEDIA_ADDR_OFFSET 4
/*
* Identifiers of supported TIPC media types
*/
#define TIPC_MEDIA_TYPE_ETH 1
#define TIPC_MEDIA_TYPE_IB 2
+#define TIPC_MEDIA_TYPE_UDP 3
/**
* struct tipc_node_map - set of node identifiers
@@ -76,7 +78,7 @@ struct tipc_node_map {
* @broadcast: non-zero if address is a broadcast address
*/
struct tipc_media_addr {
- u8 value[TIPC_MEDIA_ADDR_SIZE];
+ u8 value[TIPC_MEDIA_INFO_SIZE];
u8 media_id;
u8 broadcast;
};
@@ -103,7 +105,8 @@ struct tipc_media {
int (*send_msg)(struct net *net, struct sk_buff *buf,
struct tipc_bearer *b_ptr,
struct tipc_media_addr *dest);
- int (*enable_media)(struct net *net, struct tipc_bearer *b_ptr);
+ int (*enable_media)(struct net *net, struct tipc_bearer *b_ptr,
+ struct nlattr *attr[]);
void (*disable_media)(struct tipc_bearer *b_ptr);
int (*addr2str)(struct tipc_media_addr *addr,
char *strbuf,
@@ -182,6 +185,9 @@ extern struct tipc_media eth_media_info;
#ifdef CONFIG_TIPC_MEDIA_IB
extern struct tipc_media ib_media_info;
#endif
+#ifdef CONFIG_TIPC_MEDIA_UDP
+extern struct tipc_media udp_media_info;
+#endif
int tipc_nl_bearer_disable(struct sk_buff *skb, struct genl_info *info);
int tipc_nl_bearer_enable(struct sk_buff *skb, struct genl_info *info);
@@ -196,7 +202,8 @@ int tipc_nl_media_set(struct sk_buff *skb, struct genl_info *info);
int tipc_media_set_priority(const char *name, u32 new_value);
int tipc_media_set_window(const char *name, u32 new_value);
void tipc_media_addr_printf(char *buf, int len, struct tipc_media_addr *a);
-int tipc_enable_l2_media(struct net *net, struct tipc_bearer *b);
+int tipc_enable_l2_media(struct net *net, struct tipc_bearer *b,
+ struct nlattr *attrs[]);
void tipc_disable_l2_media(struct tipc_bearer *b);
int tipc_l2_send_msg(struct net *net, struct sk_buff *buf,
struct tipc_bearer *b, struct tipc_media_addr *dest);
diff --git a/net/tipc/discover.c b/net/tipc/discover.c
index feef3753615d..169f3dd038b9 100644
--- a/net/tipc/discover.c
+++ b/net/tipc/discover.c
@@ -86,9 +86,10 @@ static void tipc_disc_init_msg(struct net *net, struct sk_buff *buf, u32 type,
msg = buf_msg(buf);
tipc_msg_init(tn->own_addr, msg, LINK_CONFIG, type,
- INT_H_SIZE, dest_domain);
+ MAX_H_SIZE, dest_domain);
msg_set_non_seq(msg, 1);
msg_set_node_sig(msg, tn->random);
+ msg_set_node_capabilities(msg, 0);
msg_set_dest_domain(msg, dest_domain);
msg_set_bc_netid(msg, tn->net_id);
b_ptr->media->addr2msg(msg_media_addr(msg), &b_ptr->addr);
@@ -133,6 +134,7 @@ void tipc_disc_rcv(struct net *net, struct sk_buff *buf,
u32 net_id = msg_bc_netid(msg);
u32 mtyp = msg_type(msg);
u32 signature = msg_node_sig(msg);
+ u16 caps = msg_node_capabilities(msg);
bool addr_match = false;
bool sign_match = false;
bool link_up = false;
@@ -167,6 +169,7 @@ void tipc_disc_rcv(struct net *net, struct sk_buff *buf,
if (!node)
return;
tipc_node_lock(node);
+ node->capabilities = caps;
link = node->links[bearer->identity];
/* Prepare to validate requesting node's signature and media address */
@@ -249,7 +252,7 @@ void tipc_disc_rcv(struct net *net, struct sk_buff *buf,
/* Send response, if necessary */
if (respond && (mtyp == DSC_REQ_MSG)) {
- rbuf = tipc_buf_acquire(INT_H_SIZE);
+ rbuf = tipc_buf_acquire(MAX_H_SIZE);
if (rbuf) {
tipc_disc_init_msg(net, rbuf, DSC_RESP_MSG, bearer);
tipc_bearer_send(net, bearer->identity, rbuf, &maddr);
@@ -359,8 +362,7 @@ int tipc_disc_create(struct net *net, struct tipc_bearer *b_ptr,
req = kmalloc(sizeof(*req), GFP_ATOMIC);
if (!req)
return -ENOMEM;
-
- req->buf = tipc_buf_acquire(INT_H_SIZE);
+ req->buf = tipc_buf_acquire(MAX_H_SIZE);
if (!req->buf) {
kfree(req);
return -ENOMEM;
diff --git a/net/tipc/eth_media.c b/net/tipc/eth_media.c
index 5e1426f1751f..f69a2fde9f4a 100644
--- a/net/tipc/eth_media.c
+++ b/net/tipc/eth_media.c
@@ -37,8 +37,6 @@
#include "core.h"
#include "bearer.h"
-#define ETH_ADDR_OFFSET 4 /* MAC addr position inside address field */
-
/* Convert Ethernet address (media address format) to string */
static int tipc_eth_addr2str(struct tipc_media_addr *addr,
char *strbuf, int bufsz)
@@ -53,9 +51,9 @@ static int tipc_eth_addr2str(struct tipc_media_addr *addr,
/* Convert from media address format to discovery message addr format */
static int tipc_eth_addr2msg(char *msg, struct tipc_media_addr *addr)
{
- memset(msg, 0, TIPC_MEDIA_ADDR_SIZE);
+ memset(msg, 0, TIPC_MEDIA_INFO_SIZE);
msg[TIPC_MEDIA_TYPE_OFFSET] = TIPC_MEDIA_TYPE_ETH;
- memcpy(msg + ETH_ADDR_OFFSET, addr->value, ETH_ALEN);
+ memcpy(msg + TIPC_MEDIA_ADDR_OFFSET, addr->value, ETH_ALEN);
return 0;
}
@@ -79,7 +77,7 @@ static int tipc_eth_msg2addr(struct tipc_bearer *b,
char *msg)
{
/* Skip past preamble: */
- msg += ETH_ADDR_OFFSET;
+ msg += TIPC_MEDIA_ADDR_OFFSET;
return tipc_eth_raw2addr(b, addr, msg);
}
diff --git a/net/tipc/ib_media.c b/net/tipc/ib_media.c
index 8522eef9c136..e8c16718e3fa 100644
--- a/net/tipc/ib_media.c
+++ b/net/tipc/ib_media.c
@@ -57,7 +57,7 @@ static int tipc_ib_addr2str(struct tipc_media_addr *a, char *str_buf,
/* Convert from media address format to discovery message addr format */
static int tipc_ib_addr2msg(char *msg, struct tipc_media_addr *addr)
{
- memset(msg, 0, TIPC_MEDIA_ADDR_SIZE);
+ memset(msg, 0, TIPC_MEDIA_INFO_SIZE);
memcpy(msg, addr->value, INFINIBAND_ALEN);
return 0;
}
diff --git a/net/tipc/link.c b/net/tipc/link.c
index 14f09b3cb87c..8c98c4d00ad6 100644
--- a/net/tipc/link.c
+++ b/net/tipc/link.c
@@ -1,7 +1,7 @@
/*
* net/tipc/link.c: TIPC link code
*
- * Copyright (c) 1996-2007, 2012-2014, Ericsson AB
+ * Copyright (c) 1996-2007, 2012-2015, Ericsson AB
* Copyright (c) 2004-2007, 2010-2013, Wind River Systems
* All rights reserved.
*
@@ -35,6 +35,7 @@
*/
#include "core.h"
+#include "subscr.h"
#include "link.h"
#include "bcast.h"
#include "socket.h"
@@ -194,10 +195,10 @@ static void link_timeout(unsigned long data)
tipc_node_lock(l_ptr->owner);
/* update counters used in statistical profiling of send traffic */
- l_ptr->stats.accu_queue_sz += skb_queue_len(&l_ptr->outqueue);
+ l_ptr->stats.accu_queue_sz += skb_queue_len(&l_ptr->transmq);
l_ptr->stats.queue_sz_counts++;
- skb = skb_peek(&l_ptr->outqueue);
+ skb = skb_peek(&l_ptr->transmq);
if (skb) {
struct tipc_msg *msg = buf_msg(skb);
u32 length = msg_size(msg);
@@ -229,7 +230,7 @@ static void link_timeout(unsigned long data)
/* do all other link processing performed on a periodic basis */
link_state_event(l_ptr, TIMEOUT_EVT);
- if (l_ptr->next_out)
+ if (skb_queue_len(&l_ptr->backlogq))
tipc_link_push_packets(l_ptr);
tipc_node_unlock(l_ptr->owner);
@@ -305,16 +306,15 @@ struct tipc_link *tipc_link_create(struct tipc_node *n_ptr,
msg_set_session(msg, (tn->random & 0xffff));
msg_set_bearer_id(msg, b_ptr->identity);
strcpy((char *)msg_data(msg), if_name);
-
- l_ptr->priority = b_ptr->priority;
- tipc_link_set_queue_limits(l_ptr, b_ptr->window);
-
l_ptr->net_plane = b_ptr->net_plane;
link_init_max_pkt(l_ptr);
+ l_ptr->priority = b_ptr->priority;
+ tipc_link_set_queue_limits(l_ptr, b_ptr->window);
l_ptr->next_out_no = 1;
- __skb_queue_head_init(&l_ptr->outqueue);
- __skb_queue_head_init(&l_ptr->deferred_queue);
+ __skb_queue_head_init(&l_ptr->transmq);
+ __skb_queue_head_init(&l_ptr->backlogq);
+ __skb_queue_head_init(&l_ptr->deferdq);
skb_queue_head_init(&l_ptr->wakeupq);
skb_queue_head_init(&l_ptr->inputq);
skb_queue_head_init(&l_ptr->namedq);
@@ -344,6 +344,7 @@ void tipc_link_delete_list(struct net *net, unsigned int bearer_id,
struct tipc_net *tn = net_generic(net, tipc_net_id);
struct tipc_link *link;
struct tipc_node *node;
+ bool del_link;
rcu_read_lock();
list_for_each_entry_rcu(node, &tn->node_list, list) {
@@ -353,12 +354,13 @@ void tipc_link_delete_list(struct net *net, unsigned int bearer_id,
tipc_node_unlock(node);
continue;
}
+ del_link = !tipc_link_is_up(link) && !link->exp_msg_count;
tipc_link_reset(link);
if (del_timer(&link->timer))
tipc_link_put(link);
link->flags |= LINK_STOPPED;
/* Delete link now, or when failover is finished: */
- if (shutting_down || !tipc_node_is_up(node))
+ if (shutting_down || !tipc_node_is_up(node) || del_link)
tipc_link_delete(link);
tipc_node_unlock(node);
}
@@ -398,7 +400,7 @@ static bool link_schedule_user(struct tipc_link *link, u32 oport,
*/
void link_prepare_wakeup(struct tipc_link *link)
{
- uint pend_qsz = skb_queue_len(&link->outqueue);
+ uint pend_qsz = skb_queue_len(&link->backlogq);
struct sk_buff *skb, *tmp;
skb_queue_walk_safe(&link->wakeupq, skb, tmp) {
@@ -428,8 +430,9 @@ void tipc_link_reset_fragments(struct tipc_link *l_ptr)
*/
void tipc_link_purge_queues(struct tipc_link *l_ptr)
{
- __skb_queue_purge(&l_ptr->deferred_queue);
- __skb_queue_purge(&l_ptr->outqueue);
+ __skb_queue_purge(&l_ptr->deferdq);
+ __skb_queue_purge(&l_ptr->transmq);
+ __skb_queue_purge(&l_ptr->backlogq);
tipc_link_reset_fragments(l_ptr);
}
@@ -462,15 +465,15 @@ void tipc_link_reset(struct tipc_link *l_ptr)
}
/* Clean up all queues, except inputq: */
- __skb_queue_purge(&l_ptr->outqueue);
- __skb_queue_purge(&l_ptr->deferred_queue);
+ __skb_queue_purge(&l_ptr->transmq);
+ __skb_queue_purge(&l_ptr->backlogq);
+ __skb_queue_purge(&l_ptr->deferdq);
if (!owner->inputq)
owner->inputq = &l_ptr->inputq;
skb_queue_splice_init(&l_ptr->wakeupq, owner->inputq);
if (!skb_queue_empty(owner->inputq))
owner->action_flags |= TIPC_MSG_EVT;
- l_ptr->next_out = NULL;
- l_ptr->unacked_window = 0;
+ l_ptr->rcv_unacked = 0;
l_ptr->checkpoint = 1;
l_ptr->next_out_no = 1;
l_ptr->fsm_msg_cnt = 0;
@@ -704,7 +707,7 @@ static int tipc_link_cong(struct tipc_link *link, struct sk_buff_head *list)
{
struct sk_buff *skb = skb_peek(list);
struct tipc_msg *msg = buf_msg(skb);
- uint imp = tipc_msg_tot_importance(msg);
+ int imp = msg_importance(msg);
u32 oport = msg_tot_origport(msg);
if (unlikely(imp > TIPC_CRITICAL_IMPORTANCE)) {
@@ -740,54 +743,51 @@ int __tipc_link_xmit(struct net *net, struct tipc_link *link,
struct sk_buff_head *list)
{
struct tipc_msg *msg = buf_msg(skb_peek(list));
- uint psz = msg_size(msg);
- uint sndlim = link->queue_limit[0];
- uint imp = tipc_msg_tot_importance(msg);
+ unsigned int maxwin = link->window;
+ unsigned int imp = msg_importance(msg);
uint mtu = link->max_pkt;
uint ack = mod(link->next_in_no - 1);
uint seqno = link->next_out_no;
uint bc_last_in = link->owner->bclink.last_in;
struct tipc_media_addr *addr = &link->media_addr;
- struct sk_buff_head *outqueue = &link->outqueue;
+ struct sk_buff_head *transmq = &link->transmq;
+ struct sk_buff_head *backlogq = &link->backlogq;
struct sk_buff *skb, *tmp;
- /* Match queue limits against msg importance: */
- if (unlikely(skb_queue_len(outqueue) >= link->queue_limit[imp]))
+ /* Match queue limit against msg importance: */
+ if (unlikely(skb_queue_len(backlogq) >= link->queue_limit[imp]))
return tipc_link_cong(link, list);
/* Has valid packet limit been used ? */
- if (unlikely(psz > mtu)) {
+ if (unlikely(msg_size(msg) > mtu)) {
__skb_queue_purge(list);
return -EMSGSIZE;
}
- /* Prepare each packet for sending, and add to outqueue: */
+ /* Prepare each packet for sending, and add to relevant queue: */
skb_queue_walk_safe(list, skb, tmp) {
__skb_unlink(skb, list);
msg = buf_msg(skb);
- msg_set_word(msg, 2, ((ack << 16) | mod(seqno)));
+ msg_set_seqno(msg, seqno);
+ msg_set_ack(msg, ack);
msg_set_bcast_ack(msg, bc_last_in);
- if (skb_queue_len(outqueue) < sndlim) {
- __skb_queue_tail(outqueue, skb);
- tipc_bearer_send(net, link->bearer_id,
- skb, addr);
- link->next_out = NULL;
- link->unacked_window = 0;
- } else if (tipc_msg_bundle(outqueue, skb, mtu)) {
+ if (likely(skb_queue_len(transmq) < maxwin)) {
+ __skb_queue_tail(transmq, skb);
+ tipc_bearer_send(net, link->bearer_id, skb, addr);
+ link->rcv_unacked = 0;
+ seqno++;
+ continue;
+ }
+ if (tipc_msg_bundle(skb_peek_tail(backlogq), skb, mtu)) {
link->stats.sent_bundled++;
continue;
- } else if (tipc_msg_make_bundle(outqueue, skb, mtu,
- link->addr)) {
+ }
+ if (tipc_msg_make_bundle(&skb, mtu, link->addr)) {
link->stats.sent_bundled++;
link->stats.sent_bundles++;
- if (!link->next_out)
- link->next_out = skb_peek_tail(outqueue);
- } else {
- __skb_queue_tail(outqueue, skb);
- if (!link->next_out)
- link->next_out = skb;
}
+ __skb_queue_tail(backlogq, skb);
seqno++;
}
link->next_out_no = seqno;
@@ -845,8 +845,10 @@ int tipc_link_xmit(struct net *net, struct sk_buff_head *list, u32 dnode,
if (link)
return rc;
- if (likely(in_own_node(net, dnode)))
- return tipc_sk_rcv(net, list);
+ if (likely(in_own_node(net, dnode))) {
+ tipc_sk_rcv(net, list);
+ return 0;
+ }
__skb_queue_purge(list);
return rc;
@@ -893,14 +895,6 @@ static void tipc_link_sync_rcv(struct tipc_node *n, struct sk_buff *buf)
kfree_skb(buf);
}
-struct sk_buff *tipc_skb_queue_next(const struct sk_buff_head *list,
- const struct sk_buff *skb)
-{
- if (skb_queue_is_last(list, skb))
- return NULL;
- return skb->next;
-}
-
/*
* tipc_link_push_packets - push unsent packets to bearer
*
@@ -909,30 +903,23 @@ struct sk_buff *tipc_skb_queue_next(const struct sk_buff_head *list,
*
* Called with node locked
*/
-void tipc_link_push_packets(struct tipc_link *l_ptr)
+void tipc_link_push_packets(struct tipc_link *link)
{
- struct sk_buff_head *outqueue = &l_ptr->outqueue;
- struct sk_buff *skb = l_ptr->next_out;
+ struct sk_buff *skb;
struct tipc_msg *msg;
- u32 next, first;
+ unsigned int ack = mod(link->next_in_no - 1);
- skb_queue_walk_from(outqueue, skb) {
- msg = buf_msg(skb);
- next = msg_seqno(msg);
- first = buf_seqno(skb_peek(outqueue));
-
- if (mod(next - first) < l_ptr->queue_limit[0]) {
- msg_set_ack(msg, mod(l_ptr->next_in_no - 1));
- msg_set_bcast_ack(msg, l_ptr->owner->bclink.last_in);
- if (msg_user(msg) == MSG_BUNDLER)
- TIPC_SKB_CB(skb)->bundling = false;
- tipc_bearer_send(l_ptr->owner->net,
- l_ptr->bearer_id, skb,
- &l_ptr->media_addr);
- l_ptr->next_out = tipc_skb_queue_next(outqueue, skb);
- } else {
+ while (skb_queue_len(&link->transmq) < link->window) {
+ skb = __skb_dequeue(&link->backlogq);
+ if (!skb)
break;
- }
+ msg = buf_msg(skb);
+ msg_set_ack(msg, ack);
+ msg_set_bcast_ack(msg, link->owner->bclink.last_in);
+ link->rcv_unacked = 0;
+ __skb_queue_tail(&link->transmq, skb);
+ tipc_bearer_send(link->owner->net, link->bearer_id,
+ skb, &link->media_addr);
}
}
@@ -1019,8 +1006,8 @@ void tipc_link_retransmit(struct tipc_link *l_ptr, struct sk_buff *skb,
l_ptr->stale_count = 1;
}
- skb_queue_walk_from(&l_ptr->outqueue, skb) {
- if (!retransmits || skb == l_ptr->next_out)
+ skb_queue_walk_from(&l_ptr->transmq, skb) {
+ if (!retransmits)
break;
msg = buf_msg(skb);
msg_set_ack(msg, mod(l_ptr->next_in_no - 1));
@@ -1037,67 +1024,12 @@ static void link_retrieve_defq(struct tipc_link *link,
{
u32 seq_no;
- if (skb_queue_empty(&link->deferred_queue))
+ if (skb_queue_empty(&link->deferdq))
return;
- seq_no = buf_seqno(skb_peek(&link->deferred_queue));
+ seq_no = buf_seqno(skb_peek(&link->deferdq));
if (seq_no == mod(link->next_in_no))
- skb_queue_splice_tail_init(&link->deferred_queue, list);
-}
-
-/**
- * link_recv_buf_validate - validate basic format of received message
- *
- * This routine ensures a TIPC message has an acceptable header, and at least
- * as much data as the header indicates it should. The routine also ensures
- * that the entire message header is stored in the main fragment of the message
- * buffer, to simplify future access to message header fields.
- *
- * Note: Having extra info present in the message header or data areas is OK.
- * TIPC will ignore the excess, under the assumption that it is optional info
- * introduced by a later release of the protocol.
- */
-static int link_recv_buf_validate(struct sk_buff *buf)
-{
- static u32 min_data_hdr_size[8] = {
- SHORT_H_SIZE, MCAST_H_SIZE, NAMED_H_SIZE, BASIC_H_SIZE,
- MAX_H_SIZE, MAX_H_SIZE, MAX_H_SIZE, MAX_H_SIZE
- };
-
- struct tipc_msg *msg;
- u32 tipc_hdr[2];
- u32 size;
- u32 hdr_size;
- u32 min_hdr_size;
-
- /* If this packet comes from the defer queue, the skb has already
- * been validated
- */
- if (unlikely(TIPC_SKB_CB(buf)->deferred))
- return 1;
-
- if (unlikely(buf->len < MIN_H_SIZE))
- return 0;
-
- msg = skb_header_pointer(buf, 0, sizeof(tipc_hdr), tipc_hdr);
- if (msg == NULL)
- return 0;
-
- if (unlikely(msg_version(msg) != TIPC_VERSION))
- return 0;
-
- size = msg_size(msg);
- hdr_size = msg_hdr_sz(msg);
- min_hdr_size = msg_isdata(msg) ?
- min_data_hdr_size[msg_type(msg)] : INT_H_SIZE;
-
- if (unlikely((hdr_size < min_hdr_size) ||
- (size < hdr_size) ||
- (buf->len < size) ||
- (size - hdr_size > TIPC_MAX_USER_MSG_SIZE)))
- return 0;
-
- return pskb_may_pull(buf, hdr_size);
+ skb_queue_splice_tail_init(&link->deferdq, list);
}
/**
@@ -1125,16 +1057,11 @@ void tipc_rcv(struct net *net, struct sk_buff *skb, struct tipc_bearer *b_ptr)
while ((skb = __skb_dequeue(&head))) {
/* Ensure message is well-formed */
- if (unlikely(!link_recv_buf_validate(skb)))
- goto discard;
-
- /* Ensure message data is a single contiguous unit */
- if (unlikely(skb_linearize(skb)))
+ if (unlikely(!tipc_msg_validate(skb)))
goto discard;
/* Handle arrival of a non-unicast link message */
msg = buf_msg(skb);
-
if (unlikely(msg_non_seq(msg))) {
if (msg_user(msg) == LINK_CONFIG)
tipc_disc_rcv(net, skb, b_ptr);
@@ -1175,21 +1102,20 @@ void tipc_rcv(struct net *net, struct sk_buff *skb, struct tipc_bearer *b_ptr)
ackd = msg_ack(msg);
/* Release acked messages */
- if (n_ptr->bclink.recv_permitted)
+ if (unlikely(n_ptr->bclink.acked != msg_bcast_ack(msg)))
tipc_bclink_acknowledge(n_ptr, msg_bcast_ack(msg));
released = 0;
- skb_queue_walk_safe(&l_ptr->outqueue, skb1, tmp) {
- if (skb1 == l_ptr->next_out ||
- more(buf_seqno(skb1), ackd))
+ skb_queue_walk_safe(&l_ptr->transmq, skb1, tmp) {
+ if (more(buf_seqno(skb1), ackd))
break;
- __skb_unlink(skb1, &l_ptr->outqueue);
+ __skb_unlink(skb1, &l_ptr->transmq);
kfree_skb(skb1);
released = 1;
}
/* Try sending any messages link endpoint has pending */
- if (unlikely(l_ptr->next_out))
+ if (unlikely(skb_queue_len(&l_ptr->backlogq)))
tipc_link_push_packets(l_ptr);
if (released && !skb_queue_empty(&l_ptr->wakeupq))
@@ -1224,10 +1150,9 @@ void tipc_rcv(struct net *net, struct sk_buff *skb, struct tipc_bearer *b_ptr)
goto unlock;
}
l_ptr->next_in_no++;
- if (unlikely(!skb_queue_empty(&l_ptr->deferred_queue)))
+ if (unlikely(!skb_queue_empty(&l_ptr->deferdq)))
link_retrieve_defq(l_ptr, &head);
-
- if (unlikely(++l_ptr->unacked_window >= TIPC_MIN_LINK_WIN)) {
+ if (unlikely(++l_ptr->rcv_unacked >= TIPC_MIN_LINK_WIN)) {
l_ptr->stats.sent_acks++;
tipc_link_proto_xmit(l_ptr, STATE_MSG, 0, 0, 0, 0, 0);
}
@@ -1394,10 +1319,9 @@ static void link_handle_out_of_seq_msg(struct tipc_link *l_ptr,
return;
}
- if (tipc_link_defer_pkt(&l_ptr->deferred_queue, buf)) {
+ if (tipc_link_defer_pkt(&l_ptr->deferdq, buf)) {
l_ptr->stats.deferred_recv++;
- TIPC_SKB_CB(buf)->deferred = true;
- if ((skb_queue_len(&l_ptr->deferred_queue) % 16) == 1)
+ if ((skb_queue_len(&l_ptr->deferdq) % TIPC_MIN_LINK_WIN) == 1)
tipc_link_proto_xmit(l_ptr, STATE_MSG, 0, 0, 0, 0, 0);
} else {
l_ptr->stats.duplicates++;
@@ -1434,11 +1358,11 @@ void tipc_link_proto_xmit(struct tipc_link *l_ptr, u32 msg_typ, int probe_msg,
if (!tipc_link_is_up(l_ptr))
return;
- if (l_ptr->next_out)
- next_sent = buf_seqno(l_ptr->next_out);
+ if (skb_queue_len(&l_ptr->backlogq))
+ next_sent = buf_seqno(skb_peek(&l_ptr->backlogq));
msg_set_next_sent(msg, next_sent);
- if (!skb_queue_empty(&l_ptr->deferred_queue)) {
- u32 rec = buf_seqno(skb_peek(&l_ptr->deferred_queue));
+ if (!skb_queue_empty(&l_ptr->deferdq)) {
+ u32 rec = buf_seqno(skb_peek(&l_ptr->deferdq));
gap = mod(rec - mod(l_ptr->next_in_no));
}
msg_set_seq_gap(msg, gap);
@@ -1490,10 +1414,9 @@ void tipc_link_proto_xmit(struct tipc_link *l_ptr, u32 msg_typ, int probe_msg,
skb_copy_to_linear_data(buf, msg, sizeof(l_ptr->proto_msg));
buf->priority = TC_PRIO_CONTROL;
-
tipc_bearer_send(l_ptr->owner->net, l_ptr->bearer_id, buf,
&l_ptr->media_addr);
- l_ptr->unacked_window = 0;
+ l_ptr->rcv_unacked = 0;
kfree_skb(buf);
}
@@ -1628,7 +1551,7 @@ static void tipc_link_proto_rcv(struct tipc_link *l_ptr,
}
if (msg_seq_gap(msg)) {
l_ptr->stats.recv_nacks++;
- tipc_link_retransmit(l_ptr, skb_peek(&l_ptr->outqueue),
+ tipc_link_retransmit(l_ptr, skb_peek(&l_ptr->transmq),
msg_seq_gap(msg));
}
break;
@@ -1675,7 +1598,7 @@ static void tipc_link_tunnel_xmit(struct tipc_link *l_ptr,
*/
void tipc_link_failover_send_queue(struct tipc_link *l_ptr)
{
- u32 msgcount = skb_queue_len(&l_ptr->outqueue);
+ int msgcount;
struct tipc_link *tunnel = l_ptr->owner->active_links[0];
struct tipc_msg tunnel_hdr;
struct sk_buff *skb;
@@ -1686,10 +1609,12 @@ void tipc_link_failover_send_queue(struct tipc_link *l_ptr)
tipc_msg_init(link_own_addr(l_ptr), &tunnel_hdr, CHANGEOVER_PROTOCOL,
ORIGINAL_MSG, INT_H_SIZE, l_ptr->addr);
+ skb_queue_splice_tail_init(&l_ptr->backlogq, &l_ptr->transmq);
+ msgcount = skb_queue_len(&l_ptr->transmq);
msg_set_bearer_id(&tunnel_hdr, l_ptr->peer_bearer_id);
msg_set_msgcnt(&tunnel_hdr, msgcount);
- if (skb_queue_empty(&l_ptr->outqueue)) {
+ if (skb_queue_empty(&l_ptr->transmq)) {
skb = tipc_buf_acquire(INT_H_SIZE);
if (skb) {
skb_copy_to_linear_data(skb, &tunnel_hdr, INT_H_SIZE);
@@ -1705,7 +1630,7 @@ void tipc_link_failover_send_queue(struct tipc_link *l_ptr)
split_bundles = (l_ptr->owner->active_links[0] !=
l_ptr->owner->active_links[1]);
- skb_queue_walk(&l_ptr->outqueue, skb) {
+ skb_queue_walk(&l_ptr->transmq, skb) {
struct tipc_msg *msg = buf_msg(skb);
if ((msg_user(msg) == MSG_BUNDLER) && split_bundles) {
@@ -1736,80 +1661,66 @@ void tipc_link_failover_send_queue(struct tipc_link *l_ptr)
* and sequence order is preserved per sender/receiver socket pair.
* Owner node is locked.
*/
-void tipc_link_dup_queue_xmit(struct tipc_link *l_ptr,
- struct tipc_link *tunnel)
+void tipc_link_dup_queue_xmit(struct tipc_link *link,
+ struct tipc_link *tnl)
{
struct sk_buff *skb;
- struct tipc_msg tunnel_hdr;
-
- tipc_msg_init(link_own_addr(l_ptr), &tunnel_hdr, CHANGEOVER_PROTOCOL,
- DUPLICATE_MSG, INT_H_SIZE, l_ptr->addr);
- msg_set_msgcnt(&tunnel_hdr, skb_queue_len(&l_ptr->outqueue));
- msg_set_bearer_id(&tunnel_hdr, l_ptr->peer_bearer_id);
- skb_queue_walk(&l_ptr->outqueue, skb) {
+ struct tipc_msg tnl_hdr;
+ struct sk_buff_head *queue = &link->transmq;
+ int mcnt;
+
+ tipc_msg_init(link_own_addr(link), &tnl_hdr, CHANGEOVER_PROTOCOL,
+ DUPLICATE_MSG, INT_H_SIZE, link->addr);
+ mcnt = skb_queue_len(&link->transmq) + skb_queue_len(&link->backlogq);
+ msg_set_msgcnt(&tnl_hdr, mcnt);
+ msg_set_bearer_id(&tnl_hdr, link->peer_bearer_id);
+
+tunnel_queue:
+ skb_queue_walk(queue, skb) {
struct sk_buff *outskb;
struct tipc_msg *msg = buf_msg(skb);
- u32 length = msg_size(msg);
+ u32 len = msg_size(msg);
- if (msg_user(msg) == MSG_BUNDLER)
- msg_set_type(msg, CLOSED_MSG);
- msg_set_ack(msg, mod(l_ptr->next_in_no - 1)); /* Update */
- msg_set_bcast_ack(msg, l_ptr->owner->bclink.last_in);
- msg_set_size(&tunnel_hdr, length + INT_H_SIZE);
- outskb = tipc_buf_acquire(length + INT_H_SIZE);
+ msg_set_ack(msg, mod(link->next_in_no - 1));
+ msg_set_bcast_ack(msg, link->owner->bclink.last_in);
+ msg_set_size(&tnl_hdr, len + INT_H_SIZE);
+ outskb = tipc_buf_acquire(len + INT_H_SIZE);
if (outskb == NULL) {
pr_warn("%sunable to send duplicate msg\n",
link_co_err);
return;
}
- skb_copy_to_linear_data(outskb, &tunnel_hdr, INT_H_SIZE);
- skb_copy_to_linear_data_offset(outskb, INT_H_SIZE, skb->data,
- length);
- __tipc_link_xmit_skb(tunnel, outskb);
- if (!tipc_link_is_up(l_ptr))
+ skb_copy_to_linear_data(outskb, &tnl_hdr, INT_H_SIZE);
+ skb_copy_to_linear_data_offset(outskb, INT_H_SIZE,
+ skb->data, len);
+ __tipc_link_xmit_skb(tnl, outskb);
+ if (!tipc_link_is_up(link))
return;
}
-}
-
-/**
- * buf_extract - extracts embedded TIPC message from another message
- * @skb: encapsulating message buffer
- * @from_pos: offset to extract from
- *
- * Returns a new message buffer containing an embedded message. The
- * encapsulating buffer is left unchanged.
- */
-static struct sk_buff *buf_extract(struct sk_buff *skb, u32 from_pos)
-{
- struct tipc_msg *msg = (struct tipc_msg *)(skb->data + from_pos);
- u32 size = msg_size(msg);
- struct sk_buff *eb;
-
- eb = tipc_buf_acquire(size);
- if (eb)
- skb_copy_to_linear_data(eb, msg, size);
- return eb;
+ if (queue == &link->backlogq)
+ return;
+ queue = &link->backlogq;
+ goto tunnel_queue;
}
/* tipc_link_dup_rcv(): Receive a tunnelled DUPLICATE_MSG packet.
* Owner node is locked.
*/
-static void tipc_link_dup_rcv(struct tipc_link *l_ptr,
- struct sk_buff *t_buf)
+static void tipc_link_dup_rcv(struct tipc_link *link,
+ struct sk_buff *skb)
{
- struct sk_buff *buf;
+ struct sk_buff *iskb;
+ int pos = 0;
- if (!tipc_link_is_up(l_ptr))
+ if (!tipc_link_is_up(link))
return;
- buf = buf_extract(t_buf, INT_H_SIZE);
- if (buf == NULL) {
+ if (!tipc_msg_extract(skb, &iskb, &pos)) {
pr_warn("%sfailed to extract inner dup pkt\n", link_co_err);
return;
}
-
- /* Add buffer to deferred queue, if applicable: */
- link_handle_out_of_seq_msg(l_ptr, buf);
+ /* Append buffer to deferred queue, if applicable: */
+ link_handle_out_of_seq_msg(link, iskb);
}
/* tipc_link_failover_rcv(): Receive a tunnelled ORIGINAL_MSG packet
@@ -1821,6 +1732,7 @@ static struct sk_buff *tipc_link_failover_rcv(struct tipc_link *l_ptr,
struct tipc_msg *t_msg = buf_msg(t_buf);
struct sk_buff *buf = NULL;
struct tipc_msg *msg;
+ int pos = 0;
if (tipc_link_is_up(l_ptr))
tipc_link_reset(l_ptr);
@@ -1832,8 +1744,7 @@ static struct sk_buff *tipc_link_failover_rcv(struct tipc_link *l_ptr,
/* Should there be an inner packet? */
if (l_ptr->exp_msg_count) {
l_ptr->exp_msg_count--;
- buf = buf_extract(t_buf, INT_H_SIZE);
- if (buf == NULL) {
+ if (!tipc_msg_extract(t_buf, &buf, &pos)) {
pr_warn("%sno inner failover pkt\n", link_co_err);
goto exit;
}
@@ -1901,23 +1812,16 @@ static void link_set_supervision_props(struct tipc_link *l_ptr, u32 tol)
l_ptr->abort_limit = tol / (jiffies_to_msecs(l_ptr->cont_intv) / 4);
}
-void tipc_link_set_queue_limits(struct tipc_link *l_ptr, u32 window)
+void tipc_link_set_queue_limits(struct tipc_link *l, u32 win)
{
- /* Data messages from this node, inclusive FIRST_FRAGM */
- l_ptr->queue_limit[TIPC_LOW_IMPORTANCE] = window;
- l_ptr->queue_limit[TIPC_MEDIUM_IMPORTANCE] = (window / 3) * 4;
- l_ptr->queue_limit[TIPC_HIGH_IMPORTANCE] = (window / 3) * 5;
- l_ptr->queue_limit[TIPC_CRITICAL_IMPORTANCE] = (window / 3) * 6;
- /* Transiting data messages,inclusive FIRST_FRAGM */
- l_ptr->queue_limit[TIPC_LOW_IMPORTANCE + 4] = 300;
- l_ptr->queue_limit[TIPC_MEDIUM_IMPORTANCE + 4] = 600;
- l_ptr->queue_limit[TIPC_HIGH_IMPORTANCE + 4] = 900;
- l_ptr->queue_limit[TIPC_CRITICAL_IMPORTANCE + 4] = 1200;
- l_ptr->queue_limit[CONN_MANAGER] = 1200;
- l_ptr->queue_limit[CHANGEOVER_PROTOCOL] = 2500;
- l_ptr->queue_limit[NAME_DISTRIBUTOR] = 3000;
- /* FRAGMENT and LAST_FRAGMENT packets */
- l_ptr->queue_limit[MSG_FRAGMENTER] = 4000;
+ int max_bulk = TIPC_MAX_PUBLICATIONS / (l->max_pkt / ITEM_SIZE);
+
+ l->window = win;
+ l->queue_limit[TIPC_LOW_IMPORTANCE] = win / 2;
+ l->queue_limit[TIPC_MEDIUM_IMPORTANCE] = win;
+ l->queue_limit[TIPC_HIGH_IMPORTANCE] = win / 2 * 3;
+ l->queue_limit[TIPC_CRITICAL_IMPORTANCE] = win * 2;
+ l->queue_limit[TIPC_SYSTEM_IMPORTANCE] = max_bulk;
}
/* tipc_link_find_owner - locate owner node of link by link's name
diff --git a/net/tipc/link.h b/net/tipc/link.h
index 7aeb52092bf3..eec3ecf2d450 100644
--- a/net/tipc/link.h
+++ b/net/tipc/link.h
@@ -124,7 +124,8 @@ struct tipc_stats {
* @max_pkt: current maximum packet size for this link
* @max_pkt_target: desired maximum packet size for this link
* @max_pkt_probes: # of probes based on current (max_pkt, max_pkt_target)
- * @outqueue: outbound message queue
+ * @transmitq: queue for sent, non-acked messages
+ * @backlogq: queue for messages waiting to be sent
* @next_out_no: next sequence number to use for outbound messages
* @last_retransmitted: sequence number of most recently retransmitted message
* @stale_count: # of identical retransmit requests made by peer
@@ -177,20 +178,21 @@ struct tipc_link {
u32 max_pkt_probes;
/* Sending */
- struct sk_buff_head outqueue;
+ struct sk_buff_head transmq;
+ struct sk_buff_head backlogq;
u32 next_out_no;
+ u32 window;
u32 last_retransmitted;
u32 stale_count;
/* Reception */
u32 next_in_no;
- struct sk_buff_head deferred_queue;
- u32 unacked_window;
+ u32 rcv_unacked;
+ struct sk_buff_head deferdq;
struct sk_buff_head inputq;
struct sk_buff_head namedq;
/* Congestion handling */
- struct sk_buff *next_out;
struct sk_buff_head wakeupq;
/* Fragmentation/reassembly */
@@ -302,9 +304,4 @@ static inline int link_reset_reset(struct tipc_link *l_ptr)
return l_ptr->state == RESET_RESET;
}
-static inline int link_congested(struct tipc_link *l_ptr)
-{
- return skb_queue_len(&l_ptr->outqueue) >= l_ptr->queue_limit[0];
-}
-
#endif
diff --git a/net/tipc/msg.c b/net/tipc/msg.c
index b6eb90cd3ef7..0c6dad8180a0 100644
--- a/net/tipc/msg.c
+++ b/net/tipc/msg.c
@@ -1,7 +1,7 @@
/*
* net/tipc/msg.c: TIPC message header routines
*
- * Copyright (c) 2000-2006, 2014, Ericsson AB
+ * Copyright (c) 2000-2006, 2014-2015, Ericsson AB
* Copyright (c) 2005, 2010-2011, Wind River Systems
* All rights reserved.
*
@@ -165,6 +165,9 @@ int tipc_buf_append(struct sk_buff **headbuf, struct sk_buff **buf)
}
if (fragid == LAST_FRAGMENT) {
+ TIPC_SKB_CB(head)->validated = false;
+ if (unlikely(!tipc_msg_validate(head)))
+ goto err;
*buf = head;
TIPC_SKB_CB(head)->tail = NULL;
*headbuf = NULL;
@@ -172,7 +175,6 @@ int tipc_buf_append(struct sk_buff **headbuf, struct sk_buff **buf)
}
*buf = NULL;
return 0;
-
err:
pr_warn_ratelimited("Unable to build fragment list\n");
kfree_skb(*buf);
@@ -181,6 +183,48 @@ err:
return 0;
}
+/* tipc_msg_validate - validate basic format of received message
+ *
+ * This routine ensures a TIPC message has an acceptable header, and at least
+ * as much data as the header indicates it should. The routine also ensures
+ * that the entire message header is stored in the main fragment of the message
+ * buffer, to simplify future access to message header fields.
+ *
+ * Note: Having extra info present in the message header or data areas is OK.
+ * TIPC will ignore the excess, under the assumption that it is optional info
+ * introduced by a later release of the protocol.
+ */
+bool tipc_msg_validate(struct sk_buff *skb)
+{
+ struct tipc_msg *msg;
+ int msz, hsz;
+
+ if (unlikely(TIPC_SKB_CB(skb)->validated))
+ return true;
+ if (unlikely(!pskb_may_pull(skb, MIN_H_SIZE)))
+ return false;
+
+ hsz = msg_hdr_sz(buf_msg(skb));
+ if (unlikely(hsz < MIN_H_SIZE) || (hsz > MAX_H_SIZE))
+ return false;
+ if (unlikely(!pskb_may_pull(skb, hsz)))
+ return false;
+
+ msg = buf_msg(skb);
+ if (unlikely(msg_version(msg) != TIPC_VERSION))
+ return false;
+
+ msz = msg_size(msg);
+ if (unlikely(msz < hsz))
+ return false;
+ if (unlikely((msz - hsz) > TIPC_MAX_USER_MSG_SIZE))
+ return false;
+ if (unlikely(skb->len < msz))
+ return false;
+
+ TIPC_SKB_CB(skb)->validated = true;
+ return true;
+}
/**
* tipc_msg_build - create buffer chain containing specified header and data
@@ -228,6 +272,7 @@ int tipc_msg_build(struct tipc_msg *mhdr, struct msghdr *m,
FIRST_FRAGMENT, INT_H_SIZE, msg_destnode(mhdr));
msg_set_size(&pkthdr, pktmax);
msg_set_fragm_no(&pkthdr, pktno);
+ msg_set_importance(&pkthdr, msg_importance(mhdr));
/* Prepare first fragment */
skb = tipc_buf_acquire(pktmax);
@@ -286,33 +331,36 @@ error:
/**
* tipc_msg_bundle(): Append contents of a buffer to tail of an existing one
- * @list: the buffer chain of the existing buffer ("bundle")
+ * @bskb: the buffer to append to ("bundle")
* @skb: buffer to be appended
* @mtu: max allowable size for the bundle buffer
* Consumes buffer if successful
* Returns true if bundling could be performed, otherwise false
*/
-bool tipc_msg_bundle(struct sk_buff_head *list, struct sk_buff *skb, u32 mtu)
+bool tipc_msg_bundle(struct sk_buff *bskb, struct sk_buff *skb, u32 mtu)
{
- struct sk_buff *bskb = skb_peek_tail(list);
- struct tipc_msg *bmsg = buf_msg(bskb);
+ struct tipc_msg *bmsg;
struct tipc_msg *msg = buf_msg(skb);
- unsigned int bsz = msg_size(bmsg);
+ unsigned int bsz;
unsigned int msz = msg_size(msg);
- u32 start = align(bsz);
+ u32 start, pad;
u32 max = mtu - INT_H_SIZE;
- u32 pad = start - bsz;
if (likely(msg_user(msg) == MSG_FRAGMENTER))
return false;
+ if (!bskb)
+ return false;
+ bmsg = buf_msg(bskb);
+ bsz = msg_size(bmsg);
+ start = align(bsz);
+ pad = start - bsz;
+
if (unlikely(msg_user(msg) == CHANGEOVER_PROTOCOL))
return false;
if (unlikely(msg_user(msg) == BCAST_PROTOCOL))
return false;
if (likely(msg_user(bmsg) != MSG_BUNDLER))
return false;
- if (likely(!TIPC_SKB_CB(bskb)->bundling))
- return false;
if (unlikely(skb_tailroom(bskb) < (pad + msz)))
return false;
if (unlikely(max < (start + msz)))
@@ -328,34 +376,40 @@ bool tipc_msg_bundle(struct sk_buff_head *list, struct sk_buff *skb, u32 mtu)
/**
* tipc_msg_extract(): extract bundled inner packet from buffer
- * @skb: linear outer buffer, to be extracted from.
+ * @skb: buffer to be extracted from.
* @iskb: extracted inner buffer, to be returned
- * @pos: position of msg to be extracted. Returns with pointer of next msg
+ * @pos: position in outer message of msg to be extracted.
+ * Returns position of next msg
* Consumes outer buffer when last packet extracted
* Returns true when when there is an extracted buffer, otherwise false
*/
bool tipc_msg_extract(struct sk_buff *skb, struct sk_buff **iskb, int *pos)
{
- struct tipc_msg *msg = buf_msg(skb);
- int imsz;
- struct tipc_msg *imsg = (struct tipc_msg *)(msg_data(msg) + *pos);
+ struct tipc_msg *msg;
+ int imsz, offset;
+
+ *iskb = NULL;
+ if (unlikely(skb_linearize(skb)))
+ goto none;
- /* Is there space left for shortest possible message? */
- if (*pos > (msg_data_sz(msg) - SHORT_H_SIZE))
+ msg = buf_msg(skb);
+ offset = msg_hdr_sz(msg) + *pos;
+ if (unlikely(offset > (msg_size(msg) - MIN_H_SIZE)))
goto none;
- imsz = msg_size(imsg);
- /* Is there space left for current message ? */
- if ((*pos + imsz) > msg_data_sz(msg))
+ *iskb = skb_clone(skb, GFP_ATOMIC);
+ if (unlikely(!*iskb))
goto none;
- *iskb = tipc_buf_acquire(imsz);
- if (!*iskb)
+ skb_pull(*iskb, offset);
+ imsz = msg_size(buf_msg(*iskb));
+ skb_trim(*iskb, imsz);
+ if (unlikely(!tipc_msg_validate(*iskb)))
goto none;
- skb_copy_to_linear_data(*iskb, imsg, imsz);
*pos += align(imsz);
return true;
none:
kfree_skb(skb);
+ kfree_skb(*iskb);
*iskb = NULL;
return false;
}
@@ -369,12 +423,11 @@ none:
* Replaces buffer if successful
* Returns true if success, otherwise false
*/
-bool tipc_msg_make_bundle(struct sk_buff_head *list,
- struct sk_buff *skb, u32 mtu, u32 dnode)
+bool tipc_msg_make_bundle(struct sk_buff **skb, u32 mtu, u32 dnode)
{
struct sk_buff *bskb;
struct tipc_msg *bmsg;
- struct tipc_msg *msg = buf_msg(skb);
+ struct tipc_msg *msg = buf_msg(*skb);
u32 msz = msg_size(msg);
u32 max = mtu - INT_H_SIZE;
@@ -398,9 +451,9 @@ bool tipc_msg_make_bundle(struct sk_buff_head *list,
msg_set_seqno(bmsg, msg_seqno(msg));
msg_set_ack(bmsg, msg_ack(msg));
msg_set_bcast_ack(bmsg, msg_bcast_ack(msg));
- TIPC_SKB_CB(bskb)->bundling = true;
- __skb_queue_tail(list, bskb);
- return tipc_msg_bundle(list, skb, mtu);
+ tipc_msg_bundle(bskb, *skb, mtu);
+ *skb = bskb;
+ return true;
}
/**
@@ -415,21 +468,17 @@ bool tipc_msg_reverse(u32 own_addr, struct sk_buff *buf, u32 *dnode,
int err)
{
struct tipc_msg *msg = buf_msg(buf);
- uint imp = msg_importance(msg);
struct tipc_msg ohdr;
uint rdsz = min_t(uint, msg_data_sz(msg), MAX_FORWARD_SIZE);
if (skb_linearize(buf))
goto exit;
+ msg = buf_msg(buf);
if (msg_dest_droppable(msg))
goto exit;
if (msg_errcode(msg))
goto exit;
-
memcpy(&ohdr, msg, msg_hdr_sz(msg));
- imp = min_t(uint, imp + 1, TIPC_CRITICAL_IMPORTANCE);
- if (msg_isdata(msg))
- msg_set_importance(msg, imp);
msg_set_errcode(msg, err);
msg_set_origport(msg, msg_destport(&ohdr));
msg_set_destport(msg, msg_origport(&ohdr));
diff --git a/net/tipc/msg.h b/net/tipc/msg.h
index 9ace47f44a69..bd3969a80dd4 100644
--- a/net/tipc/msg.h
+++ b/net/tipc/msg.h
@@ -1,7 +1,7 @@
/*
* net/tipc/msg.h: Include file for TIPC message header routines
*
- * Copyright (c) 2000-2007, 2014, Ericsson AB
+ * Copyright (c) 2000-2007, 2014-2015 Ericsson AB
* Copyright (c) 2005-2008, 2010-2011, Wind River Systems
* All rights reserved.
*
@@ -54,6 +54,8 @@ struct plist;
* - TIPC_HIGH_IMPORTANCE
* - TIPC_CRITICAL_IMPORTANCE
*/
+#define TIPC_SYSTEM_IMPORTANCE 4
+
/*
* Payload message types
@@ -64,6 +66,19 @@ struct plist;
#define TIPC_DIRECT_MSG 3
/*
+ * Internal message users
+ */
+#define BCAST_PROTOCOL 5
+#define MSG_BUNDLER 6
+#define LINK_PROTOCOL 7
+#define CONN_MANAGER 8
+#define CHANGEOVER_PROTOCOL 10
+#define NAME_DISTRIBUTOR 11
+#define MSG_FRAGMENTER 12
+#define LINK_CONFIG 13
+#define SOCK_WAKEUP 14 /* pseudo user */
+
+/*
* Message header sizes
*/
#define SHORT_H_SIZE 24 /* In-cluster basic payload message */
@@ -76,7 +91,7 @@ struct plist;
#define MAX_MSG_SIZE (MAX_H_SIZE + TIPC_MAX_USER_MSG_SIZE)
-#define TIPC_MEDIA_ADDR_OFFSET 5
+#define TIPC_MEDIA_INFO_OFFSET 5
/**
* TIPC message buffer code
@@ -87,12 +102,12 @@ struct plist;
* Note: Headroom should be a multiple of 4 to ensure the TIPC header fields
* are word aligned for quicker access
*/
-#define BUF_HEADROOM LL_MAX_HEADER
+#define BUF_HEADROOM (LL_MAX_HEADER + 48)
struct tipc_skb_cb {
void *handle;
struct sk_buff *tail;
- bool deferred;
+ bool validated;
bool wakeup_pending;
bool bundling;
u16 chain_sz;
@@ -170,16 +185,6 @@ static inline void msg_set_user(struct tipc_msg *m, u32 n)
msg_set_bits(m, 0, 25, 0xf, n);
}
-static inline u32 msg_importance(struct tipc_msg *m)
-{
- return msg_bits(m, 0, 25, 0xf);
-}
-
-static inline void msg_set_importance(struct tipc_msg *m, u32 i)
-{
- msg_set_user(m, i);
-}
-
static inline u32 msg_hdr_sz(struct tipc_msg *m)
{
return msg_bits(m, 0, 21, 0xf) << 2;
@@ -336,6 +341,25 @@ static inline void msg_set_seqno(struct tipc_msg *m, u32 n)
/*
* Words 3-10
*/
+static inline u32 msg_importance(struct tipc_msg *m)
+{
+ if (unlikely(msg_user(m) == MSG_FRAGMENTER))
+ return msg_bits(m, 5, 13, 0x7);
+ if (likely(msg_isdata(m) && !msg_errcode(m)))
+ return msg_user(m);
+ return TIPC_SYSTEM_IMPORTANCE;
+}
+
+static inline void msg_set_importance(struct tipc_msg *m, u32 i)
+{
+ if (unlikely(msg_user(m) == MSG_FRAGMENTER))
+ msg_set_bits(m, 5, 13, 0x7, i);
+ else if (likely(i < TIPC_SYSTEM_IMPORTANCE))
+ msg_set_user(m, i);
+ else
+ pr_warn("Trying to set illegal importance in message\n");
+}
+
static inline u32 msg_prevnode(struct tipc_msg *m)
{
return msg_word(m, 3);
@@ -458,20 +482,6 @@ static inline struct tipc_msg *msg_get_wrapped(struct tipc_msg *m)
*/
/*
- * Internal message users
- */
-#define BCAST_PROTOCOL 5
-#define MSG_BUNDLER 6
-#define LINK_PROTOCOL 7
-#define CONN_MANAGER 8
-#define ROUTE_DISTRIBUTOR 9 /* obsoleted */
-#define CHANGEOVER_PROTOCOL 10
-#define NAME_DISTRIBUTOR 11
-#define MSG_FRAGMENTER 12
-#define LINK_CONFIG 13
-#define SOCK_WAKEUP 14 /* pseudo user */
-
-/*
* Connection management protocol message types
*/
#define CONN_PROBE 0
@@ -510,7 +520,6 @@ static inline struct tipc_msg *msg_get_wrapped(struct tipc_msg *m)
#define DSC_REQ_MSG 0
#define DSC_RESP_MSG 1
-
/*
* Word 1
*/
@@ -534,6 +543,16 @@ static inline void msg_set_node_sig(struct tipc_msg *m, u32 n)
msg_set_bits(m, 1, 0, 0xffff, n);
}
+static inline u32 msg_node_capabilities(struct tipc_msg *m)
+{
+ return msg_bits(m, 1, 15, 0x1fff);
+}
+
+static inline void msg_set_node_capabilities(struct tipc_msg *m, u32 n)
+{
+ msg_set_bits(m, 1, 15, 0x1fff, n);
+}
+
/*
* Word 2
@@ -688,7 +707,7 @@ static inline void msg_set_redundant_link(struct tipc_msg *m, u32 r)
static inline char *msg_media_addr(struct tipc_msg *m)
{
- return (char *)&m->hdr[TIPC_MEDIA_ADDR_OFFSET];
+ return (char *)&m->hdr[TIPC_MEDIA_INFO_OFFSET];
}
/*
@@ -734,13 +753,6 @@ static inline void msg_set_link_tolerance(struct tipc_msg *m, u32 n)
msg_set_bits(m, 9, 0, 0xffff, n);
}
-static inline u32 tipc_msg_tot_importance(struct tipc_msg *m)
-{
- if ((msg_user(m) == MSG_FRAGMENTER) && (msg_type(m) == FIRST_FRAGMENT))
- return msg_importance(msg_get_wrapped(m));
- return msg_importance(m);
-}
-
static inline u32 msg_tot_origport(struct tipc_msg *m)
{
if ((msg_user(m) == MSG_FRAGMENTER) && (msg_type(m) == FIRST_FRAGMENT))
@@ -749,6 +761,7 @@ static inline u32 msg_tot_origport(struct tipc_msg *m)
}
struct sk_buff *tipc_buf_acquire(u32 size);
+bool tipc_msg_validate(struct sk_buff *skb);
bool tipc_msg_reverse(u32 own_addr, struct sk_buff *buf, u32 *dnode,
int err);
void tipc_msg_init(u32 own_addr, struct tipc_msg *m, u32 user, u32 type,
@@ -757,9 +770,9 @@ struct sk_buff *tipc_msg_create(uint user, uint type, uint hdr_sz,
uint data_sz, u32 dnode, u32 onode,
u32 dport, u32 oport, int errcode);
int tipc_buf_append(struct sk_buff **headbuf, struct sk_buff **buf);
-bool tipc_msg_bundle(struct sk_buff_head *list, struct sk_buff *skb, u32 mtu);
-bool tipc_msg_make_bundle(struct sk_buff_head *list,
- struct sk_buff *skb, u32 mtu, u32 dnode);
+bool tipc_msg_bundle(struct sk_buff *bskb, struct sk_buff *skb, u32 mtu);
+
+bool tipc_msg_make_bundle(struct sk_buff **skb, u32 mtu, u32 dnode);
bool tipc_msg_extract(struct sk_buff *skb, struct sk_buff **iskb, int *pos);
int tipc_msg_build(struct tipc_msg *mhdr, struct msghdr *m,
int offset, int dsz, int mtu, struct sk_buff_head *list);
diff --git a/net/tipc/name_distr.c b/net/tipc/name_distr.c
index fcb07915aaac..506aaa565da7 100644
--- a/net/tipc/name_distr.c
+++ b/net/tipc/name_distr.c
@@ -98,7 +98,7 @@ void named_cluster_distribute(struct net *net, struct sk_buff *skb)
continue;
if (!tipc_node_active_links(node))
continue;
- oskb = skb_copy(skb, GFP_ATOMIC);
+ oskb = pskb_copy(skb, GFP_ATOMIC);
if (!oskb)
break;
msg_set_destnode(buf_msg(oskb), dnode);
diff --git a/net/tipc/name_table.c b/net/tipc/name_table.c
index 105ba7adf06f..ab0ac62a1287 100644
--- a/net/tipc/name_table.c
+++ b/net/tipc/name_table.c
@@ -811,8 +811,8 @@ static void tipc_purge_publications(struct net *net, struct name_seq *seq)
sseq = seq->sseqs;
info = sseq->info;
list_for_each_entry_safe(publ, safe, &info->zone_list, zone_list) {
- tipc_nametbl_remove_publ(net, publ->type, publ->lower,
- publ->node, publ->ref, publ->key);
+ tipc_nameseq_remove_publ(net, seq, publ->lower, publ->node,
+ publ->ref, publ->key);
kfree_rcu(publ, rcu);
}
hlist_del_init_rcu(&seq->ns_list);
diff --git a/net/tipc/node.c b/net/tipc/node.c
index 86152de8248d..26d1de1bf34d 100644
--- a/net/tipc/node.c
+++ b/net/tipc/node.c
@@ -111,7 +111,7 @@ struct tipc_node *tipc_node_create(struct net *net, u32 addr)
INIT_LIST_HEAD(&n_ptr->list);
INIT_LIST_HEAD(&n_ptr->publ_list);
INIT_LIST_HEAD(&n_ptr->conn_sks);
- __skb_queue_head_init(&n_ptr->bclink.deferred_queue);
+ __skb_queue_head_init(&n_ptr->bclink.deferdq);
hlist_add_head_rcu(&n_ptr->hash, &tn->node_htable[tipc_hashfn(addr)]);
list_for_each_entry_rcu(temp_node, &tn->node_list, list) {
if (n_ptr->addr < temp_node->addr)
@@ -354,7 +354,7 @@ static void node_lost_contact(struct tipc_node *n_ptr)
/* Flush broadcast link info associated with lost node */
if (n_ptr->bclink.recv_permitted) {
- __skb_queue_purge(&n_ptr->bclink.deferred_queue);
+ __skb_queue_purge(&n_ptr->bclink.deferdq);
if (n_ptr->bclink.reasm_buf) {
kfree_skb(n_ptr->bclink.reasm_buf);
diff --git a/net/tipc/node.h b/net/tipc/node.h
index 3d18c66b7f78..e89ac04ec2c3 100644
--- a/net/tipc/node.h
+++ b/net/tipc/node.h
@@ -84,7 +84,7 @@ struct tipc_node_bclink {
u32 last_sent;
u32 oos_state;
u32 deferred_size;
- struct sk_buff_head deferred_queue;
+ struct sk_buff_head deferdq;
struct sk_buff *reasm_buf;
int inputq_map;
bool recv_permitted;
@@ -106,6 +106,7 @@ struct tipc_node_bclink {
* @list: links to adjacent nodes in sorted list of cluster's nodes
* @working_links: number of working links to node (both active and standby)
* @link_cnt: number of links to node
+ * @capabilities: bitmap, indicating peer node's functional capabilities
* @signature: node instance identifier
* @link_id: local and remote bearer ids of changing link, if any
* @publ_list: list of publications
@@ -125,7 +126,8 @@ struct tipc_node {
struct tipc_node_bclink bclink;
struct list_head list;
int link_cnt;
- int working_links;
+ u16 working_links;
+ u16 capabilities;
u32 signature;
u32 link_id;
struct list_head publ_list;
diff --git a/net/tipc/server.c b/net/tipc/server.c
index eadd4ed45905..ab6183cdb121 100644
--- a/net/tipc/server.c
+++ b/net/tipc/server.c
@@ -37,11 +37,13 @@
#include "core.h"
#include "socket.h"
#include <net/sock.h>
+#include <linux/module.h>
/* Number of messages to send before rescheduling */
#define MAX_SEND_MSG_COUNT 25
#define MAX_RECV_MSG_COUNT 25
#define CF_CONNECTED 1
+#define CF_SERVER 2
#define sock2con(x) ((struct tipc_conn *)(x)->sk_user_data)
@@ -88,9 +90,19 @@ static void tipc_clean_outqueues(struct tipc_conn *con);
static void tipc_conn_kref_release(struct kref *kref)
{
struct tipc_conn *con = container_of(kref, struct tipc_conn, kref);
+ struct sockaddr_tipc *saddr = con->server->saddr;
+ struct socket *sock = con->sock;
+ struct sock *sk;
- if (con->sock) {
- tipc_sock_release_local(con->sock);
+ if (sock) {
+ sk = sock->sk;
+ if (test_bit(CF_SERVER, &con->flags)) {
+ __module_get(sock->ops->owner);
+ __module_get(sk->sk_prot_creator->owner);
+ }
+ saddr->scope = -TIPC_NODE_SCOPE;
+ kernel_bind(sock, (struct sockaddr *)saddr, sizeof(*saddr));
+ sk_release_kernel(sk);
con->sock = NULL;
}
@@ -281,7 +293,7 @@ static int tipc_accept_from_sock(struct tipc_conn *con)
struct tipc_conn *newcon;
int ret;
- ret = tipc_sock_accept_local(sock, &newsock, O_NONBLOCK);
+ ret = kernel_accept(sock, &newsock, O_NONBLOCK);
if (ret < 0)
return ret;
@@ -309,9 +321,12 @@ static struct socket *tipc_create_listen_sock(struct tipc_conn *con)
struct socket *sock = NULL;
int ret;
- ret = tipc_sock_create_local(s->net, s->type, &sock);
+ ret = sock_create_kern(AF_TIPC, SOCK_SEQPACKET, 0, &sock);
if (ret < 0)
return NULL;
+
+ sk_change_net(sock->sk, s->net);
+
ret = kernel_setsockopt(sock, SOL_TIPC, TIPC_IMPORTANCE,
(char *)&s->imp, sizeof(s->imp));
if (ret < 0)
@@ -337,11 +352,31 @@ static struct socket *tipc_create_listen_sock(struct tipc_conn *con)
pr_err("Unknown socket type %d\n", s->type);
goto create_err;
}
+
+ /* As server's listening socket owner and creator is the same module,
+ * we have to decrease TIPC module reference count to guarantee that
+ * it remains zero after the server socket is created, otherwise,
+ * executing "rmmod" command is unable to make TIPC module deleted
+ * after TIPC module is inserted successfully.
+ *
+ * However, the reference count is ever increased twice in
+ * sock_create_kern(): one is to increase the reference count of owner
+ * of TIPC socket's proto_ops struct; another is to increment the
+ * reference count of owner of TIPC proto struct. Therefore, we must
+ * decrement the module reference count twice to ensure that it keeps
+ * zero after server's listening socket is created. Of course, we
+ * must bump the module reference count twice as well before the socket
+ * is closed.
+ */
+ module_put(sock->ops->owner);
+ module_put(sock->sk->sk_prot_creator->owner);
+ set_bit(CF_SERVER, &con->flags);
+
return sock;
create_err:
- sock_release(sock);
- con->sock = NULL;
+ kernel_sock_shutdown(sock, SHUT_RDWR);
+ sk_release_kernel(sock->sk);
return NULL;
}
diff --git a/net/tipc/socket.c b/net/tipc/socket.c
index b4d4467d0bb0..73c2f518a7c0 100644
--- a/net/tipc/socket.c
+++ b/net/tipc/socket.c
@@ -74,6 +74,7 @@
* @link_cong: non-zero if owner must sleep because of link congestion
* @sent_unacked: # messages sent by socket, and not yet acked by peer
* @rcv_unacked: # messages read by user, but not yet acked back to peer
+ * @remote: 'connected' peer for dgram/rdm
* @node: hash table node
* @rcu: rcu struct for tipc_sock
*/
@@ -96,6 +97,7 @@ struct tipc_sock {
bool link_cong;
uint sent_unacked;
uint rcv_unacked;
+ struct sockaddr_tipc remote;
struct rhash_head node;
struct rcu_head rcu;
};
@@ -114,13 +116,14 @@ static int tipc_sk_withdraw(struct tipc_sock *tsk, uint scope,
static struct tipc_sock *tipc_sk_lookup(struct net *net, u32 portid);
static int tipc_sk_insert(struct tipc_sock *tsk);
static void tipc_sk_remove(struct tipc_sock *tsk);
+static int __tipc_send_stream(struct socket *sock, struct msghdr *m,
+ size_t dsz);
+static int __tipc_sendmsg(struct socket *sock, struct msghdr *m, size_t dsz);
static const struct proto_ops packet_ops;
static const struct proto_ops stream_ops;
static const struct proto_ops msg_ops;
-
static struct proto tipc_proto;
-static struct proto tipc_proto_kern;
static const struct nla_policy tipc_nl_sock_policy[TIPC_NLA_SOCK_MAX + 1] = {
[TIPC_NLA_SOCK_UNSPEC] = { .type = NLA_UNSPEC },
@@ -130,6 +133,8 @@ static const struct nla_policy tipc_nl_sock_policy[TIPC_NLA_SOCK_MAX + 1] = {
[TIPC_NLA_SOCK_HAS_PUBL] = { .type = NLA_FLAG }
};
+static const struct rhashtable_params tsk_rht_params;
+
/*
* Revised TIPC socket locking policy:
*
@@ -338,11 +343,7 @@ static int tipc_sk_create(struct net *net, struct socket *sock,
}
/* Allocate socket's protocol area */
- if (!kern)
- sk = sk_alloc(net, AF_TIPC, GFP_KERNEL, &tipc_proto);
- else
- sk = sk_alloc(net, AF_TIPC, GFP_KERNEL, &tipc_proto_kern);
-
+ sk = sk_alloc(net, AF_TIPC, GFP_KERNEL, &tipc_proto);
if (sk == NULL)
return -ENOMEM;
@@ -380,75 +381,6 @@ static int tipc_sk_create(struct net *net, struct socket *sock,
return 0;
}
-/**
- * tipc_sock_create_local - create TIPC socket from inside TIPC module
- * @type: socket type - SOCK_RDM or SOCK_SEQPACKET
- *
- * We cannot use sock_creat_kern here because it bumps module user count.
- * Since socket owner and creator is the same module we must make sure
- * that module count remains zero for module local sockets, otherwise
- * we cannot do rmmod.
- *
- * Returns 0 on success, errno otherwise
- */
-int tipc_sock_create_local(struct net *net, int type, struct socket **res)
-{
- int rc;
-
- rc = sock_create_lite(AF_TIPC, type, 0, res);
- if (rc < 0) {
- pr_err("Failed to create kernel socket\n");
- return rc;
- }
- tipc_sk_create(net, *res, 0, 1);
-
- return 0;
-}
-
-/**
- * tipc_sock_release_local - release socket created by tipc_sock_create_local
- * @sock: the socket to be released.
- *
- * Module reference count is not incremented when such sockets are created,
- * so we must keep it from being decremented when they are released.
- */
-void tipc_sock_release_local(struct socket *sock)
-{
- tipc_release(sock);
- sock->ops = NULL;
- sock_release(sock);
-}
-
-/**
- * tipc_sock_accept_local - accept a connection on a socket created
- * with tipc_sock_create_local. Use this function to avoid that
- * module reference count is inadvertently incremented.
- *
- * @sock: the accepting socket
- * @newsock: reference to the new socket to be created
- * @flags: socket flags
- */
-
-int tipc_sock_accept_local(struct socket *sock, struct socket **newsock,
- int flags)
-{
- struct sock *sk = sock->sk;
- int ret;
-
- ret = sock_create_lite(sk->sk_family, sk->sk_type,
- sk->sk_protocol, newsock);
- if (ret < 0)
- return ret;
-
- ret = tipc_accept(sock, *newsock, flags);
- if (ret < 0) {
- sock_release(*newsock);
- return ret;
- }
- (*newsock)->ops = sock->ops;
- return ret;
-}
-
static void tipc_sk_callback(struct rcu_head *head)
{
struct tipc_sock *tsk = container_of(head, struct tipc_sock, rcu);
@@ -892,7 +824,6 @@ static int tipc_wait_for_sndmsg(struct socket *sock, long *timeo_p)
/**
* tipc_sendmsg - send message in connectionless manner
- * @iocb: if NULL, indicates that socket lock is already held
* @sock: socket structure
* @m: message to send
* @dsz: amount of user data to be sent
@@ -904,9 +835,21 @@ static int tipc_wait_for_sndmsg(struct socket *sock, long *timeo_p)
*
* Returns the number of bytes sent on success, or errno otherwise
*/
-static int tipc_sendmsg(struct kiocb *iocb, struct socket *sock,
+static int tipc_sendmsg(struct socket *sock,
struct msghdr *m, size_t dsz)
{
+ struct sock *sk = sock->sk;
+ int ret;
+
+ lock_sock(sk);
+ ret = __tipc_sendmsg(sock, m, dsz);
+ release_sock(sk);
+
+ return ret;
+}
+
+static int __tipc_sendmsg(struct socket *sock, struct msghdr *m, size_t dsz)
+{
DECLARE_SOCKADDR(struct sockaddr_tipc *, dest, m->msg_name);
struct sock *sk = sock->sk;
struct tipc_sock *tsk = tipc_sk(sk);
@@ -915,49 +858,40 @@ static int tipc_sendmsg(struct kiocb *iocb, struct socket *sock,
u32 dnode, dport;
struct sk_buff_head *pktchain = &sk->sk_write_queue;
struct sk_buff *skb;
- struct tipc_name_seq *seq = &dest->addr.nameseq;
+ struct tipc_name_seq *seq;
struct iov_iter save;
u32 mtu;
long timeo;
int rc;
- if (unlikely(!dest))
- return -EDESTADDRREQ;
-
- if (unlikely((m->msg_namelen < sizeof(*dest)) ||
- (dest->family != AF_TIPC)))
- return -EINVAL;
-
if (dsz > TIPC_MAX_USER_MSG_SIZE)
return -EMSGSIZE;
-
- if (iocb)
- lock_sock(sk);
-
+ if (unlikely(!dest)) {
+ if (tsk->connected && sock->state == SS_READY)
+ dest = &tsk->remote;
+ else
+ return -EDESTADDRREQ;
+ } else if (unlikely(m->msg_namelen < sizeof(*dest)) ||
+ dest->family != AF_TIPC) {
+ return -EINVAL;
+ }
if (unlikely(sock->state != SS_READY)) {
- if (sock->state == SS_LISTENING) {
- rc = -EPIPE;
- goto exit;
- }
- if (sock->state != SS_UNCONNECTED) {
- rc = -EISCONN;
- goto exit;
- }
- if (tsk->published) {
- rc = -EOPNOTSUPP;
- goto exit;
- }
+ if (sock->state == SS_LISTENING)
+ return -EPIPE;
+ if (sock->state != SS_UNCONNECTED)
+ return -EISCONN;
+ if (tsk->published)
+ return -EOPNOTSUPP;
if (dest->addrtype == TIPC_ADDR_NAME) {
tsk->conn_type = dest->addr.name.name.type;
tsk->conn_instance = dest->addr.name.name.instance;
}
}
-
+ seq = &dest->addr.nameseq;
timeo = sock_sndtimeo(sk, m->msg_flags & MSG_DONTWAIT);
if (dest->addrtype == TIPC_ADDR_MCAST) {
- rc = tipc_sendmcast(sock, seq, m, dsz, timeo);
- goto exit;
+ return tipc_sendmcast(sock, seq, m, dsz, timeo);
} else if (dest->addrtype == TIPC_ADDR_NAME) {
u32 type = dest->addr.name.name.type;
u32 inst = dest->addr.name.name.instance;
@@ -972,10 +906,8 @@ static int tipc_sendmsg(struct kiocb *iocb, struct socket *sock,
dport = tipc_nametbl_translate(net, type, inst, &dnode);
msg_set_destnode(mhdr, dnode);
msg_set_destport(mhdr, dport);
- if (unlikely(!dport && !dnode)) {
- rc = -EHOSTUNREACH;
- goto exit;
- }
+ if (unlikely(!dport && !dnode))
+ return -EHOSTUNREACH;
} else if (dest->addrtype == TIPC_ADDR_ID) {
dnode = dest->addr.id.node;
msg_set_type(mhdr, TIPC_DIRECT_MSG);
@@ -990,7 +922,7 @@ new_mtu:
mtu = tipc_node_get_mtu(net, dnode, tsk->portid);
rc = tipc_msg_build(mhdr, m, 0, dsz, mtu, pktchain);
if (rc < 0)
- goto exit;
+ return rc;
do {
skb = skb_peek(pktchain);
@@ -1013,9 +945,6 @@ new_mtu:
if (rc)
__skb_queue_purge(pktchain);
} while (!rc);
-exit:
- if (iocb)
- release_sock(sk);
return rc;
}
@@ -1052,7 +981,6 @@ static int tipc_wait_for_sndpkt(struct socket *sock, long *timeo_p)
/**
* tipc_send_stream - send stream-oriented data
- * @iocb: (unused)
* @sock: socket structure
* @m: data to send
* @dsz: total length of data to be transmitted
@@ -1062,8 +990,19 @@ static int tipc_wait_for_sndpkt(struct socket *sock, long *timeo_p)
* Returns the number of bytes sent on success (or partial success),
* or errno if no data sent
*/
-static int tipc_send_stream(struct kiocb *iocb, struct socket *sock,
- struct msghdr *m, size_t dsz)
+static int tipc_send_stream(struct socket *sock, struct msghdr *m, size_t dsz)
+{
+ struct sock *sk = sock->sk;
+ int ret;
+
+ lock_sock(sk);
+ ret = __tipc_send_stream(sock, m, dsz);
+ release_sock(sk);
+
+ return ret;
+}
+
+static int __tipc_send_stream(struct socket *sock, struct msghdr *m, size_t dsz)
{
struct sock *sk = sock->sk;
struct net *net = sock_net(sk);
@@ -1080,7 +1019,7 @@ static int tipc_send_stream(struct kiocb *iocb, struct socket *sock,
/* Handle implied connection establishment */
if (unlikely(dest)) {
- rc = tipc_sendmsg(iocb, sock, m, dsz);
+ rc = __tipc_sendmsg(sock, m, dsz);
if (dsz && (dsz == rc))
tsk->sent_unacked = 1;
return rc;
@@ -1088,15 +1027,11 @@ static int tipc_send_stream(struct kiocb *iocb, struct socket *sock,
if (dsz > (uint)INT_MAX)
return -EMSGSIZE;
- if (iocb)
- lock_sock(sk);
-
if (unlikely(sock->state != SS_CONNECTED)) {
if (sock->state == SS_DISCONNECTING)
- rc = -EPIPE;
+ return -EPIPE;
else
- rc = -ENOTCONN;
- goto exit;
+ return -ENOTCONN;
}
timeo = sock_sndtimeo(sk, m->msg_flags & MSG_DONTWAIT);
@@ -1108,7 +1043,7 @@ next:
send = min_t(uint, dsz - sent, TIPC_MAX_USER_MSG_SIZE);
rc = tipc_msg_build(mhdr, m, sent, send, mtu, pktchain);
if (unlikely(rc < 0))
- goto exit;
+ return rc;
do {
if (likely(!tsk_conn_cong(tsk))) {
rc = tipc_link_xmit(net, pktchain, dnode, portid);
@@ -1133,15 +1068,12 @@ next:
if (rc)
__skb_queue_purge(pktchain);
} while (!rc);
-exit:
- if (iocb)
- release_sock(sk);
+
return sent ? sent : rc;
}
/**
* tipc_send_packet - send a connection-oriented message
- * @iocb: if NULL, indicates that socket lock is already held
* @sock: socket structure
* @m: message to send
* @dsz: length of data to be transmitted
@@ -1150,13 +1082,12 @@ exit:
*
* Returns the number of bytes sent on success, or errno otherwise
*/
-static int tipc_send_packet(struct kiocb *iocb, struct socket *sock,
- struct msghdr *m, size_t dsz)
+static int tipc_send_packet(struct socket *sock, struct msghdr *m, size_t dsz)
{
if (dsz > TIPC_MAX_USER_MSG_SIZE)
return -EMSGSIZE;
- return tipc_send_stream(iocb, sock, m, dsz);
+ return tipc_send_stream(sock, m, dsz);
}
/* tipc_sk_finish_conn - complete the setup of a connection
@@ -1317,12 +1248,12 @@ static int tipc_wait_for_rcvmsg(struct socket *sock, long *timeop)
err = 0;
if (!skb_queue_empty(&sk->sk_receive_queue))
break;
- err = sock_intr_errno(timeo);
- if (signal_pending(current))
- break;
err = -EAGAIN;
if (!timeo)
break;
+ err = sock_intr_errno(timeo);
+ if (signal_pending(current))
+ break;
}
finish_wait(sk_sleep(sk), &wait);
*timeop = timeo;
@@ -1331,7 +1262,6 @@ static int tipc_wait_for_rcvmsg(struct socket *sock, long *timeop)
/**
* tipc_recvmsg - receive packet-oriented message
- * @iocb: (unused)
* @m: descriptor for message info
* @buf_len: total size of user buffer area
* @flags: receive flags
@@ -1341,8 +1271,8 @@ static int tipc_wait_for_rcvmsg(struct socket *sock, long *timeop)
*
* Returns size of returned message data, errno otherwise
*/
-static int tipc_recvmsg(struct kiocb *iocb, struct socket *sock,
- struct msghdr *m, size_t buf_len, int flags)
+static int tipc_recvmsg(struct socket *sock, struct msghdr *m, size_t buf_len,
+ int flags)
{
struct sock *sk = sock->sk;
struct tipc_sock *tsk = tipc_sk(sk);
@@ -1426,7 +1356,6 @@ exit:
/**
* tipc_recv_stream - receive stream-oriented data
- * @iocb: (unused)
* @m: descriptor for message info
* @buf_len: total size of user buffer area
* @flags: receive flags
@@ -1436,8 +1365,8 @@ exit:
*
* Returns size of returned message data, errno otherwise
*/
-static int tipc_recv_stream(struct kiocb *iocb, struct socket *sock,
- struct msghdr *m, size_t buf_len, int flags)
+static int tipc_recv_stream(struct socket *sock, struct msghdr *m,
+ size_t buf_len, int flags)
{
struct sock *sk = sock->sk;
struct tipc_sock *tsk = tipc_sk(sk);
@@ -1909,17 +1838,24 @@ static int tipc_connect(struct socket *sock, struct sockaddr *dest,
int destlen, int flags)
{
struct sock *sk = sock->sk;
+ struct tipc_sock *tsk = tipc_sk(sk);
struct sockaddr_tipc *dst = (struct sockaddr_tipc *)dest;
struct msghdr m = {NULL,};
- long timeout = (flags & O_NONBLOCK) ? 0 : tipc_sk(sk)->conn_timeout;
+ long timeout = (flags & O_NONBLOCK) ? 0 : tsk->conn_timeout;
socket_state previous;
- int res;
+ int res = 0;
lock_sock(sk);
- /* For now, TIPC does not allow use of connect() with DGRAM/RDM types */
+ /* DGRAM/RDM connect(), just save the destaddr */
if (sock->state == SS_READY) {
- res = -EOPNOTSUPP;
+ if (dst->family == AF_UNSPEC) {
+ memset(&tsk->remote, 0, sizeof(struct sockaddr_tipc));
+ tsk->connected = 0;
+ } else {
+ memcpy(&tsk->remote, dest, destlen);
+ tsk->connected = 1;
+ }
goto exit;
}
@@ -1947,7 +1883,7 @@ static int tipc_connect(struct socket *sock, struct sockaddr *dest,
if (!timeout)
m.msg_flags = MSG_DONTWAIT;
- res = tipc_sendmsg(NULL, sock, &m, 0);
+ res = __tipc_sendmsg(sock, &m, 0);
if ((res < 0) && (res != -EWOULDBLOCK))
goto exit;
@@ -2027,12 +1963,12 @@ static int tipc_wait_for_accept(struct socket *sock, long timeo)
err = -EINVAL;
if (sock->state != SS_LISTENING)
break;
- err = sock_intr_errno(timeo);
- if (signal_pending(current))
- break;
err = -EAGAIN;
if (!timeo)
break;
+ err = sock_intr_errno(timeo);
+ if (signal_pending(current))
+ break;
}
finish_wait(sk_sleep(sk), &wait);
return err;
@@ -2103,7 +2039,7 @@ static int tipc_accept(struct socket *sock, struct socket *new_sock, int flags)
struct msghdr m = {NULL,};
tsk_advance_rx_queue(sk);
- tipc_send_packet(NULL, new_sock, &m, 0);
+ __tipc_send_stream(new_sock, &m, 0);
} else {
__skb_dequeue(&sk->sk_receive_queue);
__skb_queue_head(&new_sk->sk_receive_queue, buf);
@@ -2154,7 +2090,6 @@ restart:
TIPC_CONN_SHUTDOWN))
tipc_link_xmit_skb(net, skb, dnode,
tsk->portid);
- tipc_node_remove_conn(net, dnode, tsk->portid);
} else {
dnode = tsk_peer_node(tsk);
@@ -2312,7 +2247,7 @@ static struct tipc_sock *tipc_sk_lookup(struct net *net, u32 portid)
struct tipc_sock *tsk;
rcu_read_lock();
- tsk = rhashtable_lookup(&tn->sk_rht, &portid);
+ tsk = rhashtable_lookup_fast(&tn->sk_rht, &portid, tsk_rht_params);
if (tsk)
sock_hold(&tsk->sk);
rcu_read_unlock();
@@ -2334,7 +2269,8 @@ static int tipc_sk_insert(struct tipc_sock *tsk)
portid = TIPC_MIN_PORT;
tsk->portid = portid;
sock_hold(&tsk->sk);
- if (rhashtable_lookup_insert(&tn->sk_rht, &tsk->node))
+ if (!rhashtable_lookup_insert_fast(&tn->sk_rht, &tsk->node,
+ tsk_rht_params))
return 0;
sock_put(&tsk->sk);
}
@@ -2347,26 +2283,27 @@ static void tipc_sk_remove(struct tipc_sock *tsk)
struct sock *sk = &tsk->sk;
struct tipc_net *tn = net_generic(sock_net(sk), tipc_net_id);
- if (rhashtable_remove(&tn->sk_rht, &tsk->node)) {
+ if (!rhashtable_remove_fast(&tn->sk_rht, &tsk->node, tsk_rht_params)) {
WARN_ON(atomic_read(&sk->sk_refcnt) == 1);
__sock_put(sk);
}
}
+static const struct rhashtable_params tsk_rht_params = {
+ .nelem_hint = 192,
+ .head_offset = offsetof(struct tipc_sock, node),
+ .key_offset = offsetof(struct tipc_sock, portid),
+ .key_len = sizeof(u32), /* portid */
+ .hashfn = jhash,
+ .max_size = 1048576,
+ .min_size = 256,
+};
+
int tipc_sk_rht_init(struct net *net)
{
struct tipc_net *tn = net_generic(net, tipc_net_id);
- struct rhashtable_params rht_params = {
- .nelem_hint = 192,
- .head_offset = offsetof(struct tipc_sock, node),
- .key_offset = offsetof(struct tipc_sock, portid),
- .key_len = sizeof(u32), /* portid */
- .hashfn = jhash,
- .max_shift = 20, /* 1M */
- .min_shift = 8, /* 256 */
- };
- return rhashtable_init(&tn->sk_rht, &rht_params);
+ return rhashtable_init(&tn->sk_rht, &tsk_rht_params);
}
void tipc_sk_rht_destroy(struct net *net)
@@ -2609,12 +2546,6 @@ static struct proto tipc_proto = {
.sysctl_rmem = sysctl_tipc_rmem
};
-static struct proto tipc_proto_kern = {
- .name = "TIPC",
- .obj_size = sizeof(struct tipc_sock),
- .sysctl_rmem = sysctl_tipc_rmem
-};
-
/**
* tipc_socket_init - initialize TIPC socket interface
*
diff --git a/net/tipc/socket.h b/net/tipc/socket.h
index 238f1b7bd9bd..bf6551389522 100644
--- a/net/tipc/socket.h
+++ b/net/tipc/socket.h
@@ -44,10 +44,6 @@
SKB_TRUESIZE(TIPC_MAX_USER_MSG_SIZE))
int tipc_socket_init(void);
void tipc_socket_stop(void);
-int tipc_sock_create_local(struct net *net, int type, struct socket **res);
-void tipc_sock_release_local(struct socket *sock);
-int tipc_sock_accept_local(struct socket *sock, struct socket **newsock,
- int flags);
int tipc_sk_rcv(struct net *net, struct sk_buff_head *inputq);
void tipc_sk_mcast_rcv(struct net *net, struct sk_buff_head *arrvq,
struct sk_buff_head *inputq);
diff --git a/net/tipc/subscr.c b/net/tipc/subscr.c
index 72c339e432aa..1c147c869c2e 100644
--- a/net/tipc/subscr.c
+++ b/net/tipc/subscr.c
@@ -162,19 +162,6 @@ static void subscr_del(struct tipc_subscription *sub)
atomic_dec(&tn->subscription_count);
}
-/**
- * subscr_terminate - terminate communication with a subscriber
- *
- * Note: Must call it in process context since it might sleep.
- */
-static void subscr_terminate(struct tipc_subscription *sub)
-{
- struct tipc_subscriber *subscriber = sub->subscriber;
- struct tipc_net *tn = net_generic(sub->net, tipc_net_id);
-
- tipc_conn_terminate(tn->topsrv, subscriber->conid);
-}
-
static void subscr_release(struct tipc_subscriber *subscriber)
{
struct tipc_subscription *sub;
@@ -312,16 +299,14 @@ static void subscr_conn_msg_event(struct net *net, int conid,
{
struct tipc_subscriber *subscriber = usr_data;
struct tipc_subscription *sub = NULL;
+ struct tipc_net *tn = net_generic(net, tipc_net_id);
spin_lock_bh(&subscriber->lock);
- if (subscr_subscribe(net, (struct tipc_subscr *)buf, subscriber,
- &sub) < 0) {
- spin_unlock_bh(&subscriber->lock);
- subscr_terminate(sub);
- return;
- }
+ subscr_subscribe(net, (struct tipc_subscr *)buf, subscriber, &sub);
if (sub)
tipc_nametbl_subscribe(sub);
+ else
+ tipc_conn_terminate(tn->topsrv, subscriber->conid);
spin_unlock_bh(&subscriber->lock);
}
diff --git a/net/tipc/udp_media.c b/net/tipc/udp_media.c
new file mode 100644
index 000000000000..ac89101e5d1b
--- /dev/null
+++ b/net/tipc/udp_media.c
@@ -0,0 +1,446 @@
+/* net/tipc/udp_media.c: IP bearer support for TIPC
+ *
+ * Copyright (c) 2015, Ericsson AB
+ * All rights reserved.
+ *
+ * Redistribution and use in source and binary forms, with or without
+ * modification, are permitted provided that the following conditions are met:
+ *
+ * 1. Redistributions of source code must retain the above copyright
+ * notice, this list of conditions and the following disclaimer.
+ * 2. Redistributions in binary form must reproduce the above copyright
+ * notice, this list of conditions and the following disclaimer in the
+ * documentation and/or other materials provided with the distribution.
+ * 3. Neither the names of the copyright holders nor the names of its
+ * contributors may be used to endorse or promote products derived from
+ * this software without specific prior written permission.
+ *
+ * Alternatively, this software may be distributed under the terms of the
+ * GNU General Public License ("GPL") version 2 as published by the Free
+ * Software Foundation.
+ *
+ * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS"
+ * AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE
+ * IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE
+ * ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT OWNER OR CONTRIBUTORS BE
+ * LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR
+ * CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF
+ * SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS
+ * INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN
+ * CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE)
+ * ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE
+ * POSSIBILITY OF SUCH DAMAGE.
+ */
+
+#include <linux/socket.h>
+#include <linux/ip.h>
+#include <linux/udp.h>
+#include <linux/inet.h>
+#include <linux/inetdevice.h>
+#include <linux/igmp.h>
+#include <linux/kernel.h>
+#include <linux/workqueue.h>
+#include <linux/list.h>
+#include <net/sock.h>
+#include <net/ip.h>
+#include <net/udp_tunnel.h>
+#include <net/addrconf.h>
+#include <linux/tipc_netlink.h>
+#include "core.h"
+#include "bearer.h"
+
+/* IANA assigned UDP port */
+#define UDP_PORT_DEFAULT 6118
+
+static const struct nla_policy tipc_nl_udp_policy[TIPC_NLA_UDP_MAX + 1] = {
+ [TIPC_NLA_UDP_UNSPEC] = {.type = NLA_UNSPEC},
+ [TIPC_NLA_UDP_LOCAL] = {.type = NLA_BINARY,
+ .len = sizeof(struct sockaddr_storage)},
+ [TIPC_NLA_UDP_REMOTE] = {.type = NLA_BINARY,
+ .len = sizeof(struct sockaddr_storage)},
+};
+
+/**
+ * struct udp_media_addr - IP/UDP addressing information
+ *
+ * This is the bearer level originating address used in neighbor discovery
+ * messages, and all fields should be in network byte order
+ */
+struct udp_media_addr {
+ __be16 proto;
+ __be16 udp_port;
+ union {
+ struct in_addr ipv4;
+ struct in6_addr ipv6;
+ };
+};
+
+/**
+ * struct udp_bearer - ip/udp bearer data structure
+ * @bearer: associated generic tipc bearer
+ * @ubsock: bearer associated socket
+ * @ifindex: local address scope
+ * @work: used to schedule deferred work on a bearer
+ */
+struct udp_bearer {
+ struct tipc_bearer __rcu *bearer;
+ struct socket *ubsock;
+ u32 ifindex;
+ struct work_struct work;
+};
+
+/* udp_media_addr_set - convert a ip/udp address to a TIPC media address */
+static void tipc_udp_media_addr_set(struct tipc_media_addr *addr,
+ struct udp_media_addr *ua)
+{
+ memset(addr, 0, sizeof(struct tipc_media_addr));
+ addr->media_id = TIPC_MEDIA_TYPE_UDP;
+ memcpy(addr->value, ua, sizeof(struct udp_media_addr));
+ if (ntohs(ua->proto) == ETH_P_IP) {
+ if (ipv4_is_multicast(ua->ipv4.s_addr))
+ addr->broadcast = 1;
+ } else if (ntohs(ua->proto) == ETH_P_IPV6) {
+ if (ipv6_addr_type(&ua->ipv6) & IPV6_ADDR_MULTICAST)
+ addr->broadcast = 1;
+ } else {
+ pr_err("Invalid UDP media address\n");
+ }
+}
+
+/* tipc_udp_addr2str - convert ip/udp address to string */
+static int tipc_udp_addr2str(struct tipc_media_addr *a, char *buf, int size)
+{
+ struct udp_media_addr *ua = (struct udp_media_addr *)&a->value;
+
+ if (ntohs(ua->proto) == ETH_P_IP)
+ snprintf(buf, size, "%pI4:%u", &ua->ipv4, ntohs(ua->udp_port));
+ else if (ntohs(ua->proto) == ETH_P_IPV6)
+ snprintf(buf, size, "%pI6:%u", &ua->ipv6, ntohs(ua->udp_port));
+ else
+ pr_err("Invalid UDP media address\n");
+ return 0;
+}
+
+/* tipc_udp_msg2addr - extract an ip/udp address from a TIPC ndisc message */
+static int tipc_udp_msg2addr(struct tipc_bearer *b, struct tipc_media_addr *a,
+ char *msg)
+{
+ struct udp_media_addr *ua;
+
+ ua = (struct udp_media_addr *) (msg + TIPC_MEDIA_ADDR_OFFSET);
+ if (msg[TIPC_MEDIA_TYPE_OFFSET] != TIPC_MEDIA_TYPE_UDP)
+ return -EINVAL;
+ tipc_udp_media_addr_set(a, ua);
+ return 0;
+}
+
+/* tipc_udp_addr2msg - write an ip/udp address to a TIPC ndisc message */
+static int tipc_udp_addr2msg(char *msg, struct tipc_media_addr *a)
+{
+ memset(msg, 0, TIPC_MEDIA_INFO_SIZE);
+ msg[TIPC_MEDIA_TYPE_OFFSET] = TIPC_MEDIA_TYPE_UDP;
+ memcpy(msg + TIPC_MEDIA_ADDR_OFFSET, a->value,
+ sizeof(struct udp_media_addr));
+ return 0;
+}
+
+/* tipc_send_msg - enqueue a send request */
+static int tipc_udp_send_msg(struct net *net, struct sk_buff *skb,
+ struct tipc_bearer *b,
+ struct tipc_media_addr *dest)
+{
+ int ttl, err = 0;
+ struct udp_bearer *ub;
+ struct udp_media_addr *dst = (struct udp_media_addr *)&dest->value;
+ struct udp_media_addr *src = (struct udp_media_addr *)&b->addr.value;
+ struct sk_buff *clone;
+ struct rtable *rt;
+
+ clone = skb_clone(skb, GFP_ATOMIC);
+ skb_set_inner_protocol(clone, htons(ETH_P_TIPC));
+ ub = rcu_dereference_rtnl(b->media_ptr);
+ if (!ub) {
+ err = -ENODEV;
+ goto tx_error;
+ }
+ if (dst->proto == htons(ETH_P_IP)) {
+ struct flowi4 fl = {
+ .daddr = dst->ipv4.s_addr,
+ .saddr = src->ipv4.s_addr,
+ .flowi4_mark = clone->mark,
+ .flowi4_proto = IPPROTO_UDP
+ };
+ rt = ip_route_output_key(net, &fl);
+ if (IS_ERR(rt)) {
+ err = PTR_ERR(rt);
+ goto tx_error;
+ }
+ ttl = ip4_dst_hoplimit(&rt->dst);
+ err = udp_tunnel_xmit_skb(rt, clone, src->ipv4.s_addr,
+ dst->ipv4.s_addr, 0, ttl, 0,
+ src->udp_port, dst->udp_port,
+ false, true);
+ if (err < 0) {
+ ip_rt_put(rt);
+ goto tx_error;
+ }
+#if IS_ENABLED(CONFIG_IPV6)
+ } else {
+ struct dst_entry *ndst;
+ struct flowi6 fl6 = {
+ .flowi6_oif = ub->ifindex,
+ .daddr = dst->ipv6,
+ .saddr = src->ipv6,
+ .flowi6_proto = IPPROTO_UDP
+ };
+ err = ip6_dst_lookup(ub->ubsock->sk, &ndst, &fl6);
+ if (err)
+ goto tx_error;
+ ttl = ip6_dst_hoplimit(ndst);
+ err = udp_tunnel6_xmit_skb(ndst, clone, ndst->dev, &src->ipv6,
+ &dst->ipv6, 0, ttl, src->udp_port,
+ dst->udp_port, false);
+#endif
+ }
+ return err;
+
+tx_error:
+ kfree_skb(clone);
+ return err;
+}
+
+/* tipc_udp_recv - read data from bearer socket */
+static int tipc_udp_recv(struct sock *sk, struct sk_buff *skb)
+{
+ struct udp_bearer *ub;
+ struct tipc_bearer *b;
+
+ ub = rcu_dereference_sk_user_data(sk);
+ if (!ub) {
+ pr_err_ratelimited("Failed to get UDP bearer reference");
+ kfree_skb(skb);
+ return 0;
+ }
+
+ skb_pull(skb, sizeof(struct udphdr));
+ rcu_read_lock();
+ b = rcu_dereference_rtnl(ub->bearer);
+
+ if (b) {
+ tipc_rcv(sock_net(sk), skb, b);
+ rcu_read_unlock();
+ return 0;
+ }
+ rcu_read_unlock();
+ kfree_skb(skb);
+ return 0;
+}
+
+static int enable_mcast(struct udp_bearer *ub, struct udp_media_addr *remote)
+{
+ int err = 0;
+ struct ip_mreqn mreqn;
+ struct sock *sk = ub->ubsock->sk;
+
+ if (ntohs(remote->proto) == ETH_P_IP) {
+ if (!ipv4_is_multicast(remote->ipv4.s_addr))
+ return 0;
+ mreqn.imr_multiaddr = remote->ipv4;
+ mreqn.imr_ifindex = ub->ifindex;
+ err = ip_mc_join_group(sk, &mreqn);
+#if IS_ENABLED(CONFIG_IPV6)
+ } else {
+ if (!ipv6_addr_is_multicast(&remote->ipv6))
+ return 0;
+ err = ipv6_stub->ipv6_sock_mc_join(sk, ub->ifindex,
+ &remote->ipv6);
+#endif
+ }
+ return err;
+}
+
+/**
+ * parse_options - build local/remote addresses from configuration
+ * @attrs: netlink config data
+ * @ub: UDP bearer instance
+ * @local: local bearer IP address/port
+ * @remote: peer or multicast IP/port
+ */
+static int parse_options(struct nlattr *attrs[], struct udp_bearer *ub,
+ struct udp_media_addr *local,
+ struct udp_media_addr *remote)
+{
+ struct nlattr *opts[TIPC_NLA_UDP_MAX + 1];
+ struct sockaddr_storage *sa_local, *sa_remote;
+
+ if (!attrs[TIPC_NLA_BEARER_UDP_OPTS])
+ goto err;
+ if (nla_parse_nested(opts, TIPC_NLA_UDP_MAX,
+ attrs[TIPC_NLA_BEARER_UDP_OPTS],
+ tipc_nl_udp_policy))
+ goto err;
+ if (opts[TIPC_NLA_UDP_LOCAL] && opts[TIPC_NLA_UDP_REMOTE]) {
+ sa_local = nla_data(opts[TIPC_NLA_UDP_LOCAL]);
+ sa_remote = nla_data(opts[TIPC_NLA_UDP_REMOTE]);
+ } else {
+err:
+ pr_err("Invalid UDP bearer configuration");
+ return -EINVAL;
+ }
+ if ((sa_local->ss_family & sa_remote->ss_family) == AF_INET) {
+ struct sockaddr_in *ip4;
+
+ ip4 = (struct sockaddr_in *)sa_local;
+ local->proto = htons(ETH_P_IP);
+ local->udp_port = ip4->sin_port;
+ local->ipv4.s_addr = ip4->sin_addr.s_addr;
+
+ ip4 = (struct sockaddr_in *)sa_remote;
+ remote->proto = htons(ETH_P_IP);
+ remote->udp_port = ip4->sin_port;
+ remote->ipv4.s_addr = ip4->sin_addr.s_addr;
+ return 0;
+
+#if IS_ENABLED(CONFIG_IPV6)
+ } else if ((sa_local->ss_family & sa_remote->ss_family) == AF_INET6) {
+ struct sockaddr_in6 *ip6;
+
+ ip6 = (struct sockaddr_in6 *)sa_local;
+ local->proto = htons(ETH_P_IPV6);
+ local->udp_port = ip6->sin6_port;
+ local->ipv6 = ip6->sin6_addr;
+ ub->ifindex = ip6->sin6_scope_id;
+
+ ip6 = (struct sockaddr_in6 *)sa_remote;
+ remote->proto = htons(ETH_P_IPV6);
+ remote->udp_port = ip6->sin6_port;
+ remote->ipv6 = ip6->sin6_addr;
+ return 0;
+#endif
+ }
+ return -EADDRNOTAVAIL;
+}
+
+/**
+ * tipc_udp_enable - callback to create a new udp bearer instance
+ * @net: network namespace
+ * @b: pointer to generic tipc_bearer
+ * @attrs: netlink bearer configuration
+ *
+ * validate the bearer parameters and initialize the udp bearer
+ * rtnl_lock should be held
+ */
+static int tipc_udp_enable(struct net *net, struct tipc_bearer *b,
+ struct nlattr *attrs[])
+{
+ int err = -EINVAL;
+ struct udp_bearer *ub;
+ struct udp_media_addr *remote;
+ struct udp_media_addr local = {0};
+ struct udp_port_cfg udp_conf = {0};
+ struct udp_tunnel_sock_cfg tuncfg = {NULL};
+
+ ub = kzalloc(sizeof(*ub), GFP_ATOMIC);
+ if (!ub)
+ return -ENOMEM;
+
+ remote = (struct udp_media_addr *)&b->bcast_addr.value;
+ memset(remote, 0, sizeof(struct udp_media_addr));
+ err = parse_options(attrs, ub, &local, remote);
+ if (err)
+ goto err;
+
+ b->bcast_addr.media_id = TIPC_MEDIA_TYPE_UDP;
+ b->bcast_addr.broadcast = 1;
+ rcu_assign_pointer(b->media_ptr, ub);
+ rcu_assign_pointer(ub->bearer, b);
+ tipc_udp_media_addr_set(&b->addr, &local);
+ if (local.proto == htons(ETH_P_IP)) {
+ struct net_device *dev;
+
+ dev = __ip_dev_find(net, local.ipv4.s_addr, false);
+ if (!dev) {
+ err = -ENODEV;
+ goto err;
+ }
+ udp_conf.family = AF_INET;
+ udp_conf.local_ip.s_addr = htonl(INADDR_ANY);
+ udp_conf.use_udp_checksums = false;
+ ub->ifindex = dev->ifindex;
+ b->mtu = dev->mtu - sizeof(struct iphdr)
+ - sizeof(struct udphdr);
+#if IS_ENABLED(CONFIG_IPV6)
+ } else if (local.proto == htons(ETH_P_IPV6)) {
+ udp_conf.family = AF_INET6;
+ udp_conf.use_udp6_tx_checksums = true;
+ udp_conf.use_udp6_rx_checksums = true;
+ udp_conf.local_ip6 = in6addr_any;
+ b->mtu = 1280;
+#endif
+ } else {
+ err = -EAFNOSUPPORT;
+ goto err;
+ }
+ udp_conf.local_udp_port = local.udp_port;
+ err = udp_sock_create(net, &udp_conf, &ub->ubsock);
+ if (err)
+ goto err;
+ tuncfg.sk_user_data = ub;
+ tuncfg.encap_type = 1;
+ tuncfg.encap_rcv = tipc_udp_recv;
+ tuncfg.encap_destroy = NULL;
+ setup_udp_tunnel_sock(net, ub->ubsock, &tuncfg);
+
+ if (enable_mcast(ub, remote))
+ goto err;
+ return 0;
+err:
+ kfree(ub);
+ return err;
+}
+
+/* cleanup_bearer - break the socket/bearer association */
+static void cleanup_bearer(struct work_struct *work)
+{
+ struct udp_bearer *ub = container_of(work, struct udp_bearer, work);
+
+ if (ub->ubsock)
+ udp_tunnel_sock_release(ub->ubsock);
+ synchronize_net();
+ kfree(ub);
+}
+
+/* tipc_udp_disable - detach bearer from socket */
+static void tipc_udp_disable(struct tipc_bearer *b)
+{
+ struct udp_bearer *ub;
+
+ ub = rcu_dereference_rtnl(b->media_ptr);
+ if (!ub) {
+ pr_err("UDP bearer instance not found\n");
+ return;
+ }
+ if (ub->ubsock)
+ sock_set_flag(ub->ubsock->sk, SOCK_DEAD);
+ RCU_INIT_POINTER(b->media_ptr, NULL);
+ RCU_INIT_POINTER(ub->bearer, NULL);
+
+ /* sock_release need to be done outside of rtnl lock */
+ INIT_WORK(&ub->work, cleanup_bearer);
+ schedule_work(&ub->work);
+}
+
+struct tipc_media udp_media_info = {
+ .send_msg = tipc_udp_send_msg,
+ .enable_media = tipc_udp_enable,
+ .disable_media = tipc_udp_disable,
+ .addr2str = tipc_udp_addr2str,
+ .addr2msg = tipc_udp_addr2msg,
+ .msg2addr = tipc_udp_msg2addr,
+ .priority = TIPC_DEF_LINK_PRI,
+ .tolerance = TIPC_DEF_LINK_TOL,
+ .window = TIPC_DEF_LINK_WIN,
+ .type_id = TIPC_MEDIA_TYPE_UDP,
+ .hwaddr_len = 0,
+ .name = "udp"
+};
Powered by cgit, Gruvboxed by Blaster4385.