Skip to content

Commit

Permalink
tipc: connection oriented transport uses new send functions
Browse files Browse the repository at this point in the history 
We move the message sending across established connections
to use the message preparation and send functions introduced
earlier in this series. We now do the message preparation
and call to the link send function directly from the socket,
instead of going via the port layer.

As a consequence of this change, the functions tipc_send(),
tipc_port_iovec_rcv(), tipc_port_iovec_reject() and tipc_reject_msg()
become unreferenced and can be eliminated from port.c. For the same
reason, the functions tipc_link_xmit_fast(), tipc_link_iovec_xmit_long()
and tipc_link_iovec_fast() can be eliminated from link.c.

Signed-off-by: Jon Maloy <jon.maloy@ericsson.com>
Reviewed-by: Erik Hugne <erik.hugne@ericsson.com>
Reviewed-by: Ying Xue <ying.xue@windriver.com>
Signed-off-by: David S. Miller <davem@davemloft.net>
  • Loading branch information
Jon Paul Maloy authored and David S. Miller committed Jun 27, 2014
1 parent e2dafe8 commit 4ccfe5e
Show file tree
Hide file tree
Showing 4 changed files with 70 additions and 513 deletions.
249 changes: 0 additions & 249 deletions net/tipc/link.c
View file
Original file line number Diff line number Diff line change
Expand Up @@ -82,9 +82,6 @@ static void tipc_link_proto_rcv(struct tipc_link *l_ptr, struct sk_buff *buf);
static int tipc_link_tunnel_rcv(struct tipc_node *n_ptr,
struct sk_buff **buf);
static void link_set_supervision_props(struct tipc_link *l_ptr, u32 tolerance);
static int tipc_link_iovec_long_xmit(struct tipc_port *sender,
struct iovec const *msg_sect,
unsigned int len, u32 destnode);
static void link_state_event(struct tipc_link *l_ptr, u32 event);
static void link_reset_statistics(struct tipc_link *l_ptr);
static void link_print(struct tipc_link *l_ptr, const char *str);
Expand Down Expand Up @@ -1070,252 +1067,6 @@ void tipc_link_names_xmit(struct list_head *message_list, u32 dest)
}
}

/*
* tipc_link_xmit_fast: Entry for data messages where the
* destination link is known and the header is complete,
* inclusive total message length. Very time critical.
* Link is locked. Returns user data length.
*/
static int tipc_link_xmit_fast(struct tipc_link *l_ptr, struct sk_buff *buf,
u32 *used_max_pkt)
{
struct tipc_msg *msg = buf_msg(buf);
int res = msg_data_sz(msg);

if (likely(!link_congested(l_ptr))) {
if (likely(msg_size(msg) <= l_ptr->max_pkt)) {
link_add_to_outqueue(l_ptr, buf, msg);
tipc_bearer_send(l_ptr->bearer_id, buf,
&l_ptr->media_addr);
l_ptr->unacked_window = 0;
return res;
}
else
*used_max_pkt = l_ptr->max_pkt;
}
return __tipc_link_xmit(l_ptr, buf); /* All other cases */
}

/*
* tipc_link_iovec_xmit_fast: Entry for messages where the
* destination processor is known and the header is complete,
* except for total message length.
* Returns user data length or errno.
*/
int tipc_link_iovec_xmit_fast(struct tipc_port *sender,
struct iovec const *msg_sect,
unsigned int len, u32 destaddr)
{
struct tipc_msg *hdr = &sender->phdr;
struct tipc_link *l_ptr;
struct sk_buff *buf;
struct tipc_node *node;
int res;
u32 selector = msg_origport(hdr) & 1;

again:
/*
* Try building message using port's max_pkt hint.
* (Must not hold any locks while building message.)
*/
res = tipc_msg_build(hdr, msg_sect, len, sender->max_pkt, &buf);
/* Exit if build request was invalid */
if (unlikely(res < 0))
return res;

node = tipc_node_find(destaddr);
if (likely(node)) {
tipc_node_lock(node);
l_ptr = node->active_links[selector];
if (likely(l_ptr)) {
if (likely(buf)) {
res = tipc_link_xmit_fast(l_ptr, buf,
&sender->max_pkt);
exit:
tipc_node_unlock(node);
return res;
}

/* Exit if link (or bearer) is congested */
if (link_congested(l_ptr)) {
res = link_schedule_port(l_ptr,
sender->ref, res);
goto exit;
}

/*
* Message size exceeds max_pkt hint; update hint,
* then re-try fast path or fragment the message
*/
sender->max_pkt = l_ptr->max_pkt;
tipc_node_unlock(node);


if ((msg_hdr_sz(hdr) + res) <= sender->max_pkt)
goto again;

return tipc_link_iovec_long_xmit(sender, msg_sect,
len, destaddr);
}
tipc_node_unlock(node);
}

/* Couldn't find a link to the destination node */
kfree_skb(buf);
tipc_port_iovec_reject(sender, hdr, msg_sect, len, TIPC_ERR_NO_NODE);
return -ENETUNREACH;
}

/*
* tipc_link_iovec_long_xmit(): Entry for long messages where the
* destination node is known and the header is complete,
* inclusive total message length.
* Link and bearer congestion status have been checked to be ok,
* and are ignored if they change.
*
* Note that fragments do not use the full link MTU so that they won't have
* to undergo refragmentation if link changeover causes them to be sent
* over another link with an additional tunnel header added as prefix.
* (Refragmentation will still occur if the other link has a smaller MTU.)
*
* Returns user data length or errno.
*/
static int tipc_link_iovec_long_xmit(struct tipc_port *sender,
struct iovec const *msg_sect,
unsigned int len, u32 destaddr)
{
struct tipc_link *l_ptr;
struct tipc_node *node;
struct tipc_msg *hdr = &sender->phdr;
u32 dsz = len;
u32 max_pkt, fragm_sz, rest;
struct tipc_msg fragm_hdr;
struct sk_buff *buf, *buf_chain, *prev;
u32 fragm_crs, fragm_rest, hsz, sect_rest;
const unchar __user *sect_crs;
int curr_sect;
u32 fragm_no;
int res = 0;

again:
fragm_no = 1;
max_pkt = sender->max_pkt - INT_H_SIZE;
/* leave room for tunnel header in case of link changeover */
fragm_sz = max_pkt - INT_H_SIZE;
/* leave room for fragmentation header in each fragment */
rest = dsz;
fragm_crs = 0;
fragm_rest = 0;
sect_rest = 0;
sect_crs = NULL;
curr_sect = -1;

/* Prepare reusable fragment header */
tipc_msg_init(&fragm_hdr, MSG_FRAGMENTER, FIRST_FRAGMENT,
INT_H_SIZE, msg_destnode(hdr));
msg_set_size(&fragm_hdr, max_pkt);
msg_set_fragm_no(&fragm_hdr, 1);

/* Prepare header of first fragment */
buf_chain = buf = tipc_buf_acquire(max_pkt);
if (!buf)
return -ENOMEM;
buf->next = NULL;
skb_copy_to_linear_data(buf, &fragm_hdr, INT_H_SIZE);
hsz = msg_hdr_sz(hdr);
skb_copy_to_linear_data_offset(buf, INT_H_SIZE, hdr, hsz);

/* Chop up message */
fragm_crs = INT_H_SIZE + hsz;
fragm_rest = fragm_sz - hsz;

do { /* For all sections */
u32 sz;

if (!sect_rest) {
sect_rest = msg_sect[++curr_sect].iov_len;
sect_crs = msg_sect[curr_sect].iov_base;
}

if (sect_rest < fragm_rest)
sz = sect_rest;
else
sz = fragm_rest;

if (copy_from_user(buf->data + fragm_crs, sect_crs, sz)) {
res = -EFAULT;
error:
kfree_skb_list(buf_chain);
return res;
}
sect_crs += sz;
sect_rest -= sz;
fragm_crs += sz;
fragm_rest -= sz;
rest -= sz;

if (!fragm_rest && rest) {

/* Initiate new fragment: */
if (rest <= fragm_sz) {
fragm_sz = rest;
msg_set_type(&fragm_hdr, LAST_FRAGMENT);
} else {
msg_set_type(&fragm_hdr, FRAGMENT);
}
msg_set_size(&fragm_hdr, fragm_sz + INT_H_SIZE);
msg_set_fragm_no(&fragm_hdr, ++fragm_no);
prev = buf;
buf = tipc_buf_acquire(fragm_sz + INT_H_SIZE);
if (!buf) {
res = -ENOMEM;
goto error;
}

buf->next = NULL;
prev->next = buf;
skb_copy_to_linear_data(buf, &fragm_hdr, INT_H_SIZE);
fragm_crs = INT_H_SIZE;
fragm_rest = fragm_sz;
}
} while (rest > 0);

/*
* Now we have a buffer chain. Select a link and check
* that packet size is still OK
*/
node = tipc_node_find(destaddr);
if (likely(node)) {
tipc_node_lock(node);
l_ptr = node->active_links[sender->ref & 1];
if (!l_ptr) {
tipc_node_unlock(node);
goto reject;
}
if (l_ptr->max_pkt < max_pkt) {
sender->max_pkt = l_ptr->max_pkt;
tipc_node_unlock(node);
kfree_skb_list(buf_chain);
goto again;
}
} else {
reject:
kfree_skb_list(buf_chain);
tipc_port_iovec_reject(sender, hdr, msg_sect, len,
TIPC_ERR_NO_NODE);
return -ENETUNREACH;
}

/* Append chain of fragments to send queue & send them */
l_ptr->long_msg_seq_no++;
link_add_chain_to_outqueue(l_ptr, buf_chain, l_ptr->long_msg_seq_no);
l_ptr->stats.sent_fragments += fragm_no;
l_ptr->stats.sent_fragmented++;
tipc_link_push_queue(l_ptr);
tipc_node_unlock(node);
return dsz;
}

/*
* tipc_link_push_packet: Push one unsent packet to the media
*/
Expand Down
Loading

0 comments on commit 4ccfe5e

Please sign in to comment.