Merge git://git.kernel.org/pub/scm/linux/kernel/git/davem/net-next

Pull networking updates from David Millar:
 "Here are some highlights from the 2065 networking commits that
  happened this development cycle:

   1) XDP support for IXGBE (John Fastabend) and thunderx (Sunil Kowuri)

   2) Add a generic XDP driver, so that anyone can test XDP even if they
      lack a networking device whose driver has explicit XDP support
      (me).

   3) Sparc64 now has an eBPF JIT too (me)

   4) Add a BPF program testing framework via BPF_PROG_TEST_RUN (Alexei
      Starovoitov)

   5) Make netfitler network namespace teardown less expensive (Florian
      Westphal)

   6) Add symmetric hashing support to nft_hash (Laura Garcia Liebana)

   7) Implement NAPI and GRO in netvsc driver (Stephen Hemminger)

   8) Support TC flower offload statistics in mlxsw (Arkadi Sharshevsky)

   9) Multiqueue support in stmmac driver (Joao Pinto)

  10) Remove TCP timewait recycling, it never really could possibly work
      well in the real world and timestamp randomization really zaps any
      hint of usability this feature had (Soheil Hassas Yeganeh)

  11) Support level3 vs level4 ECMP route hashing in ipv4 (Nikolay
      Aleksandrov)

  12) Add socket busy poll support to epoll (Sridhar Samudrala)

  13) Netlink extended ACK support (Johannes Berg, Pablo Neira Ayuso,
      and several others)

  14) IPSEC hw offload infrastructure (Steffen Klassert)"

* git://git.kernel.org/pub/scm/linux/kernel/git/davem/net-next: (2065 commits)
  tipc: refactor function tipc_sk_recv_stream()
  tipc: refactor function tipc_sk_recvmsg()
  net: thunderx: Optimize page recycling for XDP
  net: thunderx: Support for XDP header adjustment
  net: thunderx: Add support for XDP_TX
  net: thunderx: Add support for XDP_DROP
  net: thunderx: Add basic XDP support
  net: thunderx: Cleanup receive buffer allocation
  net: thunderx: Optimize CQE_TX handling
  net: thunderx: Optimize RBDR descriptor handling
  net: thunderx: Support for page recycling
  ipx: call ipxitf_put() in ioctl error path
  net: sched: add helpers to handle extended actions
  qed*: Fix issues in the ptp filter config implementation.
  qede: Fix concurrency issue in PTP Tx path processing.
  stmmac: Add support for SIMATIC IOT2000 platform
  net: hns: fix ethtool_get_strings overflow in hns driver
  tcp: fix wraparound issue in tcp_lp
  bpf, arm64: fix jit branch offset related to ldimm64
  bpf, arm64: implement jiting of BPF_XADD
  ...

11 files changed, 197 insertions, 197 deletions

diff --git a/net/tipc/bearer.c b/net/tipc/bearer.c
index 33a5bdfbef76..d174ee3254ee 100644
--- a/net/tipc/bearer.c
+++ b/net/tipc/bearer.c
@@ -802,7 +802,7 @@ int tipc_nl_bearer_get(struct sk_buff *skb, struct genl_info *info)
 
         err = nla_parse_nested(attrs, TIPC_NLA_BEARER_MAX,
                                info->attrs[TIPC_NLA_BEARER],
-                               tipc_nl_bearer_policy);
+                               tipc_nl_bearer_policy, info->extack);
         if (err)
                 return err;
 
@@ -851,7 +851,7 @@ int tipc_nl_bearer_disable(struct sk_buff *skb, struct genl_info *info)
 
         err = nla_parse_nested(attrs, TIPC_NLA_BEARER_MAX,
                                info->attrs[TIPC_NLA_BEARER],
-                               tipc_nl_bearer_policy);
+                               tipc_nl_bearer_policy, info->extack);
         if (err)
                 return err;
 
@@ -891,7 +891,7 @@ int tipc_nl_bearer_enable(struct sk_buff *skb, struct genl_info *info)
 
         err = nla_parse_nested(attrs, TIPC_NLA_BEARER_MAX,
                                info->attrs[TIPC_NLA_BEARER],
-                               tipc_nl_bearer_policy);
+                               tipc_nl_bearer_policy, info->extack);
         if (err)
                 return err;
 
@@ -939,7 +939,7 @@ int tipc_nl_bearer_add(struct sk_buff *skb, struct genl_info *info)
 
         err = nla_parse_nested(attrs, TIPC_NLA_BEARER_MAX,
                                info->attrs[TIPC_NLA_BEARER],
-                               tipc_nl_bearer_policy);
+                               tipc_nl_bearer_policy, info->extack);
         if (err)
                 return err;
 
@@ -982,7 +982,7 @@ int tipc_nl_bearer_set(struct sk_buff *skb, struct genl_info *info)
 
         err = nla_parse_nested(attrs, TIPC_NLA_BEARER_MAX,
                                info->attrs[TIPC_NLA_BEARER],
-                               tipc_nl_bearer_policy);
+                               tipc_nl_bearer_policy, info->extack);
         if (err)
                 return err;
 
@@ -1104,7 +1104,7 @@ int tipc_nl_media_get(struct sk_buff *skb, struct genl_info *info)
 
         err = nla_parse_nested(attrs, TIPC_NLA_MEDIA_MAX,
                                info->attrs[TIPC_NLA_MEDIA],
-                               tipc_nl_media_policy);
+                               tipc_nl_media_policy, info->extack);
         if (err)
                 return err;
 
@@ -1152,7 +1152,7 @@ int tipc_nl_media_set(struct sk_buff *skb, struct genl_info *info)
 
         err = nla_parse_nested(attrs, TIPC_NLA_MEDIA_MAX,
                                info->attrs[TIPC_NLA_MEDIA],
-                               tipc_nl_media_policy);
+                               tipc_nl_media_policy, info->extack);
 
         if (!attrs[TIPC_NLA_MEDIA_NAME])
                 return -EINVAL;
diff --git a/net/tipc/link.c b/net/tipc/link.c
index ddd2dd6f77aa..60820dc35a08 100644
--- a/net/tipc/link.c
+++ b/net/tipc/link.c
@@ -1827,7 +1827,7 @@ int tipc_nl_parse_link_prop(struct nlattr *prop, struct nlattr *props[])
         int err;
 
         err = nla_parse_nested(props, TIPC_NLA_PROP_MAX, prop,
-                               tipc_nl_prop_policy);
+                               tipc_nl_prop_policy, NULL);
         if (err)
                 return err;
 
diff --git a/net/tipc/name_table.c b/net/tipc/name_table.c
index 9be6592e4a6f..bd0aac87b41a 100644
--- a/net/tipc/name_table.c
+++ b/net/tipc/name_table.c
@@ -416,6 +416,7 @@ static void tipc_nameseq_subscribe(struct name_seq *nseq,
 
         tipc_subscrp_convert_seq(&s->evt.s.seq, s->swap, &ns);
 
+        tipc_subscrp_get(s);
         list_add(&s->nameseq_list, &nseq->subscriptions);
 
         if (!sseq)
@@ -787,6 +788,7 @@ void tipc_nametbl_unsubscribe(struct tipc_subscription *s)
         if (seq != NULL) {
                 spin_lock_bh(&seq->lock);
                 list_del_init(&s->nameseq_list);
+                tipc_subscrp_put(s);
                 if (!seq->first_free && list_empty(&seq->subscriptions)) {
                         hlist_del_init_rcu(&seq->ns_list);
                         kfree(seq->sseqs);
diff --git a/net/tipc/net.c b/net/tipc/net.c
index ab8a2d5d1e32..719c5924b638 100644
--- a/net/tipc/net.c
+++ b/net/tipc/net.c
@@ -211,8 +211,8 @@ int tipc_nl_net_set(struct sk_buff *skb, struct genl_info *info)
                 return -EINVAL;
 
         err = nla_parse_nested(attrs, TIPC_NLA_NET_MAX,
-                               info->attrs[TIPC_NLA_NET],
-                               tipc_nl_net_policy);
+                               info->attrs[TIPC_NLA_NET], tipc_nl_net_policy,
+                               info->extack);
         if (err)
                 return err;
 
diff --git a/net/tipc/netlink.c b/net/tipc/netlink.c
index 26ca8dd64ded..b76f13f6fea1 100644
--- a/net/tipc/netlink.c
+++ b/net/tipc/netlink.c
@@ -268,7 +268,8 @@ int tipc_nlmsg_parse(const struct nlmsghdr *nlh, struct nlattr ***attr)
         if (!*attr)
                 return -EOPNOTSUPP;
 
-        return nlmsg_parse(nlh, GENL_HDRLEN, *attr, maxattr, tipc_nl_policy);
+        return nlmsg_parse(nlh, GENL_HDRLEN, *attr, maxattr, tipc_nl_policy,
+                           NULL);
 }
 
 int __init tipc_netlink_start(void)
diff --git a/net/tipc/netlink_compat.c b/net/tipc/netlink_compat.c
index e1ae8a8a2b8e..9bfe886ab330 100644
--- a/net/tipc/netlink_compat.c
+++ b/net/tipc/netlink_compat.c
@@ -296,7 +296,7 @@ static int __tipc_nl_compat_doit(struct tipc_nl_compat_cmd_doit *cmd,
 
         err = nla_parse(attrbuf, tipc_genl_family.maxattr,
                         (const struct nlattr *)trans_buf->data,
-                        trans_buf->len, NULL);
+                        trans_buf->len, NULL, NULL);
         if (err)
                 goto parse_out;
 
@@ -352,7 +352,7 @@ static int tipc_nl_compat_bearer_dump(struct tipc_nl_compat_msg *msg,
                 return -EINVAL;
 
         err = nla_parse_nested(bearer, TIPC_NLA_BEARER_MAX,
-                               attrs[TIPC_NLA_BEARER], NULL);
+                               attrs[TIPC_NLA_BEARER], NULL, NULL);
         if (err)
                 return err;
 
@@ -472,7 +472,7 @@ static int tipc_nl_compat_link_stat_dump(struct tipc_nl_compat_msg *msg,
                 return -EINVAL;
 
         err = nla_parse_nested(link, TIPC_NLA_LINK_MAX, attrs[TIPC_NLA_LINK],
-                               NULL);
+                               NULL, NULL);
         if (err)
                 return err;
 
@@ -480,7 +480,7 @@ static int tipc_nl_compat_link_stat_dump(struct tipc_nl_compat_msg *msg,
                 return -EINVAL;
 
         err = nla_parse_nested(prop, TIPC_NLA_PROP_MAX,
-                               link[TIPC_NLA_LINK_PROP], NULL);
+                               link[TIPC_NLA_LINK_PROP], NULL, NULL);
         if (err)
                 return err;
 
@@ -488,7 +488,7 @@ static int tipc_nl_compat_link_stat_dump(struct tipc_nl_compat_msg *msg,
                 return -EINVAL;
 
         err = nla_parse_nested(stats, TIPC_NLA_STATS_MAX,
-                               link[TIPC_NLA_LINK_STATS], NULL);
+                               link[TIPC_NLA_LINK_STATS], NULL, NULL);
         if (err)
                 return err;
 
@@ -598,7 +598,7 @@ static int tipc_nl_compat_link_dump(struct tipc_nl_compat_msg *msg,
                 return -EINVAL;
 
         err = nla_parse_nested(link, TIPC_NLA_LINK_MAX, attrs[TIPC_NLA_LINK],
-                               NULL);
+                               NULL, NULL);
         if (err)
                 return err;
 
@@ -795,7 +795,7 @@ static int tipc_nl_compat_name_table_dump(struct tipc_nl_compat_msg *msg,
                 return -EINVAL;
 
         err = nla_parse_nested(nt, TIPC_NLA_NAME_TABLE_MAX,
-                               attrs[TIPC_NLA_NAME_TABLE], NULL);
+                               attrs[TIPC_NLA_NAME_TABLE], NULL, NULL);
         if (err)
                 return err;
 
@@ -803,7 +803,7 @@ static int tipc_nl_compat_name_table_dump(struct tipc_nl_compat_msg *msg,
                 return -EINVAL;
 
         err = nla_parse_nested(publ, TIPC_NLA_PUBL_MAX,
-                               nt[TIPC_NLA_NAME_TABLE_PUBL], NULL);
+                               nt[TIPC_NLA_NAME_TABLE_PUBL], NULL, NULL);
         if (err)
                 return err;
 
@@ -863,7 +863,7 @@ static int __tipc_nl_compat_publ_dump(struct tipc_nl_compat_msg *msg,
                 return -EINVAL;
 
         err = nla_parse_nested(publ, TIPC_NLA_PUBL_MAX, attrs[TIPC_NLA_PUBL],
-                               NULL);
+                               NULL, NULL);
         if (err)
                 return err;
 
@@ -929,7 +929,7 @@ static int tipc_nl_compat_sk_dump(struct tipc_nl_compat_msg *msg,
                 return -EINVAL;
 
         err = nla_parse_nested(sock, TIPC_NLA_SOCK_MAX, attrs[TIPC_NLA_SOCK],
-                               NULL);
+                               NULL, NULL);
         if (err)
                 return err;
 
@@ -940,8 +940,8 @@ static int tipc_nl_compat_sk_dump(struct tipc_nl_compat_msg *msg,
                 u32 node;
                 struct nlattr *con[TIPC_NLA_CON_MAX + 1];
 
-                nla_parse_nested(con, TIPC_NLA_CON_MAX, sock[TIPC_NLA_SOCK_CON],
-                                 NULL);
+                nla_parse_nested(con, TIPC_NLA_CON_MAX,
+                                 sock[TIPC_NLA_SOCK_CON], NULL, NULL);
 
                 node = nla_get_u32(con[TIPC_NLA_CON_NODE]);
                 tipc_tlv_sprintf(msg->rep, "  connected to <%u.%u.%u:%u>",
@@ -977,8 +977,8 @@ static int tipc_nl_compat_media_dump(struct tipc_nl_compat_msg *msg,
         if (!attrs[TIPC_NLA_MEDIA])
                 return -EINVAL;
 
-        err = nla_parse_nested(media, TIPC_NLA_MEDIA_MAX, attrs[TIPC_NLA_MEDIA],
-                               NULL);
+        err = nla_parse_nested(media, TIPC_NLA_MEDIA_MAX,
+                               attrs[TIPC_NLA_MEDIA], NULL, NULL);
         if (err)
                 return err;
 
@@ -998,7 +998,7 @@ static int tipc_nl_compat_node_dump(struct tipc_nl_compat_msg *msg,
                 return -EINVAL;
 
         err = nla_parse_nested(node, TIPC_NLA_NODE_MAX, attrs[TIPC_NLA_NODE],
-                               NULL);
+                               NULL, NULL);
         if (err)
                 return err;
 
@@ -1045,7 +1045,7 @@ static int tipc_nl_compat_net_dump(struct tipc_nl_compat_msg *msg,
                 return -EINVAL;
 
         err = nla_parse_nested(net, TIPC_NLA_NET_MAX, attrs[TIPC_NLA_NET],
-                               NULL);
+                               NULL, NULL);
         if (err)
                 return err;
 
diff --git a/net/tipc/node.c b/net/tipc/node.c
index 4512e83652b1..aeef8011ac7d 100644
--- a/net/tipc/node.c
+++ b/net/tipc/node.c
@@ -1607,8 +1607,8 @@ int tipc_nl_peer_rm(struct sk_buff *skb, struct genl_info *info)
                 return -EINVAL;
 
         err = nla_parse_nested(attrs, TIPC_NLA_NET_MAX,
-                               info->attrs[TIPC_NLA_NET],
-                               tipc_nl_net_policy);
+                               info->attrs[TIPC_NLA_NET], tipc_nl_net_policy,
+                               info->extack);
         if (err)
                 return err;
 
@@ -1774,7 +1774,7 @@ int tipc_nl_node_set_link(struct sk_buff *skb, struct genl_info *info)
 
         err = nla_parse_nested(attrs, TIPC_NLA_LINK_MAX,
                                info->attrs[TIPC_NLA_LINK],
-                               tipc_nl_link_policy);
+                               tipc_nl_link_policy, info->extack);
         if (err)
                 return err;
 
@@ -1902,7 +1902,7 @@ int tipc_nl_node_reset_link_stats(struct sk_buff *skb, struct genl_info *info)
 
         err = nla_parse_nested(attrs, TIPC_NLA_LINK_MAX,
                                info->attrs[TIPC_NLA_LINK],
-                               tipc_nl_link_policy);
+                               tipc_nl_link_policy, info->extack);
         if (err)
                 return err;
 
@@ -2042,7 +2042,7 @@ int tipc_nl_node_set_monitor(struct sk_buff *skb, struct genl_info *info)
 
         err = nla_parse_nested(attrs, TIPC_NLA_MON_MAX,
                                info->attrs[TIPC_NLA_MON],
-                               tipc_nl_monitor_policy);
+                               tipc_nl_monitor_policy, info->extack);
         if (err)
                 return err;
 
@@ -2098,6 +2098,8 @@ int tipc_nl_node_get_monitor(struct sk_buff *skb, struct genl_info *info)
         int err;
 
         msg.skb = nlmsg_new(NLMSG_GOODSIZE, GFP_KERNEL);
+        if (!msg.skb)
+                return -ENOMEM;
         msg.portid = info->snd_portid;
         msg.seq = info->snd_seq;
 
@@ -2163,7 +2165,7 @@ int tipc_nl_node_dump_monitor_peer(struct sk_buff *skb,
 
                 err = nla_parse_nested(mon, TIPC_NLA_MON_MAX,
                                        attrs[TIPC_NLA_MON],
-                                       tipc_nl_monitor_policy);
+                                       tipc_nl_monitor_policy, NULL);
                 if (err)
                         return err;
 
diff --git a/net/tipc/socket.c b/net/tipc/socket.c
index bdce99f9407a..0d4f2f455a7c 100644
--- a/net/tipc/socket.c
+++ b/net/tipc/socket.c
@@ -51,6 +51,7 @@
 #define TIPC_FWD_MSG            1
 #define TIPC_MAX_PORT           0xffffffff
 #define TIPC_MIN_PORT           1
+#define TIPC_ACK_RATE           4       /* ACK at 1/4 of of rcv window size */
 
 enum {
         TIPC_LISTEN = TCP_LISTEN,
@@ -1305,7 +1306,7 @@ static int tipc_wait_for_rcvmsg(struct socket *sock, long *timeop)
 /**
  * tipc_recvmsg - receive packet-oriented message
  * @m: descriptor for message info
- * @buf_len: total size of user buffer area
+ * @buflen: length of user buffer area
  * @flags: receive flags
  *
  * Used for SOCK_DGRAM, SOCK_RDM, and SOCK_SEQPACKET messages.
@@ -1313,95 +1314,85 @@ static int tipc_wait_for_rcvmsg(struct socket *sock, long *timeop)
  *
  * Returns size of returned message data, errno otherwise
  */
-static int tipc_recvmsg(struct socket *sock, struct msghdr *m, size_t buf_len,
-                        int flags)
+static int tipc_recvmsg(struct socket *sock, struct msghdr *m,
+                        size_t buflen,  int flags)
 {
         struct sock *sk = sock->sk;
         struct tipc_sock *tsk = tipc_sk(sk);
-        struct sk_buff *buf;
-        struct tipc_msg *msg;
-        bool is_connectionless = tipc_sk_type_connectionless(sk);
-        long timeo;
-        unsigned int sz;
-        u32 err;
-        int res, hlen;
+        struct sk_buff *skb;
+        struct tipc_msg *hdr;
+        bool connected = !tipc_sk_type_connectionless(sk);
+        int rc, err, hlen, dlen, copy;
+        long timeout;
 
         /* Catch invalid receive requests */
-        if (unlikely(!buf_len))
+        if (unlikely(!buflen))
                 return -EINVAL;
 
         lock_sock(sk);
-
-        if (!is_connectionless && unlikely(sk->sk_state == TIPC_OPEN)) {
-                res = -ENOTCONN;
+        if (unlikely(connected && sk->sk_state == TIPC_OPEN)) {
+                rc = -ENOTCONN;
                 goto exit;
         }
+        timeout = sock_rcvtimeo(sk, flags & MSG_DONTWAIT);
 
-        timeo = sock_rcvtimeo(sk, flags & MSG_DONTWAIT);
-restart:
-
-        /* Look for a message in receive queue; wait if necessary */
-        res = tipc_wait_for_rcvmsg(sock, &timeo);
-        if (res)
-                goto exit;
-
-        /* Look at first message in receive queue */
-        buf = skb_peek(&sk->sk_receive_queue);
-        msg = buf_msg(buf);
-        sz = msg_data_sz(msg);
-        hlen = msg_hdr_sz(msg);
-        err = msg_errcode(msg);
-
-        /* Discard an empty non-errored message & try again */
-        if ((!sz) && (!err)) {
+        do {
+                /* Look at first msg in receive queue; wait if necessary */
+                rc = tipc_wait_for_rcvmsg(sock, &timeout);
+                if (unlikely(rc))
+                        goto exit;
+                skb = skb_peek(&sk->sk_receive_queue);
+                hdr = buf_msg(skb);
+                dlen = msg_data_sz(hdr);
+                hlen = msg_hdr_sz(hdr);
+                err = msg_errcode(hdr);
+                if (likely(dlen || err))
+                        break;
                 tsk_advance_rx_queue(sk);
-                goto restart;
-        }
-
-        /* Capture sender's address (optional) */
-        set_orig_addr(m, msg);
+        } while (1);
 
-        /* Capture ancillary data (optional) */
-        res = tipc_sk_anc_data_recv(m, msg, tsk);
-        if (res)
+        /* Collect msg meta data, including error code and rejected data */
+        set_orig_addr(m, hdr);
+        rc = tipc_sk_anc_data_recv(m, hdr, tsk);
+        if (unlikely(rc))
                 goto exit;
 
-        /* Capture message data (if valid) & compute return value (always) */
-        if (!err) {
-                if (unlikely(buf_len < sz)) {
-                        sz = buf_len;
+        /* Capture data if non-error msg, otherwise just set return value */
+        if (likely(!err)) {
+                copy = min_t(int, dlen, buflen);
+                if (unlikely(copy != dlen))
                         m->msg_flags |= MSG_TRUNC;
-                }
-                res = skb_copy_datagram_msg(buf, hlen, m, sz);
-                if (res)
-                        goto exit;
-                res = sz;
+                rc = skb_copy_datagram_msg(skb, hlen, m, copy);
         } else {
-                if (is_connectionless || err == TIPC_CONN_SHUTDOWN ||
-                    m->msg_control)
-                        res = 0;
-                else
-                        res = -ECONNRESET;
+                copy = 0;
+                rc = 0;
+                if (err != TIPC_CONN_SHUTDOWN && connected && !m->msg_control)
+                        rc = -ECONNRESET;
         }
+        if (unlikely(rc))
+                goto exit;
 
+        /* Caption of data or error code/rejected data was successful */
         if (unlikely(flags & MSG_PEEK))
                 goto exit;
 
-        if (likely(!is_connectionless)) {
-                tsk->rcv_unacked += tsk_inc(tsk, hlen + sz);
-                if (unlikely(tsk->rcv_unacked >= (tsk->rcv_win / 4)))
-                        tipc_sk_send_ack(tsk);
-        }
         tsk_advance_rx_queue(sk);
+        if (likely(!connected))
+                goto exit;
+
+        /* Send connection flow control ack when applicable */
+        tsk->rcv_unacked += tsk_inc(tsk, hlen + dlen);
+        if (tsk->rcv_unacked >= tsk->rcv_win / TIPC_ACK_RATE)
+                tipc_sk_send_ack(tsk);
 exit:
         release_sock(sk);
-        return res;
+        return rc ? rc : copy;
 }
 
 /**
- * tipc_recv_stream - receive stream-oriented data
+ * tipc_recvstream - receive stream-oriented data
  * @m: descriptor for message info
- * @buf_len: total size of user buffer area
+ * @buflen: total size of user buffer area
  * @flags: receive flags
  *
  * Used for SOCK_STREAM messages only.  If not enough data is available
@@ -1409,111 +1400,98 @@ exit:
  *
  * Returns size of returned message data, errno otherwise
  */
-static int tipc_recv_stream(struct socket *sock, struct msghdr *m,
-                            size_t buf_len, int flags)
+static int tipc_recvstream(struct socket *sock, struct msghdr *m,
+                           size_t buflen, int flags)
 {
         struct sock *sk = sock->sk;
         struct tipc_sock *tsk = tipc_sk(sk);
-        struct sk_buff *buf;
-        struct tipc_msg *msg;
-        long timeo;
-        unsigned int sz;
-        int target;
-        int sz_copied = 0;
-        u32 err;
-        int res = 0, hlen;
+        struct sk_buff *skb;
+        struct tipc_msg *hdr;
+        struct tipc_skb_cb *skb_cb;
+        bool peek = flags & MSG_PEEK;
+        int offset, required, copy, copied = 0;
+        int hlen, dlen, err, rc;
+        long timeout;
 
         /* Catch invalid receive attempts */
-        if (unlikely(!buf_len))
+        if (unlikely(!buflen))
                 return -EINVAL;
 
         lock_sock(sk);
 
         if (unlikely(sk->sk_state == TIPC_OPEN)) {
-                res = -ENOTCONN;
-                goto exit;
-        }
-
-        target = sock_rcvlowat(sk, flags & MSG_WAITALL, buf_len);
-        timeo = sock_rcvtimeo(sk, flags & MSG_DONTWAIT);
-
-restart:
-        /* Look for a message in receive queue; wait if necessary */
-        res = tipc_wait_for_rcvmsg(sock, &timeo);
-        if (res)
+                rc = -ENOTCONN;
                 goto exit;
-
-        /* Look at first message in receive queue */
-        buf = skb_peek(&sk->sk_receive_queue);
-        msg = buf_msg(buf);
-        sz = msg_data_sz(msg);
-        hlen = msg_hdr_sz(msg);
-        err = msg_errcode(msg);
-
-        /* Discard an empty non-errored message & try again */
-        if ((!sz) && (!err)) {
-                tsk_advance_rx_queue(sk);
-                goto restart;
         }
+        required = sock_rcvlowat(sk, flags & MSG_WAITALL, buflen);
+        timeout = sock_rcvtimeo(sk, flags & MSG_DONTWAIT);
 
-        /* Optionally capture sender's address & ancillary data of first msg */
-        if (sz_copied == 0) {
-                set_orig_addr(m, msg);
-                res = tipc_sk_anc_data_recv(m, msg, tsk);
-                if (res)
-                        goto exit;
-        }
-
-        /* Capture message data (if valid) & compute return value (always) */
-        if (!err) {
-                u32 offset = TIPC_SKB_CB(buf)->bytes_read;
-                u32 needed;
-                int sz_to_copy;
-
-                sz -= offset;
-                needed = (buf_len - sz_copied);
-                sz_to_copy = min(sz, needed);
-
-                res = skb_copy_datagram_msg(buf, hlen + offset, m, sz_to_copy);
-                if (res)
-                        goto exit;
+        do {
+                /* Look at first msg in receive queue; wait if necessary */
+                rc = tipc_wait_for_rcvmsg(sock, &timeout);
+                if (unlikely(rc))
+                        break;
+                skb = skb_peek(&sk->sk_receive_queue);
+                skb_cb = TIPC_SKB_CB(skb);
+                hdr = buf_msg(skb);
+                dlen = msg_data_sz(hdr);
+                hlen = msg_hdr_sz(hdr);
+                err = msg_errcode(hdr);
+
+                /* Discard any empty non-errored (SYN-) message */
+                if (unlikely(!dlen && !err)) {
+                        tsk_advance_rx_queue(sk);
+                        continue;
+                }
 
-                sz_copied += sz_to_copy;
+                /* Collect msg meta data, incl. error code and rejected data */
+                if (!copied) {
+                        set_orig_addr(m, hdr);
+                        rc = tipc_sk_anc_data_recv(m, hdr, tsk);
+                        if (rc)
+                                break;
+                }
 
-                if (sz_to_copy < sz) {
-                        if (!(flags & MSG_PEEK))
-                                TIPC_SKB_CB(buf)->bytes_read =
-                                        offset + sz_to_copy;
-                        goto exit;
+                /* Copy data if msg ok, otherwise return error/partial data */
+                if (likely(!err)) {
+                        offset = skb_cb->bytes_read;
+                        copy = min_t(int, dlen - offset, buflen - copied);
+                        rc = skb_copy_datagram_msg(skb, hlen + offset, m, copy);
+                        if (unlikely(rc))
+                                break;
+                        copied += copy;
+                        offset += copy;
+                        if (unlikely(offset < dlen)) {
+                                if (!peek)
+                                        skb_cb->bytes_read = offset;
+                                break;
+                        }
+                } else {
+                        rc = 0;
+                        if ((err != TIPC_CONN_SHUTDOWN) && !m->msg_control)
+                                rc = -ECONNRESET;
+                        if (copied || rc)
+                                break;
                 }
-        } else {
-                if (sz_copied != 0)
-                        goto exit; /* can't add error msg to valid data */
 
-                if ((err == TIPC_CONN_SHUTDOWN) || m->msg_control)
-                        res = 0;
-                else
-                        res = -ECONNRESET;
-        }
+                if (unlikely(peek))
+                        break;
 
-        if (unlikely(flags & MSG_PEEK))
-                goto exit;
+                tsk_advance_rx_queue(sk);
 
-        tsk->rcv_unacked += tsk_inc(tsk, hlen + msg_data_sz(msg));
-        if (unlikely(tsk->rcv_unacked >= (tsk->rcv_win / 4)))
-                tipc_sk_send_ack(tsk);
-        tsk_advance_rx_queue(sk);
+                /* Send connection flow control advertisement when applicable */
+                tsk->rcv_unacked += tsk_inc(tsk, hlen + dlen);
+                if (unlikely(tsk->rcv_unacked >= tsk->rcv_win / TIPC_ACK_RATE))
+                        tipc_sk_send_ack(tsk);
 
-        /* Loop around if more data is required */
-        if ((sz_copied < buf_len) &&    /* didn't get all requested data */
-            (!skb_queue_empty(&sk->sk_receive_queue) ||
-            (sz_copied < target)) &&    /* and more is ready or required */
-            (!err))                     /* and haven't reached a FIN */
-                goto restart;
+                /* Exit if all requested data or FIN/error received */
+                if (copied == buflen || err)
+                        break;
 
+        } while (!skb_queue_empty(&sk->sk_receive_queue) || copied < required);
 exit:
         release_sock(sk);
-        return sz_copied ? sz_copied : res;
+        return copied ? copied : rc;
 }
 
 /**
@@ -2537,6 +2515,28 @@ static int tipc_ioctl(struct socket *sock, unsigned int cmd, unsigned long arg)
         }
 }
 
+static int tipc_socketpair(struct socket *sock1, struct socket *sock2)
+{
+        struct tipc_sock *tsk2 = tipc_sk(sock2->sk);
+        struct tipc_sock *tsk1 = tipc_sk(sock1->sk);
+        u32 onode = tipc_own_addr(sock_net(sock1->sk));
+
+        tsk1->peer.family = AF_TIPC;
+        tsk1->peer.addrtype = TIPC_ADDR_ID;
+        tsk1->peer.scope = TIPC_NODE_SCOPE;
+        tsk1->peer.addr.id.ref = tsk2->portid;
+        tsk1->peer.addr.id.node = onode;
+        tsk2->peer.family = AF_TIPC;
+        tsk2->peer.addrtype = TIPC_ADDR_ID;
+        tsk2->peer.scope = TIPC_NODE_SCOPE;
+        tsk2->peer.addr.id.ref = tsk1->portid;
+        tsk2->peer.addr.id.node = onode;
+
+        tipc_sk_finish_conn(tsk1, tsk2->portid, onode);
+        tipc_sk_finish_conn(tsk2, tsk1->portid, onode);
+        return 0;
+}
+
 /* Protocol switches for the various types of TIPC sockets */
 
 static const struct proto_ops msg_ops = {
@@ -2545,7 +2545,7 @@ static const struct proto_ops msg_ops = {
         .release        = tipc_release,
         .bind           = tipc_bind,
         .connect        = tipc_connect,
-        .socketpair     = sock_no_socketpair,
+        .socketpair     = tipc_socketpair,
         .accept         = sock_no_accept,
         .getname        = tipc_getname,
         .poll           = tipc_poll,
@@ -2566,7 +2566,7 @@ static const struct proto_ops packet_ops = {
         .release        = tipc_release,
         .bind           = tipc_bind,
         .connect        = tipc_connect,
-        .socketpair     = sock_no_socketpair,
+        .socketpair     = tipc_socketpair,
         .accept         = tipc_accept,
         .getname        = tipc_getname,
         .poll           = tipc_poll,
@@ -2587,7 +2587,7 @@ static const struct proto_ops stream_ops = {
         .release        = tipc_release,
         .bind           = tipc_bind,
         .connect        = tipc_connect,
-        .socketpair     = sock_no_socketpair,
+        .socketpair     = tipc_socketpair,
         .accept         = tipc_accept,
         .getname        = tipc_getname,
         .poll           = tipc_poll,
@@ -2597,7 +2597,7 @@ static const struct proto_ops stream_ops = {
         .setsockopt     = tipc_setsockopt,
         .getsockopt     = tipc_getsockopt,
         .sendmsg        = tipc_sendstream,
-        .recvmsg        = tipc_recv_stream,
+        .recvmsg        = tipc_recvstream,
         .mmap           = sock_no_mmap,
         .sendpage       = sock_no_sendpage
 };
@@ -2870,7 +2870,7 @@ int tipc_nl_publ_dump(struct sk_buff *skb, struct netlink_callback *cb)
 
                 err = nla_parse_nested(sock, TIPC_NLA_SOCK_MAX,
                                        attrs[TIPC_NLA_SOCK],
-                                       tipc_nl_sock_policy);
+                                       tipc_nl_sock_policy, NULL);
                 if (err)
                         return err;
 
diff --git a/net/tipc/subscr.c b/net/tipc/subscr.c
index 271cd66e4b3b..0bf91cd3733c 100644
--- a/net/tipc/subscr.c
+++ b/net/tipc/subscr.c
@@ -54,8 +54,6 @@ struct tipc_subscriber {
 
 static void tipc_subscrp_delete(struct tipc_subscription *sub);
 static void tipc_subscrb_put(struct tipc_subscriber *subscriber);
-static void tipc_subscrp_put(struct tipc_subscription *subscription);
-static void tipc_subscrp_get(struct tipc_subscription *subscription);
 
 /**
  * htohl - convert value to endianness used by destination
@@ -125,7 +123,6 @@ void tipc_subscrp_report_overlap(struct tipc_subscription *sub, u32 found_lower,
 {
         struct tipc_name_seq seq;
 
-        tipc_subscrp_get(sub);
         tipc_subscrp_convert_seq(&sub->evt.s.seq, sub->swap, &seq);
         if (!tipc_subscrp_check_overlap(&seq, found_lower, found_upper))
                 return;
@@ -135,7 +132,6 @@ void tipc_subscrp_report_overlap(struct tipc_subscription *sub, u32 found_lower,
 
         tipc_subscrp_send_event(sub, found_lower, found_upper, event, port_ref,
                                 node);
-        tipc_subscrp_put(sub);
 }
 
 static void tipc_subscrp_timeout(unsigned long data)
@@ -145,6 +141,7 @@ static void tipc_subscrp_timeout(unsigned long data)
 
         spin_lock_bh(&subscriber->lock);
         tipc_nametbl_unsubscribe(sub);
+        list_del(&sub->subscrp_list);
         spin_unlock_bh(&subscriber->lock);
 
         /* Notify subscriber of timeout */
@@ -177,20 +174,17 @@ static void tipc_subscrp_kref_release(struct kref *kref)
         struct tipc_net *tn = net_generic(sub->net, tipc_net_id);
         struct tipc_subscriber *subscriber = sub->subscriber;
 
-        spin_lock_bh(&subscriber->lock);
-        list_del(&sub->subscrp_list);
         atomic_dec(&tn->subscription_count);
-        spin_unlock_bh(&subscriber->lock);
         kfree(sub);
         tipc_subscrb_put(subscriber);
 }
 
-static void tipc_subscrp_put(struct tipc_subscription *subscription)
+void tipc_subscrp_put(struct tipc_subscription *subscription)
 {
         kref_put(&subscription->kref, tipc_subscrp_kref_release);
 }
 
-static void tipc_subscrp_get(struct tipc_subscription *subscription)
+void tipc_subscrp_get(struct tipc_subscription *subscription)
 {
         kref_get(&subscription->kref);
 }
@@ -210,11 +204,8 @@ static void tipc_subscrb_subscrp_delete(struct tipc_subscriber *subscriber,
                         continue;
 
                 tipc_nametbl_unsubscribe(sub);
-                tipc_subscrp_get(sub);
-                spin_unlock_bh(&subscriber->lock);
+                list_del(&sub->subscrp_list);
                 tipc_subscrp_delete(sub);
-                tipc_subscrp_put(sub);
-                spin_lock_bh(&subscriber->lock);
 
                 if (s)
                         break;
diff --git a/net/tipc/subscr.h b/net/tipc/subscr.h
index ffdc214c117a..ee52957dc952 100644
--- a/net/tipc/subscr.h
+++ b/net/tipc/subscr.h
@@ -78,4 +78,7 @@ u32 tipc_subscrp_convert_seq_type(u32 type, int swap);
 int tipc_topsrv_start(struct net *net);
 void tipc_topsrv_stop(struct net *net);
 
+void tipc_subscrp_put(struct tipc_subscription *subscription);
+void tipc_subscrp_get(struct tipc_subscription *subscription);
+
 #endif
diff --git a/net/tipc/udp_media.c b/net/tipc/udp_media.c
index 46061cf48cd1..ecca64fc6a6f 100644
--- a/net/tipc/udp_media.c
+++ b/net/tipc/udp_media.c
@@ -457,7 +457,7 @@ int tipc_udp_nl_dump_remoteip(struct sk_buff *skb, struct netlink_callback *cb)
 
                 err = nla_parse_nested(battrs, TIPC_NLA_BEARER_MAX,
                                        attrs[TIPC_NLA_BEARER],
-                                       tipc_nl_bearer_policy);
+                                       tipc_nl_bearer_policy, NULL);
                 if (err)
                         return err;
 
@@ -609,7 +609,8 @@ int tipc_udp_nl_bearer_add(struct tipc_bearer *b, struct nlattr *attr)
         struct nlattr *opts[TIPC_NLA_UDP_MAX + 1];
         struct udp_media_addr *dst;
 
-        if (nla_parse_nested(opts, TIPC_NLA_UDP_MAX, attr, tipc_nl_udp_policy))
+        if (nla_parse_nested(opts, TIPC_NLA_UDP_MAX, attr,
+                             tipc_nl_udp_policy, NULL))
                 return -EINVAL;
 
         if (!opts[TIPC_NLA_UDP_REMOTE])
@@ -662,7 +663,7 @@ static int tipc_udp_enable(struct net *net, struct tipc_bearer *b,
 
         if (nla_parse_nested(opts, TIPC_NLA_UDP_MAX,
                              attrs[TIPC_NLA_BEARER_UDP_OPTS],
-                             tipc_nl_udp_policy))
+                             tipc_nl_udp_policy, NULL))
                 goto err;
 
         if (!opts[TIPC_NLA_UDP_LOCAL] || !opts[TIPC_NLA_UDP_REMOTE]) {