Commit 1d70ec70 authored by David S. Miller's avatar David S. Miller

Merge branch 'tipc'

Jon Maloy says:

====================
tipc: clean up socket message reception

Despite recent improvements the message reception code in socket.c is
perceived as obscure and hard to follow, especially regarding the logics
for message rejection. With the commits in this series we try to remedy
this situation.
====================
Signed-off-by: default avatarDavid S. Miller <davem@davemloft.net>
parents a69e5a0d cda3696d
...@@ -463,60 +463,72 @@ bool tipc_msg_make_bundle(struct sk_buff **skb, struct tipc_msg *msg, ...@@ -463,60 +463,72 @@ bool tipc_msg_make_bundle(struct sk_buff **skb, struct tipc_msg *msg,
/** /**
* tipc_msg_reverse(): swap source and destination addresses and add error code * tipc_msg_reverse(): swap source and destination addresses and add error code
* @buf: buffer containing message to be reversed * @own_node: originating node id for reversed message
* @dnode: return value: node where to send message after reversal * @skb: buffer containing message to be reversed; may be replaced.
* @err: error code to be set in message * @err: error code to be set in message, if any
* Consumes buffer if failure * Consumes buffer at failure
* Returns true if success, otherwise false * Returns true if success, otherwise false
*/ */
bool tipc_msg_reverse(u32 own_addr, struct sk_buff *buf, u32 *dnode, bool tipc_msg_reverse(u32 own_node, struct sk_buff **skb, int err)
int err)
{ {
struct tipc_msg *msg = buf_msg(buf); struct sk_buff *_skb = *skb;
struct tipc_msg *hdr = buf_msg(_skb);
struct tipc_msg ohdr; struct tipc_msg ohdr;
uint rdsz = min_t(uint, msg_data_sz(msg), MAX_FORWARD_SIZE); int dlen = min_t(uint, msg_data_sz(hdr), MAX_FORWARD_SIZE);
if (skb_linearize(buf)) if (skb_linearize(_skb))
goto exit; goto exit;
msg = buf_msg(buf); hdr = buf_msg(_skb);
if (msg_dest_droppable(msg)) if (msg_dest_droppable(hdr))
goto exit; goto exit;
if (msg_errcode(msg)) if (msg_errcode(hdr))
goto exit; goto exit;
memcpy(&ohdr, msg, msg_hdr_sz(msg));
msg_set_errcode(msg, err); /* Take a copy of original header before altering message */
msg_set_origport(msg, msg_destport(&ohdr)); memcpy(&ohdr, hdr, msg_hdr_sz(hdr));
msg_set_destport(msg, msg_origport(&ohdr));
msg_set_prevnode(msg, own_addr); /* Never return SHORT header; expand by replacing buffer if necessary */
if (!msg_short(msg)) { if (msg_short(hdr)) {
msg_set_orignode(msg, msg_destnode(&ohdr)); *skb = tipc_buf_acquire(BASIC_H_SIZE + dlen);
msg_set_destnode(msg, msg_orignode(&ohdr)); if (!*skb)
goto exit;
memcpy((*skb)->data + BASIC_H_SIZE, msg_data(hdr), dlen);
kfree_skb(_skb);
_skb = *skb;
hdr = buf_msg(_skb);
memcpy(hdr, &ohdr, BASIC_H_SIZE);
msg_set_hdr_sz(hdr, BASIC_H_SIZE);
} }
msg_set_size(msg, msg_hdr_sz(msg) + rdsz);
skb_trim(buf, msg_size(msg)); /* Now reverse the concerned fields */
skb_orphan(buf); msg_set_errcode(hdr, err);
*dnode = msg_orignode(&ohdr); msg_set_origport(hdr, msg_destport(&ohdr));
msg_set_destport(hdr, msg_origport(&ohdr));
msg_set_destnode(hdr, msg_prevnode(&ohdr));
msg_set_prevnode(hdr, own_node);
msg_set_orignode(hdr, own_node);
msg_set_size(hdr, msg_hdr_sz(hdr) + dlen);
skb_trim(_skb, msg_size(hdr));
skb_orphan(_skb);
return true; return true;
exit: exit:
kfree_skb(buf); kfree_skb(_skb);
*dnode = 0; *skb = NULL;
return false; return false;
} }
/** /**
* tipc_msg_lookup_dest(): try to find new destination for named message * tipc_msg_lookup_dest(): try to find new destination for named message
* @skb: the buffer containing the message. * @skb: the buffer containing the message.
* @dnode: return value: next-hop node, if destination found * @err: error code to be used by caller if lookup fails
* @err: return value: error code to use, if message to be rejected
* Does not consume buffer * Does not consume buffer
* Returns true if a destination is found, false otherwise * Returns true if a destination is found, false otherwise
*/ */
bool tipc_msg_lookup_dest(struct net *net, struct sk_buff *skb, bool tipc_msg_lookup_dest(struct net *net, struct sk_buff *skb, int *err)
u32 *dnode, int *err)
{ {
struct tipc_msg *msg = buf_msg(skb); struct tipc_msg *msg = buf_msg(skb);
u32 dport; u32 dport, dnode;
u32 own_addr = tipc_own_addr(net); u32 onode = tipc_own_addr(net);
if (!msg_isdata(msg)) if (!msg_isdata(msg))
return false; return false;
...@@ -529,15 +541,15 @@ bool tipc_msg_lookup_dest(struct net *net, struct sk_buff *skb, ...@@ -529,15 +541,15 @@ bool tipc_msg_lookup_dest(struct net *net, struct sk_buff *skb,
return false; return false;
if (msg_reroute_cnt(msg)) if (msg_reroute_cnt(msg))
return false; return false;
*dnode = addr_domain(net, msg_lookup_scope(msg)); dnode = addr_domain(net, msg_lookup_scope(msg));
dport = tipc_nametbl_translate(net, msg_nametype(msg), dport = tipc_nametbl_translate(net, msg_nametype(msg),
msg_nameinst(msg), dnode); msg_nameinst(msg), &dnode);
if (!dport) if (!dport)
return false; return false;
msg_incr_reroute_cnt(msg); msg_incr_reroute_cnt(msg);
if (*dnode != own_addr) if (dnode != onode)
msg_set_prevnode(msg, own_addr); msg_set_prevnode(msg, onode);
msg_set_destnode(msg, *dnode); msg_set_destnode(msg, dnode);
msg_set_destport(msg, dport); msg_set_destport(msg, dport);
*err = TIPC_OK; *err = TIPC_OK;
return true; return true;
......
...@@ -785,8 +785,7 @@ static inline bool msg_peer_is_up(struct tipc_msg *m) ...@@ -785,8 +785,7 @@ static inline bool msg_peer_is_up(struct tipc_msg *m)
struct sk_buff *tipc_buf_acquire(u32 size); struct sk_buff *tipc_buf_acquire(u32 size);
bool tipc_msg_validate(struct sk_buff *skb); bool tipc_msg_validate(struct sk_buff *skb);
bool tipc_msg_reverse(u32 own_addr, struct sk_buff *buf, u32 *dnode, bool tipc_msg_reverse(u32 own_addr, struct sk_buff **skb, int err);
int err);
void tipc_msg_init(u32 own_addr, struct tipc_msg *m, u32 user, u32 type, void tipc_msg_init(u32 own_addr, struct tipc_msg *m, u32 user, u32 type,
u32 hsize, u32 destnode); u32 hsize, u32 destnode);
struct sk_buff *tipc_msg_create(uint user, uint type, uint hdr_sz, struct sk_buff *tipc_msg_create(uint user, uint type, uint hdr_sz,
...@@ -799,8 +798,7 @@ bool tipc_msg_make_bundle(struct sk_buff **skb, struct tipc_msg *msg, ...@@ -799,8 +798,7 @@ bool tipc_msg_make_bundle(struct sk_buff **skb, struct tipc_msg *msg,
bool tipc_msg_extract(struct sk_buff *skb, struct sk_buff **iskb, int *pos); 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 tipc_msg_build(struct tipc_msg *mhdr, struct msghdr *m,
int offset, int dsz, int mtu, struct sk_buff_head *list); int offset, int dsz, int mtu, struct sk_buff_head *list);
bool tipc_msg_lookup_dest(struct net *net, struct sk_buff *skb, u32 *dnode, bool tipc_msg_lookup_dest(struct net *net, struct sk_buff *skb, int *err);
int *err);
struct sk_buff *tipc_msg_reassemble(struct sk_buff_head *list); struct sk_buff *tipc_msg_reassemble(struct sk_buff_head *list);
static inline u16 buf_seqno(struct sk_buff *skb) static inline u16 buf_seqno(struct sk_buff *skb)
......
...@@ -248,6 +248,22 @@ static void tsk_advance_rx_queue(struct sock *sk) ...@@ -248,6 +248,22 @@ static void tsk_advance_rx_queue(struct sock *sk)
kfree_skb(__skb_dequeue(&sk->sk_receive_queue)); kfree_skb(__skb_dequeue(&sk->sk_receive_queue));
} }
/* tipc_sk_respond() : send response message back to sender
*/
static void tipc_sk_respond(struct sock *sk, struct sk_buff *skb, int err)
{
u32 selector;
u32 dnode;
u32 onode = tipc_own_addr(sock_net(sk));
if (!tipc_msg_reverse(onode, &skb, err))
return;
dnode = msg_destnode(buf_msg(skb));
selector = msg_origport(buf_msg(skb));
tipc_node_xmit_skb(sock_net(sk), skb, dnode, selector);
}
/** /**
* tsk_rej_rx_queue - reject all buffers in socket receive queue * tsk_rej_rx_queue - reject all buffers in socket receive queue
* *
...@@ -256,13 +272,9 @@ static void tsk_advance_rx_queue(struct sock *sk) ...@@ -256,13 +272,9 @@ static void tsk_advance_rx_queue(struct sock *sk)
static void tsk_rej_rx_queue(struct sock *sk) static void tsk_rej_rx_queue(struct sock *sk)
{ {
struct sk_buff *skb; struct sk_buff *skb;
u32 dnode;
u32 own_node = tsk_own_node(tipc_sk(sk));
while ((skb = __skb_dequeue(&sk->sk_receive_queue))) { while ((skb = __skb_dequeue(&sk->sk_receive_queue)))
if (tipc_msg_reverse(own_node, skb, &dnode, TIPC_ERR_NO_PORT)) tipc_sk_respond(sk, skb, TIPC_ERR_NO_PORT);
tipc_node_xmit_skb(sock_net(sk), skb, dnode, 0);
}
} }
/* tsk_peer_msg - verify if message was sent by connected port's peer /* tsk_peer_msg - verify if message was sent by connected port's peer
...@@ -441,9 +453,7 @@ static int tipc_release(struct socket *sock) ...@@ -441,9 +453,7 @@ static int tipc_release(struct socket *sock)
tsk->connected = 0; tsk->connected = 0;
tipc_node_remove_conn(net, dnode, tsk->portid); tipc_node_remove_conn(net, dnode, tsk->portid);
} }
if (tipc_msg_reverse(tsk_own_node(tsk), skb, &dnode, tipc_sk_respond(sk, skb, TIPC_ERR_NO_PORT);
TIPC_ERR_NO_PORT))
tipc_node_xmit_skb(net, skb, dnode, 0);
} }
} }
...@@ -764,35 +774,35 @@ void tipc_sk_mcast_rcv(struct net *net, struct sk_buff_head *arrvq, ...@@ -764,35 +774,35 @@ void tipc_sk_mcast_rcv(struct net *net, struct sk_buff_head *arrvq,
/** /**
* tipc_sk_proto_rcv - receive a connection mng protocol message * tipc_sk_proto_rcv - receive a connection mng protocol message
* @tsk: receiving socket * @tsk: receiving socket
* @skb: pointer to message buffer. Set to NULL if buffer is consumed. * @skb: pointer to message buffer.
*/ */
static void tipc_sk_proto_rcv(struct tipc_sock *tsk, struct sk_buff **skb) static void tipc_sk_proto_rcv(struct tipc_sock *tsk, struct sk_buff *skb)
{ {
struct tipc_msg *msg = buf_msg(*skb); struct sock *sk = &tsk->sk;
struct tipc_msg *hdr = buf_msg(skb);
int mtyp = msg_type(hdr);
int conn_cong; int conn_cong;
u32 dnode;
u32 own_node = tsk_own_node(tsk);
/* Ignore if connection cannot be validated: */ /* Ignore if connection cannot be validated: */
if (!tsk_peer_msg(tsk, msg)) if (!tsk_peer_msg(tsk, hdr))
goto exit; goto exit;
tsk->probing_state = TIPC_CONN_OK; tsk->probing_state = TIPC_CONN_OK;
if (msg_type(msg) == CONN_ACK) { if (mtyp == CONN_PROBE) {
msg_set_type(hdr, CONN_PROBE_REPLY);
tipc_sk_respond(sk, skb, TIPC_OK);
return;
} else if (mtyp == CONN_ACK) {
conn_cong = tsk_conn_cong(tsk); conn_cong = tsk_conn_cong(tsk);
tsk->sent_unacked -= msg_msgcnt(msg); tsk->sent_unacked -= msg_msgcnt(hdr);
if (conn_cong) if (conn_cong)
tsk->sk.sk_write_space(&tsk->sk); sk->sk_write_space(sk);
} else if (msg_type(msg) == CONN_PROBE) { } else if (mtyp != CONN_PROBE_REPLY) {
if (tipc_msg_reverse(own_node, *skb, &dnode, TIPC_OK)) { pr_warn("Received unknown CONN_PROTO msg\n");
msg_set_type(msg, CONN_PROBE_REPLY);
return;
}
} }
/* Do nothing if msg_type() == CONN_PROBE_REPLY */
exit: exit:
kfree_skb(*skb); kfree_skb(skb);
*skb = NULL;
} }
static int tipc_wait_for_sndmsg(struct socket *sock, long *timeo_p) static int tipc_wait_for_sndmsg(struct socket *sock, long *timeo_p)
...@@ -1510,82 +1520,81 @@ static void tipc_data_ready(struct sock *sk) ...@@ -1510,82 +1520,81 @@ static void tipc_data_ready(struct sock *sk)
* @tsk: TIPC socket * @tsk: TIPC socket
* @skb: pointer to message buffer. Set to NULL if buffer is consumed * @skb: pointer to message buffer. Set to NULL if buffer is consumed
* *
* Returns 0 (TIPC_OK) if everything ok, -TIPC_ERR_NO_PORT otherwise * Returns true if everything ok, false otherwise
*/ */
static int filter_connect(struct tipc_sock *tsk, struct sk_buff **skb) static bool filter_connect(struct tipc_sock *tsk, struct sk_buff *skb)
{ {
struct sock *sk = &tsk->sk; struct sock *sk = &tsk->sk;
struct net *net = sock_net(sk); struct net *net = sock_net(sk);
struct socket *sock = sk->sk_socket; struct socket *sock = sk->sk_socket;
struct tipc_msg *msg = buf_msg(*skb); struct tipc_msg *hdr = buf_msg(skb);
int retval = -TIPC_ERR_NO_PORT;
if (msg_mcast(msg)) if (unlikely(msg_mcast(hdr)))
return retval; return false;
switch ((int)sock->state) { switch ((int)sock->state) {
case SS_CONNECTED: case SS_CONNECTED:
/* Accept only connection-based messages sent by peer */ /* Accept only connection-based messages sent by peer */
if (tsk_peer_msg(tsk, msg)) { if (unlikely(!tsk_peer_msg(tsk, hdr)))
if (unlikely(msg_errcode(msg))) { return false;
sock->state = SS_DISCONNECTING;
tsk->connected = 0; if (unlikely(msg_errcode(hdr))) {
/* let timer expire on it's own */ sock->state = SS_DISCONNECTING;
tipc_node_remove_conn(net, tsk_peer_node(tsk), tsk->connected = 0;
tsk->portid); /* Let timer expire on it's own */
} tipc_node_remove_conn(net, tsk_peer_node(tsk),
retval = TIPC_OK; tsk->portid);
} }
break; return true;
case SS_CONNECTING: case SS_CONNECTING:
/* Accept only ACK or NACK message */
if (unlikely(!msg_connected(msg))) /* Accept only ACK or NACK message */
break; if (unlikely(!msg_connected(hdr)))
return false;
if (unlikely(msg_errcode(msg))) { if (unlikely(msg_errcode(hdr))) {
sock->state = SS_DISCONNECTING; sock->state = SS_DISCONNECTING;
sk->sk_err = ECONNREFUSED; sk->sk_err = ECONNREFUSED;
retval = TIPC_OK; return true;
break;
} }
if (unlikely(msg_importance(msg) > TIPC_CRITICAL_IMPORTANCE)) { if (unlikely(!msg_isdata(hdr))) {
sock->state = SS_DISCONNECTING; sock->state = SS_DISCONNECTING;
sk->sk_err = EINVAL; sk->sk_err = EINVAL;
retval = TIPC_OK; return true;
break;
} }
tipc_sk_finish_conn(tsk, msg_origport(msg), msg_orignode(msg)); tipc_sk_finish_conn(tsk, msg_origport(hdr), msg_orignode(hdr));
msg_set_importance(&tsk->phdr, msg_importance(msg)); msg_set_importance(&tsk->phdr, msg_importance(hdr));
sock->state = SS_CONNECTED; sock->state = SS_CONNECTED;
/* If an incoming message is an 'ACK-', it should be /* If 'ACK+' message, add to socket receive queue */
* discarded here because it doesn't contain useful if (msg_data_sz(hdr))
* data. In addition, we should try to wake up return true;
* connect() routine if sleeping.
*/ /* If empty 'ACK-' message, wake up sleeping connect() */
if (msg_data_sz(msg) == 0) { if (waitqueue_active(sk_sleep(sk)))
kfree_skb(*skb); wake_up_interruptible(sk_sleep(sk));
*skb = NULL;
if (waitqueue_active(sk_sleep(sk))) /* 'ACK-' message is neither accepted nor rejected: */
wake_up_interruptible(sk_sleep(sk)); msg_set_dest_droppable(hdr, 1);
} return false;
retval = TIPC_OK;
break;
case SS_LISTENING: case SS_LISTENING:
case SS_UNCONNECTED: case SS_UNCONNECTED:
/* Accept only SYN message */ /* Accept only SYN message */
if (!msg_connected(msg) && !(msg_errcode(msg))) if (!msg_connected(hdr) && !(msg_errcode(hdr)))
retval = TIPC_OK; return true;
break; break;
case SS_DISCONNECTING: case SS_DISCONNECTING:
break; break;
default: default:
pr_err("Unknown socket state %u\n", sock->state); pr_err("Unknown socket state %u\n", sock->state);
} }
return retval; return false;
} }
/** /**
...@@ -1620,61 +1629,70 @@ static unsigned int rcvbuf_limit(struct sock *sk, struct sk_buff *buf) ...@@ -1620,61 +1629,70 @@ static unsigned int rcvbuf_limit(struct sock *sk, struct sk_buff *buf)
/** /**
* filter_rcv - validate incoming message * filter_rcv - validate incoming message
* @sk: socket * @sk: socket
* @skb: pointer to message. Set to NULL if buffer is consumed. * @skb: pointer to message.
* *
* Enqueues message on receive queue if acceptable; optionally handles * Enqueues message on receive queue if acceptable; optionally handles
* disconnect indication for a connected socket. * disconnect indication for a connected socket.
* *
* Called with socket lock already taken * Called with socket lock already taken
* *
* Returns 0 (TIPC_OK) if message was ok, -TIPC error code if rejected * Returns true if message was added to socket receive queue, otherwise false
*/ */
static int filter_rcv(struct sock *sk, struct sk_buff **skb) static bool filter_rcv(struct sock *sk, struct sk_buff *skb)
{ {
struct socket *sock = sk->sk_socket; struct socket *sock = sk->sk_socket;
struct tipc_sock *tsk = tipc_sk(sk); struct tipc_sock *tsk = tipc_sk(sk);
struct tipc_msg *msg = buf_msg(*skb); struct tipc_msg *hdr = buf_msg(skb);
unsigned int limit = rcvbuf_limit(sk, *skb); unsigned int limit = rcvbuf_limit(sk, skb);
int rc = TIPC_OK; int err = TIPC_OK;
int usr = msg_user(hdr);
if (unlikely(msg_user(msg) == CONN_MANAGER)) { if (unlikely(msg_user(hdr) == CONN_MANAGER)) {
tipc_sk_proto_rcv(tsk, skb); tipc_sk_proto_rcv(tsk, skb);
return TIPC_OK; return false;
} }
if (unlikely(msg_user(msg) == SOCK_WAKEUP)) { if (unlikely(usr == SOCK_WAKEUP)) {
kfree_skb(*skb); kfree_skb(skb);
tsk->link_cong = 0; tsk->link_cong = 0;
sk->sk_write_space(sk); sk->sk_write_space(sk);
*skb = NULL; return false;
return TIPC_OK;
} }
/* Reject message if it is wrong sort of message for socket */ /* Drop if illegal message type */
if (msg_type(msg) > TIPC_DIRECT_MSG) if (unlikely(msg_type(hdr) > TIPC_DIRECT_MSG)) {
return -TIPC_ERR_NO_PORT; kfree_skb(skb);
return false;
}
if (sock->state == SS_READY) { /* Reject if wrong message type for current socket state */
if (msg_connected(msg)) if (unlikely(sock->state == SS_READY)) {
return -TIPC_ERR_NO_PORT; if (msg_connected(hdr)) {
} else { err = TIPC_ERR_NO_PORT;
rc = filter_connect(tsk, skb); goto reject;
if (rc != TIPC_OK || !*skb) }
return rc; } else if (unlikely(!filter_connect(tsk, skb))) {
err = TIPC_ERR_NO_PORT;
goto reject;
} }
/* Reject message if there isn't room to queue it */ /* Reject message if there isn't room to queue it */
if (sk_rmem_alloc_get(sk) + (*skb)->truesize >= limit) if (unlikely(sk_rmem_alloc_get(sk) + skb->truesize >= limit)) {
return -TIPC_ERR_OVERLOAD; err = TIPC_ERR_OVERLOAD;
goto reject;
}
/* Enqueue message */ /* Enqueue message */
TIPC_SKB_CB(*skb)->handle = NULL; TIPC_SKB_CB(skb)->handle = NULL;
__skb_queue_tail(&sk->sk_receive_queue, *skb); __skb_queue_tail(&sk->sk_receive_queue, skb);
skb_set_owner_r(*skb, sk); skb_set_owner_r(skb, sk);
sk->sk_data_ready(sk); sk->sk_data_ready(sk);
*skb = NULL; return true;
return TIPC_OK;
reject:
tipc_sk_respond(sk, skb, err);
return false;
} }
/** /**
...@@ -1688,22 +1706,10 @@ static int filter_rcv(struct sock *sk, struct sk_buff **skb) ...@@ -1688,22 +1706,10 @@ static int filter_rcv(struct sock *sk, struct sk_buff **skb)
*/ */
static int tipc_backlog_rcv(struct sock *sk, struct sk_buff *skb) static int tipc_backlog_rcv(struct sock *sk, struct sk_buff *skb)
{ {
int err; unsigned int truesize = skb->truesize;
atomic_t *dcnt;
u32 dnode;
struct tipc_sock *tsk = tipc_sk(sk);
struct net *net = sock_net(sk);
uint truesize = skb->truesize;
err = filter_rcv(sk, &skb); if (likely(filter_rcv(sk, skb)))
if (likely(!skb)) { atomic_add(truesize, &tipc_sk(sk)->dupl_rcvcnt);
dcnt = &tsk->dupl_rcvcnt;
if (atomic_read(dcnt) < TIPC_CONN_OVERLOAD_LIMIT)
atomic_add(truesize, dcnt);
return 0;
}
if (!err || tipc_msg_reverse(tsk_own_node(tsk), skb, &dnode, -err))
tipc_node_xmit_skb(net, skb, dnode, tsk->portid);
return 0; return 0;
} }
...@@ -1713,45 +1719,43 @@ static int tipc_backlog_rcv(struct sock *sk, struct sk_buff *skb) ...@@ -1713,45 +1719,43 @@ static int tipc_backlog_rcv(struct sock *sk, struct sk_buff *skb)
* @inputq: list of incoming buffers with potentially different destinations * @inputq: list of incoming buffers with potentially different destinations
* @sk: socket where the buffers should be enqueued * @sk: socket where the buffers should be enqueued
* @dport: port number for the socket * @dport: port number for the socket
* @_skb: returned buffer to be forwarded or rejected, if applicable
* *
* Caller must hold socket lock * Caller must hold socket lock
*
* Returns TIPC_OK if all buffers enqueued, otherwise -TIPC_ERR_OVERLOAD
* or -TIPC_ERR_NO_PORT
*/ */
static int tipc_sk_enqueue(struct sk_buff_head *inputq, struct sock *sk, static void tipc_sk_enqueue(struct sk_buff_head *inputq, struct sock *sk,
u32 dport, struct sk_buff **_skb) u32 dport)
{ {
unsigned int lim; unsigned int lim;
atomic_t *dcnt; atomic_t *dcnt;
int err;
struct sk_buff *skb; struct sk_buff *skb;
unsigned long time_limit = jiffies + 2; unsigned long time_limit = jiffies + 2;
while (skb_queue_len(inputq)) { while (skb_queue_len(inputq)) {
if (unlikely(time_after_eq(jiffies, time_limit))) if (unlikely(time_after_eq(jiffies, time_limit)))
return TIPC_OK; return;
skb = tipc_skb_dequeue(inputq, dport); skb = tipc_skb_dequeue(inputq, dport);
if (unlikely(!skb)) if (unlikely(!skb))
return TIPC_OK; return;
/* Add message directly to receive queue if possible */
if (!sock_owned_by_user(sk)) { if (!sock_owned_by_user(sk)) {
err = filter_rcv(sk, &skb); filter_rcv(sk, skb);
if (likely(!skb)) continue;
continue;
*_skb = skb;
return err;
} }
/* Try backlog, compensating for double-counted bytes */
dcnt = &tipc_sk(sk)->dupl_rcvcnt; dcnt = &tipc_sk(sk)->dupl_rcvcnt;
if (sk->sk_backlog.len) if (sk->sk_backlog.len)
atomic_set(dcnt, 0); atomic_set(dcnt, 0);
lim = rcvbuf_limit(sk, skb) + atomic_read(dcnt); lim = rcvbuf_limit(sk, skb) + atomic_read(dcnt);
if (likely(!sk_add_backlog(sk, skb, lim))) if (likely(!sk_add_backlog(sk, skb, lim)))
continue; continue;
*_skb = skb;
return -TIPC_ERR_OVERLOAD; /* Overload => reject message back to sender */
tipc_sk_respond(sk, skb, TIPC_ERR_OVERLOAD);
break;
} }
return TIPC_OK;
} }
/** /**
...@@ -1759,49 +1763,46 @@ static int tipc_sk_enqueue(struct sk_buff_head *inputq, struct sock *sk, ...@@ -1759,49 +1763,46 @@ static int tipc_sk_enqueue(struct sk_buff_head *inputq, struct sock *sk,
* @inputq: buffer list containing the buffers * @inputq: buffer list containing the buffers
* Consumes all buffers in list until inputq is empty * Consumes all buffers in list until inputq is empty
* Note: may be called in multiple threads referring to the same queue * Note: may be called in multiple threads referring to the same queue
* Returns 0 if last buffer was accepted, otherwise -EHOSTUNREACH
* Only node local calls check the return value, sending single-buffer queues
*/ */
int tipc_sk_rcv(struct net *net, struct sk_buff_head *inputq) void tipc_sk_rcv(struct net *net, struct sk_buff_head *inputq)
{ {
u32 dnode, dport = 0; u32 dnode, dport = 0;
int err; int err;
struct sk_buff *skb;
struct tipc_sock *tsk; struct tipc_sock *tsk;
struct tipc_net *tn;
struct sock *sk; struct sock *sk;
struct sk_buff *skb;
while (skb_queue_len(inputq)) { while (skb_queue_len(inputq)) {
err = -TIPC_ERR_NO_PORT;
skb = NULL;
dport = tipc_skb_peek_port(inputq, dport); dport = tipc_skb_peek_port(inputq, dport);
tsk = tipc_sk_lookup(net, dport); tsk = tipc_sk_lookup(net, dport);
if (likely(tsk)) { if (likely(tsk)) {
sk = &tsk->sk; sk = &tsk->sk;
if (likely(spin_trylock_bh(&sk->sk_lock.slock))) { if (likely(spin_trylock_bh(&sk->sk_lock.slock))) {
err = tipc_sk_enqueue(inputq, sk, dport, &skb); tipc_sk_enqueue(inputq, sk, dport);
spin_unlock_bh(&sk->sk_lock.slock); spin_unlock_bh(&sk->sk_lock.slock);
dport = 0;
} }
sock_put(sk); sock_put(sk);
} else {
skb = tipc_skb_dequeue(inputq, dport);
}
if (likely(!skb))
continue; continue;
if (tipc_msg_lookup_dest(net, skb, &dnode, &err))
goto xmit;
if (!err) {
dnode = msg_destnode(buf_msg(skb));
goto xmit;
} }
tn = net_generic(net, tipc_net_id);
if (!tipc_msg_reverse(tn->own_addr, skb, &dnode, -err)) /* No destination socket => dequeue skb if still there */
skb = tipc_skb_dequeue(inputq, dport);
if (!skb)
return;
/* Try secondary lookup if unresolved named message */
err = TIPC_ERR_NO_PORT;
if (tipc_msg_lookup_dest(net, skb, &err))
goto xmit;
/* Prepare for message rejection */
if (!tipc_msg_reverse(tipc_own_addr(net), &skb, err))
continue; continue;
xmit: xmit:
dnode = msg_destnode(buf_msg(skb));
tipc_node_xmit_skb(net, skb, dnode, dport); tipc_node_xmit_skb(net, skb, dnode, dport);
} }
return err ? -EHOSTUNREACH : 0;
} }
static int tipc_wait_for_connect(struct socket *sock, long *timeo_p) static int tipc_wait_for_connect(struct socket *sock, long *timeo_p)
...@@ -2070,7 +2071,10 @@ static int tipc_shutdown(struct socket *sock, int how) ...@@ -2070,7 +2071,10 @@ static int tipc_shutdown(struct socket *sock, int how)
struct net *net = sock_net(sk); struct net *net = sock_net(sk);
struct tipc_sock *tsk = tipc_sk(sk); struct tipc_sock *tsk = tipc_sk(sk);
struct sk_buff *skb; struct sk_buff *skb;
u32 dnode; u32 dnode = tsk_peer_node(tsk);
u32 dport = tsk_peer_port(tsk);
u32 onode = tipc_own_addr(net);
u32 oport = tsk->portid;
int res; int res;
if (how != SHUT_RDWR) if (how != SHUT_RDWR)
...@@ -2083,6 +2087,8 @@ static int tipc_shutdown(struct socket *sock, int how) ...@@ -2083,6 +2087,8 @@ static int tipc_shutdown(struct socket *sock, int how)
case SS_CONNECTED: case SS_CONNECTED:
restart: restart:
dnode = tsk_peer_node(tsk);
/* Disconnect and send a 'FIN+' or 'FIN-' message to peer */ /* Disconnect and send a 'FIN+' or 'FIN-' message to peer */
skb = __skb_dequeue(&sk->sk_receive_queue); skb = __skb_dequeue(&sk->sk_receive_queue);
if (skb) { if (skb) {
...@@ -2090,18 +2096,12 @@ static int tipc_shutdown(struct socket *sock, int how) ...@@ -2090,18 +2096,12 @@ static int tipc_shutdown(struct socket *sock, int how)
kfree_skb(skb); kfree_skb(skb);
goto restart; goto restart;
} }
if (tipc_msg_reverse(tsk_own_node(tsk), skb, &dnode, tipc_sk_respond(sk, skb, TIPC_CONN_SHUTDOWN);
TIPC_CONN_SHUTDOWN))
tipc_node_xmit_skb(net, skb, dnode,
tsk->portid);
} else { } else {
dnode = tsk_peer_node(tsk);
skb = tipc_msg_create(TIPC_CRITICAL_IMPORTANCE, skb = tipc_msg_create(TIPC_CRITICAL_IMPORTANCE,
TIPC_CONN_MSG, SHORT_H_SIZE, TIPC_CONN_MSG, SHORT_H_SIZE,
0, dnode, tsk_own_node(tsk), 0, dnode, onode, dport, oport,
tsk_peer_port(tsk), TIPC_CONN_SHUTDOWN);
tsk->portid, TIPC_CONN_SHUTDOWN);
tipc_node_xmit_skb(net, skb, dnode, tsk->portid); tipc_node_xmit_skb(net, skb, dnode, tsk->portid);
} }
tsk->connected = 0; tsk->connected = 0;
......
...@@ -44,7 +44,7 @@ ...@@ -44,7 +44,7 @@
SKB_TRUESIZE(TIPC_MAX_USER_MSG_SIZE)) SKB_TRUESIZE(TIPC_MAX_USER_MSG_SIZE))
int tipc_socket_init(void); int tipc_socket_init(void);
void tipc_socket_stop(void); void tipc_socket_stop(void);
int tipc_sk_rcv(struct net *net, struct sk_buff_head *inputq); void tipc_sk_rcv(struct net *net, struct sk_buff_head *inputq);
void tipc_sk_mcast_rcv(struct net *net, struct sk_buff_head *arrvq, void tipc_sk_mcast_rcv(struct net *net, struct sk_buff_head *arrvq,
struct sk_buff_head *inputq); struct sk_buff_head *inputq);
void tipc_sk_reinit(struct net *net); void tipc_sk_reinit(struct net *net);
......
Markdown is supported
0%
or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment