SO_REUSEPORT - can it be done in kernel?

Hi Tom,

How did you solved the issue regarding scaling TCP listeners?
I think SO_REUSEPORT proposed by patch [1] can be a good
start. Where there any follow ups?

Also, solving the problem in users pace can be an option. I want
to run multiple instances of a DNS server on a multi core system.
Any suggestions would be welcomed.

SO_REUSEPORT option seems to be already there [2]. Where
there any plans to have a kernel implementation?

thanks,
Daniel.

[1] http://amailbox.org/mailarchive/linux-netdev/2010/4/19/6274993
[2] http://lxr.linux.no/linux+v2.6.37/include/asm-generic/socket.h#L25

On Thu, Jan 27, 2011 at 02:07, Daniel Baluta [off-list ref] wrote:

How did you solved the issue regarding scaling TCP listeners?
I think SO_REUSEPORT proposed by patch [1] can be a good
start. Where there any follow ups?

Google is using the patch internally.  I've recently joined google and
have picked up this work from Tom; I'm starting to rework how it
interacts with TCP (in particular, changing how it interacts with
request sockets and listen sockets so that incoming connections are
not prematurely bound to a specific listener sharing the port).  I
have nothing worth sharing yet.

On Thu, Jan 27, 2011 at 2:07 AM, Daniel Baluta [off-list ref] wrote:

Hi Tom,

How did you solved the issue regarding scaling TCP listeners?
I think SO_REUSEPORT proposed by patch [1] can be a good
start. Where there any follow ups?

As Bill mentioned we are continue to work on the TCP issues.  Looks
like modifying the TCP listener structures will probably be required.

Also, solving the problem in users pace can be an option. I want
to run multiple instances of a DNS server on a multi core system.
Any suggestions would be welcomed.

SO_REUSEPORT option seems to be already there [2]. Where
there any plans to have a kernel implementation?

Yes, we are still planning this.  The UDP implementation for my
earlier patch should be usable to try for DNS/UDP-- this is in fact
where we saw a major performance gain.  Eric Dumazet had some nice
improvements that should probably be looked at also.

Tom

On Thu, Jan 27, 2011 at 01:32:25PM -0800, Tom Herbert wrote:

Yes, we are still planning this.  The UDP implementation for my
earlier patch should be usable to try for DNS/UDP-- this is in fact
where we saw a major performance gain.  Eric Dumazet had some nice
improvements that should probably be looked at also.

I can confirm this.

Serious scalability issues have been reported on a 12 core system
running bind 9.7-2. The system was only able to deliver ~110K queries
per second.

Using your SO_REUSEPORT patch and a modified bind using it. The same
system is able to deliver ~650K queries per seconds while maxing out
all cores completely.

Tom, Bill: do you have a timeline for merging this? Especially the
UDP bits?

-Thomas

On Fri, 2011-02-25 at 07:56 -0500, Thomas Graf wrote:

On Thu, Jan 27, 2011 at 01:32:25PM -0800, Tom Herbert wrote:

quoted

Yes, we are still planning this.  The UDP implementation for my
earlier patch should be usable to try for DNS/UDP-- this is in fact
where we saw a major performance gain.  Eric Dumazet had some nice
improvements that should probably be looked at also.

I can confirm this.

Serious scalability issues have been reported on a 12 core system
running bind 9.7-2. The system was only able to deliver ~110K queries
per second.

Using your SO_REUSEPORT patch and a modified bind using it. The same
system is able to deliver ~650K queries per seconds while maxing out
all cores completely.

I think the idea is goodness, but will ask, was the (first) bottleneck
actually in the kernel, or was it in bind itself?  I've seen
single-instance, single-byte burst-mode netperf TCP_RR do in excess of
300K transactions per second (with TCP_NODELAY set) on an X5560 core.

ftp://ftp.netperf.org/netperf/misc/dl380g6_X5560_rhel54_ad386_cxgb3_1.4.1.2_b2b_to_same_agg_1500mtu_20100513-2.csv

and that was with now ancient RHEL5.4 bits...  yes, there is a bit of
apples, oranges and kumquats but still, I am wondering if this didn't
also "work around" some internal BIND scaling issues as well.

rick jones

Tom, Bill: do you have a timeline for merging this? Especially the
UDP bits?

-Thomas
--
To unsubscribe from this list: send the line "unsubscribe netdev" in
the body of a message to majordomo@vger.kernel.org
More majordomo info at  http://vger.kernel.org/majordomo-info.html

From: Rick Jones <redacted>
Date: Fri, 25 Feb 2011 11:18:15 -0800

and that was with now ancient RHEL5.4 bits...  yes, there is a bit of
apples, oranges and kumquats but still, I am wondering if this didn't
also "work around" some internal BIND scaling issues as well.

I think this is fundamentally a bind problem as well.

Le vendredi 25 février 2011 à 11:18 -0800, Rick Jones a écrit :

I think the idea is goodness, but will ask, was the (first) bottleneck
actually in the kernel, or was it in bind itself?  I've seen
single-instance, single-byte burst-mode netperf TCP_RR do in excess of
300K transactions per second (with TCP_NODELAY set) on an X5560 core.

ftp://ftp.netperf.org/netperf/misc/dl380g6_X5560_rhel54_ad386_cxgb3_1.4.1.2_b2b_to_same_agg_1500mtu_20100513-2.csv

and that was with now ancient RHEL5.4 bits...  yes, there is a bit of
apples, oranges and kumquats but still, I am wondering if this didn't
also "work around" some internal BIND scaling issues as well.

A single core can probably give 300K transactions.

But if you use several cores, accessing a single socket (the one bound
on port 53), then performance drops because of false sharing,
locking....

Using your SO_REUSEPORT patch and a modified bind using it. The same
system is able to deliver ~650K queries per seconds while maxing out
all cores completely.

Nice data point.

Tom, Bill: do you have a timeline for merging this? Especially the
UDP bits?

Bill has been working on the TCP implementation which is requiring
some fairly major surgery on the listener connections in syn-rcvd
state, this is ongoing.

On the UDP side, I believe the patch is functional, but as Eric
pointed out it probably could be further optimized.  I'll split out
the UDP bits into a separate patch and post that...

Tom

-Thomas

On Fri, Feb 25, 2011 at 11:18:15AM -0800, Rick Jones wrote:

I think the idea is goodness, but will ask, was the (first) bottleneck
actually in the kernel, or was it in bind itself?  I've seen
single-instance, single-byte burst-mode netperf TCP_RR do in excess of
300K transactions per second (with TCP_NODELAY set) on an X5560 core.

ftp://ftp.netperf.org/netperf/misc/dl380g6_X5560_rhel54_ad386_cxgb3_1.4.1.2_b2b_to_same_agg_1500mtu_20100513-2.csv

and that was with now ancient RHEL5.4 bits...  yes, there is a bit of
apples, oranges and kumquats but still, I am wondering if this didn't
also "work around" some internal BIND scaling issues as well.

Yes it is. We have observed two separate bottlenecks.

The first we have discovered is within BIND. As soon as more than 1
worker thread is being used strace showed a ton of futex() system
calls to the kernel as soon as the number of queries crossed a magic
barrier. This suggested heavy lock contention within BIND.

This BIND lock contetion was not visible on all systems having scalability
issues though. Some machines were not able to deliver enough queries to
BIND in order for the lock contention to appear.

On Fri, Feb 25, 2011 at 11:51:13AM -0800, Tom Herbert wrote:

On the UDP side, I believe the patch is functional, but as Eric
pointed out it probably could be further optimized.  I'll split out
the UDP bits into a separate patch and post that...

Cool! I will be happy to assist in improving it further.

We already see 97-99% CPU utizliation spread evenly over all cores
(with about 20-30% spent in softirq) so at least it already scales
perfectly well.

On Fri, 2011-02-25 at 17:48 -0500, Thomas Graf wrote:

On Fri, Feb 25, 2011 at 11:18:15AM -0800, Rick Jones wrote:

quoted

I think the idea is goodness, but will ask, was the (first) bottleneck
actually in the kernel, or was it in bind itself?  I've seen
single-instance, single-byte burst-mode netperf TCP_RR do in excess of
300K transactions per second (with TCP_NODELAY set) on an X5560 core.

ftp://ftp.netperf.org/netperf/misc/dl380g6_X5560_rhel54_ad386_cxgb3_1.4.1.2_b2b_to_same_agg_1500mtu_20100513-2.csv

and that was with now ancient RHEL5.4 bits...  yes, there is a bit of
apples, oranges and kumquats but still, I am wondering if this didn't
also "work around" some internal BIND scaling issues as well.

Yes it is. We have observed two separate bottlenecks.

The first we have discovered is within BIND. As soon as more than 1
worker thread is being used strace showed a ton of futex() system
calls to the kernel as soon as the number of queries crossed a magic
barrier. This suggested heavy lock contention within BIND.

The more things change, the more they remain the same, or perhaps "Code
may come and go, but lock contention is forever:

ftp://ftp.cup.hp.com/dist/networking/briefs/bind9_perf.txt

rick jones

The system ftp.cup.hp.com is probably going away before long, I will
probably put its collection of ancient writeups somewhere on netperf.org

This BIND lock contetion was not visible on all systems having scalability
issues though. Some machines were not able to deliver enough queries to
BIND in order for the lock contention to appear.

On Fri, Feb 25, 2011 at 11:51, Tom Herbert [off-list ref] wrote:

quoted

Tom, Bill: do you have a timeline for merging this? Especially the
UDP bits?

Bill has been working on the TCP implementation which is requiring
some fairly major surgery on the listener connections in syn-rcvd
state, this is ongoing.

Yup.  The broad approach I settled on is to delay binding of new
connections to listener sockets by moving receive_sock's from a
per-listen_sock hash table to new hash chains in the global hash
table.

This is very much a work-in-progress.  I'm part way through the
conversion and have running code with most of the new structures in
place in parallel with the old; I'm about to start relying exclusively
on the new, and then will tear down the old; once that's done I'll be
in a position to hook that up to SO_REUSEPORT and start actually
measuring the difference.  In short: it will be a while.

So splitting SO_REUSEPORT for UDP from SO_REUSEPORT for TCP makes a
lot of sense to me.

David Miller [off-list ref] wrote:

I think this is fundamentally a bind problem as well.

I'm fairly certain the bottleneck is indeed in the kernel, and
in the UDP stack in particular.

This is born out by a test where I used two named worker threads,
both working on the same socket.  Stracing shows that they're
working flat out only doing sendmsg/recvmsg.

The result was that they obtained (in aggregate) half the throughput
of a single worker thread.

I then retested by having them use two sockets and the performance
greatly improved.

Now this is actually expected since our UDP stack is essentially
single-threaded on the send side when only one socket is being
used, mostly due to the corking functionality.

I'm unsure how big a role the receive side scalability actually
plays in this case, but I suspect it isn't great.

Which is why I'm quite skeptical about this REUSEPORT patch as
IMHO the only reason it produces a great result is solely because
it is allowing parallel sends going out.

Rather than modifying all UDP applications out there to fix what
is fundamentally a kernel problem, I think what we should do is
fix the UDP stack so that it actually scales.

It isn't all that hard since the easy way would be to only take
the lock if we're already corked or about to cork.

For the receive side we also don't need REUSEPORT as we can simply
make our UDP stack multiqueue.

Cheers,
-- 
Email: Herbert Xu [off-list ref]
Home Page: http://gondor.apana.org.au/~herbert/
PGP Key: http://gondor.apana.org.au/~herbert/pubkey.txt

From: Herbert Xu <herbert@gondor.apana.org.au>
Date: Sat, 26 Feb 2011 08:57:18 +0800

It isn't all that hard since the easy way would be to only take
the lock if we're already corked or about to cork.

We take the lock unconditionally because we essentially have to after
UDP takes on the socket buffer accounting facilities similar to TCP.

On Fri, Feb 25, 2011 at 06:12:44PM -0800, David Miller wrote:

We take the lock unconditionally because we essentially have to after
UDP takes on the socket buffer accounting facilities similar to TCP.

Well I just checked out the history tree (2.6.12) and it too had
the unconditional lock on the send path.  So this predates the
system-wide buffer limit change.

I'm looking at redoing this and the bulk of the work is going to
be restructuring ip_append_data/ip_push_pending_frames so that it
doesn't store the states in sk/inet_sk.

Cheers,
-- 
Email: Herbert Xu [off-list ref]
Home Page: http://gondor.apana.org.au/~herbert/
PGP Key: http://gondor.apana.org.au/~herbert/pubkey.txt

From: Herbert Xu <herbert@gondor.apana.org.au>
Date: Sat, 26 Feb 2011 10:48:48 +0800

I'm looking at redoing this and the bulk of the work is going to
be restructuring ip_append_data/ip_push_pending_frames so that it
doesn't store the states in sk/inet_sk.

I suppose you're going to replace that stuff with an on-stack
control structure that gets passed around by reference or
similar?

On Fri, Feb 25, 2011 at 07:07:23PM -0800, David Miller wrote:

From: Herbert Xu <herbert@gondor.apana.org.au>
Date: Sat, 26 Feb 2011 10:48:48 +0800

quoted

I'm looking at redoing this and the bulk of the work is going to
be restructuring ip_append_data/ip_push_pending_frames so that it
doesn't store the states in sk/inet_sk.

I suppose you're going to replace that stuff with an on-stack
control structure that gets passed around by reference or
similar?

Either that or have ip_append_data do ip_push_pending_frames
directly.

That function's signature is a mess already and I need to think
about this a bit more :)

Cheers,
-- 
Email: Herbert Xu [off-list ref]
Home Page: http://gondor.apana.org.au/~herbert/
PGP Key: http://gondor.apana.org.au/~herbert/pubkey.txt

Le samedi 26 février 2011 à 11:11 +0800, Herbert Xu a écrit :

On Fri, Feb 25, 2011 at 07:07:23PM -0800, David Miller wrote:

quoted

From: Herbert Xu <herbert@gondor.apana.org.au>
Date: Sat, 26 Feb 2011 10:48:48 +0800

quoted

I'm looking at redoing this and the bulk of the work is going to
be restructuring ip_append_data/ip_push_pending_frames so that it
doesn't store the states in sk/inet_sk.

I suppose you're going to replace that stuff with an on-stack
control structure that gets passed around by reference or
similar?

Either that or have ip_append_data do ip_push_pending_frames
directly.

That function's signature is a mess already and I need to think
about this a bit more :)

Cheers,

UDP CORK is a problem indeed. I wonder who really uses it ?

From: Eric Dumazet <redacted>
Date: Sat, 26 Feb 2011 08:31:24 +0100

UDP CORK is a problem indeed. I wonder who really uses it ?

git grep MSG_MORE -- net/sunrpc

On Sat, Feb 26, 2011 at 08:57:18AM +0800, Herbert Xu wrote:

I'm fairly certain the bottleneck is indeed in the kernel, and
in the UDP stack in particular.

This is born out by a test where I used two named worker threads,
both working on the same socket.  Stracing shows that they're
working flat out only doing sendmsg/recvmsg.

The result was that they obtained (in aggregate) half the throughput
of a single worker thread.

I agree. This is the bottleneck that I described were the kernel is
not able to deliver enough queries for BIND to show the lock
contention issues.

But there is also the situation where netperf RR performance numbers
indicate a mugh higher kernel capability but BIND is not able to
deliver more even though the CPU utilization is very low. This is
the situation where we see the large number of futex calls indicating
the lock contention due to too many queries on a single socket.

Which is why I'm quite skeptical about this REUSEPORT patch as
IMHO the only reason it produces a great result is solely because
it is allowing parallel sends going out.

Rather than modifying all UDP applications out there to fix what
is fundamentally a kernel problem, I think what we should do is
fix the UDP stack so that it actually scales.

I am not suggesting that this is the ultimate and final fix for this
problem. It is fixing a symptom rather than fixing the cause but
sometimes being able to fix the symptom becomes really handy :-)

Adding SO_REUSEPORT does not prevent us from fixing the UDP stack
in the long run.

It isn't all that hard since the easy way would be to only take
the lock if we're already corked or about to cork.

For the receive side we also don't need REUSEPORT as we can simply
make our UDP stack multiqueue.

OK, it is not required and there is definitely a better way to fix
the kernel bottleneck in the long term. Even better.

I still suggest to merge this patch as a immediate workaround fix
until we scale properly on a single socket and also as a workaround
for applications which can't get rid of their per socket mutex quickly.

On Sun, Feb 27, 2011 at 06:02:05AM -0500, Thomas Graf wrote:

I still suggest to merge this patch as a immediate workaround fix
until we scale properly on a single socket and also as a workaround
for applications which can't get rid of their per socket mutex quickly.

I disagree completely.

This patch adds a user-space API that we will have to carry
with us for perpetuity.  I would only support this if we had
no other way around the problem.

If this does turn out to be mostly due to sendmsg contention
then fixing it is going to be much simpler than making the UDP
stack multiqueue capable.

I'm working on this right now.

Cheers,
-- 
Email: Herbert Xu [off-list ref]
Home Page: http://gondor.apana.org.au/~herbert/
PGP Key: http://gondor.apana.org.au/~herbert/pubkey.txt

I disagree completely.

This patch adds a user-space API that we will have to carry
with us for perpetuity.  I would only support this if we had
no other way around the problem.

If this does turn out to be mostly due to sendmsg contention
then fixing it is going to be much simpler than making the UDP
stack multiqueue capable.

That sounds promising, but receive side will still have problems.
There is lock contention on the queue as well as cache line bouncing
on the sock structures.  Also multiple threads sleeping on same socket
typically leads to asymmetric load across the threads (and
degenerative cases where receiving thread is woken up and other
threads have already processed all the packets).  TCP listener threads
suffer from these same problems.

Tom

I'm working on this right now.

Cheers,
--
Email: Herbert Xu [off-list ref]
Home Page: http://gondor.apana.org.au/~herbert/
PGP Key: http://gondor.apana.org.au/~herbert/pubkey.txt

On Sun, Feb 27, 2011 at 07:45:55PM -0800, Tom Herbert wrote:

That sounds promising, but receive side will still have problems.
There is lock contention on the queue as well as cache line bouncing
on the sock structures.  Also multiple threads sleeping on same socket
typically leads to asymmetric load across the threads (and
degenerative cases where receiving thread is woken up and other
threads have already processed all the packets).  TCP listener threads
suffer from these same problems.

IOW this is something that we have to solve anyway.  I'm just
being overly cautious here because user-space API changes are
something that we should not enter into lightly.

If this patch was completely internal to the kernel, then I would
have much less of an objection as we can always revert/change it
later on.  With a user-space API we don't have that flexibility.

Thanks,
-- 
Email: Herbert Xu [off-list ref]
Home Page: http://gondor.apana.org.au/~herbert/
PGP Key: http://gondor.apana.org.au/~herbert/pubkey.txt

On Sun, Feb 27, 2011 at 07:06:14PM +0800, Herbert Xu wrote:

I'm working on this right now.

OK I think I was definitely on the right track.  With the send
patch made lockless I now get numbers which are even better than
those obtained with running named with multiple sockets.  That's
right, a single socket is now faster than what multiple sockets
were without the patch (of course, multiple sockets may still
faster with the patch vs. a single socket for obvious reasons,
but I couldn't measure any significant difference).

Also worthy of note is that prior to the patch all CPUs showed
idleness (lazy bastards!), with the patch they're all maxed out.

In retrospect, the idleness was simply the result of the socket
lock scheduling away and was an indication of lock contention.

Here are the patches I used.  Please don't them yet as I intend
to clean them up quite a bit.

But please do test them heavily, especially if you have an AMD
NUMA machine as that's where scalability problems really show
up.  Intel tends to be a lot more forgiving.  My last AMD machine
blew up years ago :)

Thanks!
-- 
Email: Herbert Xu [off-list ref]
Home Page: http://gondor.apana.org.au/~herbert/
PGP Key: http://gondor.apana.org.au/~herbert/pubkey.txt

inet: Add ip_make_skb and ip_send_skb

This patch adds the helper ip_make_skb which is like ip_append_data
and ip_push_pending_frames all rolled into one, except that it does
not send the skb produced.  The sending part is carried out by
ip_send_skb, which the transport protocol can call after it has
tweaked the skb.

It is meant to be called in cases where corking is not used should
have a one-to-one correspondence to sendmsg.

Signed-off-by: Herbert Xu <herbert@gondor.apana.org.au>
---

 include/net/ip.h     |    8 ++++++
 net/ipv4/ip_output.c |   65 ++++++++++++++++++++++++++++++++++++++++-----------
 2 files changed, 59 insertions(+), 14 deletions(-)

diff --git a/include/net/ip.h b/include/net/ip.h
index 67fac78..a96e525 100644
--- a/include/net/ip.h
+++ b/include/net/ip.h

@@ -116,8 +116,16 @@ extern int		ip_append_data(struct sock *sk,
 extern int		ip_generic_getfrag(void *from, char *to, int offset, int len, int odd, struct sk_buff *skb);
 extern ssize_t		ip_append_page(struct sock *sk, struct page *page,
 				int offset, size_t size, int flags);
+extern int		ip_send_skb(struct sk_buff *skb);
 extern int		ip_push_pending_frames(struct sock *sk);
 extern void		ip_flush_pending_frames(struct sock *sk);
+extern struct sk_buff  *ip_make_skb(struct sock *sk,
+				    int getfrag(void *from, char *to, int offset, int len,
+						int odd, struct sk_buff *skb),
+				    void *from, int length, int transhdrlen,
+				    struct ipcm_cookie *ipc,
+				    struct rtable **rtp,
+				    unsigned int flags);
 
 /* datagram.c */
 extern int		ip4_datagram_connect(struct sock *sk, 

diff --git a/net/ipv4/ip_output.c b/net/ipv4/ip_output.c
index 1dd5ecc..dba14c6 100644
--- a/net/ipv4/ip_output.c
+++ b/net/ipv4/ip_output.c

@@ -1267,9 +1267,9 @@ static void ip_cork_release(struct inet_cork *cork)
  *	Combined all pending IP fragments on the socket as one IP datagram
  *	and push them out.
  */
-static int __ip_push_pending_frames(struct sock *sk,
-				    struct sk_buff_head *queue,
-				    struct inet_cork *cork)
+static struct sk_buff *__ip_make_skb(struct sock *sk,
+				     struct sk_buff_head *queue,
+				     struct inet_cork *cork)
 {
 	struct sk_buff *skb, *tmp_skb;
 	struct sk_buff **tail_skb;

@@ -1280,7 +1280,6 @@ static int __ip_push_pending_frames(struct sock *sk,
 	struct iphdr *iph;
 	__be16 df = 0;
 	__u8 ttl;
-	int err = 0;
 
 	if ((skb = __skb_dequeue(queue)) == NULL)
 		goto out;

@@ -1351,28 +1350,37 @@ static int __ip_push_pending_frames(struct sock *sk,
 		icmp_out_count(net, ((struct icmphdr *)
 			skb_transport_header(skb))->type);
 
-	/* Netfilter gets whole the not fragmented skb. */
+	ip_cork_release(cork);
+out:
+	return skb;
+}
+
+int ip_send_skb(struct sk_buff *skb)
+{
+	struct net *net = sock_net(skb->sk);
+	int err;
+
 	err = ip_local_out(skb);
 	if (err) {
 		if (err > 0)
 			err = net_xmit_errno(err);
 		if (err)
-			goto error;
+			IP_INC_STATS(net, IPSTATS_MIB_OUTDISCARDS);
 	}
 
-out:
-	ip_cork_release(cork);
 	return err;
-
-error:
-	IP_INC_STATS(net, IPSTATS_MIB_OUTDISCARDS);
-	goto out;
 }
 
 int ip_push_pending_frames(struct sock *sk)
 {
-	return __ip_push_pending_frames(sk, &sk->sk_write_queue,
-					&inet_sk(sk)->cork);
+	struct sk_buff *skb;
+
+	skb = __ip_make_skb(sk, &sk->sk_write_queue, &inet_sk(sk)->cork);
+	if (!skb)
+		return 0;
+
+	/* Netfilter gets whole the not fragmented skb. */
+	return ip_send_skb(skb);
 }
 
 /*

@@ -1395,6 +1403,35 @@ void ip_flush_pending_frames(struct sock *sk)
 	__ip_flush_pending_frames(sk, &sk->sk_write_queue, &inet_sk(sk)->cork);
 }
 
+struct sk_buff *ip_make_skb(struct sock *sk,
+			    int getfrag(void *from, char *to, int offset,
+					int len, int odd, struct sk_buff *skb),
+			    void *from, int length, int transhdrlen,
+			    struct ipcm_cookie *ipc, struct rtable **rtp,
+			    unsigned int flags)
+{
+	struct inet_cork cork = {};
+	struct sk_buff_head queue;
+	int err;
+
+	if (flags & MSG_PROBE)
+		return NULL;
+
+	__skb_queue_head_init(&queue);
+
+	err = ip_setup_cork(sk, &cork, ipc, rtp);
+	if (err)
+		return ERR_PTR(err);
+
+	err = __ip_append_data(sk, &queue, &cork, getfrag,
+			       from, length, transhdrlen, flags);
+	if (err) {
+		__ip_flush_pending_frames(sk, &queue, &cork);
+		return ERR_PTR(err);
+	}
+
+	return __ip_make_skb(sk, &queue, &cork);
+}
 
 /*
  *	Fetch data from kernel space and fill in checksum if needed.

net: Remove unused sk_sndmsg_* from UFO

UFO doesn't really use the sk_sndmsg_* parameters so touching
them is pointless.  It can't use them anyway since the whole
point of UFO is to use the original pages without copying.

Signed-off-by: Herbert Xu <herbert@gondor.apana.org.au>
---

 net/core/skbuff.c     |    3 ---
 net/ipv4/ip_output.c  |    1 -
 net/ipv6/ip6_output.c |    1 -
 3 files changed, 5 deletions(-)

diff --git a/net/core/skbuff.c b/net/core/skbuff.c
index d883dcc..97011a7 100644
--- a/net/core/skbuff.c
+++ b/net/core/skbuff.c

@@ -2434,8 +2434,6 @@ int skb_append_datato_frags(struct sock *sk, struct sk_buff *skb,
 			return -ENOMEM;
 
 		/* initialize the next frag */
-		sk->sk_sndmsg_page = page;
-		sk->sk_sndmsg_off = 0;
 		skb_fill_page_desc(skb, frg_cnt, page, 0, 0);
 		skb->truesize += PAGE_SIZE;
 		atomic_add(PAGE_SIZE, &sk->sk_wmem_alloc);

@@ -2455,7 +2453,6 @@ int skb_append_datato_frags(struct sock *sk, struct sk_buff *skb,
 			return -EFAULT;
 
 		/* copy was successful so update the size parameters */
-		sk->sk_sndmsg_off += copy;
 		frag->size += copy;
 		skb->len += copy;
 		skb->data_len += copy;

diff --git a/net/ipv4/ip_output.c b/net/ipv4/ip_output.c
index 04c7b3b..d3a4540 100644
--- a/net/ipv4/ip_output.c
+++ b/net/ipv4/ip_output.c

@@ -767,7 +767,6 @@ static inline int ip_ufo_append_data(struct sock *sk,
 
 		skb->ip_summed = CHECKSUM_PARTIAL;
 		skb->csum = 0;
-		sk->sk_sndmsg_off = 0;
 
 		/* specify the length of each IP datagram fragment */
 		skb_shinfo(skb)->gso_size = mtu - fragheaderlen;

diff --git a/net/ipv6/ip6_output.c b/net/ipv6/ip6_output.c
index 5f8d242..9965182 100644
--- a/net/ipv6/ip6_output.c
+++ b/net/ipv6/ip6_output.c

@@ -1061,7 +1061,6 @@ static inline int ip6_ufo_append_data(struct sock *sk,
 
 		skb->ip_summed = CHECKSUM_PARTIAL;
 		skb->csum = 0;
-		sk->sk_sndmsg_off = 0;
 	}
 
 	err = skb_append_datato_frags(sk,skb, getfrag, from,

udp: Add lockless transmit path

The UDP transmit path has been running under the socket lock
for a long time because of the corking feature.  This means that
transmitting to the same socket in multiple threads does not
scale at all.

However, as most users don't actually use corking, the locking
can be removed in the common case.

This patch creates a lockless fast path where corking is not used.

Please note that this does create a slight inaccuracy in the
enforcement of socket send buffer limits.  In particular, we
may exceed the socket limit by up to (number of CPUs) * (packet
size) because of the way the limit is computed.

As the primary purpose of socket buffers is to indicate congestion,
this should not be a great problem for now.

Signed-off-by: Herbert Xu <herbert@gondor.apana.org.au>
---

 include/net/udp.h     |   11 +++++
 include/net/udplite.h |   12 +++++
 net/ipv4/udp.c        |  104 +++++++++++++++++++++++++++++++++++++++++++++++++-
 3 files changed, 126 insertions(+), 1 deletion(-)

diff --git a/include/net/udp.h b/include/net/udp.h
index bb967dd..b8563ba 100644
--- a/include/net/udp.h
+++ b/include/net/udp.h

@@ -144,6 +144,17 @@ static inline __wsum udp_csum_outgoing(struct sock *sk, struct sk_buff *skb)
 	return csum;
 }
 
+static inline __wsum udp_csum(struct sk_buff *skb)
+{
+	__wsum csum = csum_partial(skb_transport_header(skb),
+				   sizeof(struct udphdr), skb->csum);
+
+	for (skb = skb_shinfo(skb)->frag_list; skb; skb = skb->next) {
+		csum = csum_add(csum, skb->csum);
+	}
+	return csum;
+}
+
 /* hash routines shared between UDPv4/6 and UDP-Litev4/6 */
 static inline void udp_lib_hash(struct sock *sk)
 {

diff --git a/include/net/udplite.h b/include/net/udplite.h
index afdffe6..673a024 100644
--- a/include/net/udplite.h
+++ b/include/net/udplite.h

@@ -115,6 +115,18 @@ static inline __wsum udplite_csum_outgoing(struct sock *sk, struct sk_buff *skb)
 	return csum;
 }
 
+static inline __wsum udplite_csum(struct sk_buff *skb)
+{
+	struct sock *sk = skb->sk;
+	int cscov = udplite_sender_cscov(udp_sk(sk), udp_hdr(skb));
+	const int off = skb_transport_offset(skb);
+	const int len = skb->len - off;
+
+	skb->ip_summed = CHECKSUM_NONE;     /* no HW support for checksumming */
+
+	return skb_checksum(skb, off, min(cscov, len), 0);
+}
+
 extern void	udplite4_register(void);
 extern int 	udplite_get_port(struct sock *sk, unsigned short snum,
 			int (*scmp)(const struct sock *, const struct sock *));

diff --git a/net/ipv4/udp.c b/net/ipv4/udp.c
index 8157b17..7fd3664 100644
--- a/net/ipv4/udp.c
+++ b/net/ipv4/udp.c

@@ -769,6 +769,95 @@ out:
 	return err;
 }
 
+static void udp4_hwcsum(struct sk_buff *skb, __be32 src, __be32 dst)
+{
+	struct udphdr *uh = udp_hdr(skb);
+	struct sk_buff *frags = skb_shinfo(skb)->frag_list;
+	int offset = skb_transport_offset(skb);
+	int len = skb->len - offset;
+	int hlen = len;
+	__wsum csum = 0;
+
+	if (!frags) {
+		/*
+		 * Only one fragment on the socket.
+		 */
+		skb->csum_start = skb_transport_header(skb) - skb->head;
+		skb->csum_offset = offsetof(struct udphdr, check);
+		uh->check = ~csum_tcpudp_magic(src, dst, len,
+					       IPPROTO_UDP, 0);
+	} else {
+		/*
+		 * HW-checksum won't work as there are two or more
+		 * fragments on the socket so that all csums of sk_buffs
+		 * should be together
+		 */
+		do {
+			csum = csum_add(csum, frags->csum);
+			hlen -= frags->len;
+		} while ((frags = frags->next));
+
+		csum = skb_checksum(skb, offset, hlen, csum);
+		skb->ip_summed = CHECKSUM_NONE;
+
+		uh->check = csum_tcpudp_magic(src, dst, len, IPPROTO_UDP, csum);
+		if (uh->check == 0)
+			uh->check = CSUM_MANGLED_0;
+	}
+}
+
+static int udp_send_skb(struct sk_buff *skb, __be32 daddr, __be32 dport)
+{
+	struct sock *sk = skb->sk;
+	struct inet_sock *inet = inet_sk(sk);
+	struct udphdr *uh;
+	struct rtable *rt = (struct rtable *)skb_dst(skb);
+	int err = 0;
+	int is_udplite = IS_UDPLITE(sk);
+	int offset = skb_transport_offset(skb);
+	int len = skb->len - offset;
+	__wsum csum = 0;
+
+	/*
+	 * Create a UDP header
+	 */
+	uh = udp_hdr(skb);
+	uh->source = inet->inet_sport;
+	uh->dest = dport;
+	uh->len = htons(len);
+	uh->check = 0;
+
+	if (is_udplite)
+		csum = udplite_csum(skb);
+	else if (sk->sk_no_check == UDP_CSUM_NOXMIT) {
+		skb->ip_summed = CHECKSUM_NONE;
+		goto send;
+	} else if (skb->ip_summed == CHECKSUM_PARTIAL) {
+		udp4_hwcsum(skb, rt->rt_src, daddr);
+		goto send;
+	} else
+		csum = udp_csum(skb);
+
+	/* add protocol-dependent pseudo-header */
+	uh->check = csum_tcpudp_magic(rt->rt_src, daddr, len,
+				      sk->sk_protocol, csum);
+	if (uh->check == 0)
+		uh->check = CSUM_MANGLED_0;
+
+send:
+	err = ip_send_skb(skb);
+	if (err) {
+		if (err == -ENOBUFS && !inet->recverr) {
+			UDP_INC_STATS_USER(sock_net(sk),
+					   UDP_MIB_SNDBUFERRORS, is_udplite);
+			err = 0;
+		}
+	} else
+		UDP_INC_STATS_USER(sock_net(sk),
+				   UDP_MIB_OUTDATAGRAMS, is_udplite);
+	return err;
+}
+
 int udp_sendmsg(struct kiocb *iocb, struct sock *sk, struct msghdr *msg,
 		size_t len)
 {

@@ -785,6 +874,7 @@ int udp_sendmsg(struct kiocb *iocb, struct sock *sk, struct msghdr *msg,
 	int err, is_udplite = IS_UDPLITE(sk);
 	int corkreq = up->corkflag || msg->msg_flags&MSG_MORE;
 	int (*getfrag)(void *, char *, int, int, int, struct sk_buff *);
+	struct sk_buff *skb;
 
 	if (len > 0xFFFF)
 		return -EMSGSIZE;

@@ -799,6 +889,8 @@ int udp_sendmsg(struct kiocb *iocb, struct sock *sk, struct msghdr *msg,
 	ipc.opt = NULL;
 	ipc.tx_flags = 0;
 
+	getfrag = is_udplite ? udplite_getfrag : ip_generic_getfrag;
+
 	if (up->pending) {
 		/*
 		 * There are pending frames.

@@ -923,6 +1015,17 @@ back_from_confirm:
 	if (!ipc.addr)
 		daddr = ipc.addr = rt->rt_dst;
 
+	/* Lockless fast path for the non-corking case. */
+	if (!corkreq) {
+		skb = ip_make_skb(sk, getfrag, msg->msg_iov, ulen,
+				  sizeof(struct udphdr), &ipc, &rt,
+				  msg->msg_flags);
+		err = PTR_ERR(skb);
+		if (skb && !IS_ERR(skb))
+			err = udp_send_skb(skb, daddr, dport);
+		goto out;
+	}
+
 	lock_sock(sk);
 	if (unlikely(up->pending)) {
 		/* The socket is already corked while preparing it. */

@@ -944,7 +1047,6 @@ back_from_confirm:
 
 do_append_data:
 	up->len += ulen;
-	getfrag  =  is_udplite ?  udplite_getfrag : ip_generic_getfrag;
 	err = ip_append_data(sk, getfrag, msg->msg_iov, ulen,
 			sizeof(struct udphdr), &ipc, &rt,
 			corkreq ? msg->msg_flags|MSG_MORE : msg->msg_flags);

net: Remove explicit write references to sk/inet in ip_append_data

In order to allow simultaneous calls to ip_append_data on the same
socket, it must not modify any shared state in sk or inet (other
than those that are designed to allow that such as atomic counters).

This patch abstracts out write references to sk and inet_sk in
ip_append_data and its friends so that we may use the underlying
code in parallel.

Signed-off-by: Herbert Xu <herbert@gondor.apana.org.au>
---

 include/net/inet_sock.h |   23 ++--
 net/ipv4/ip_output.c    |  238 ++++++++++++++++++++++++++++--------------------
 2 files changed, 154 insertions(+), 107 deletions(-)

diff --git a/include/net/inet_sock.h b/include/net/inet_sock.h
index 8181498..b3de102 100644
--- a/include/net/inet_sock.h
+++ b/include/net/inet_sock.h

@@ -86,6 +86,19 @@ static inline struct inet_request_sock *inet_rsk(const struct request_sock *sk)
 	return (struct inet_request_sock *)sk;
 }
 
+struct inet_cork {
+	unsigned int		flags;
+	unsigned int		fragsize;
+	struct ip_options	*opt;
+	struct dst_entry	*dst;
+	int			length; /* Total length of all frames */
+	__be32			addr;
+	struct flowi		fl;
+	struct page		*page;
+	u32			off;
+	u8			tx_flags;
+};
+
 struct ip_mc_socklist;
 struct ipv6_pinfo;
 struct rtable;

@@ -143,15 +156,7 @@ struct inet_sock {
 	int			mc_index;
 	__be32			mc_addr;
 	struct ip_mc_socklist __rcu	*mc_list;
-	struct {
-		unsigned int		flags;
-		unsigned int		fragsize;
-		struct ip_options	*opt;
-		struct dst_entry	*dst;
-		int			length; /* Total length of all frames */
-		__be32			addr;
-		struct flowi		fl;
-	} cork;
+	struct inet_cork	cork;
 };
 
 #define IPCORK_OPT	1	/* ip-options has been held in ipcork.opt */

diff --git a/net/ipv4/ip_output.c b/net/ipv4/ip_output.c
index d3a4540..1dd5ecc 100644
--- a/net/ipv4/ip_output.c
+++ b/net/ipv4/ip_output.c

@@ -733,6 +733,7 @@ csum_page(struct page *page, int offset, int copy)
 }
 
 static inline int ip_ufo_append_data(struct sock *sk,
+			struct sk_buff_head *queue,
 			int getfrag(void *from, char *to, int offset, int len,
 			       int odd, struct sk_buff *skb),
 			void *from, int length, int hh_len, int fragheaderlen,

@@ -745,7 +746,7 @@ static inline int ip_ufo_append_data(struct sock *sk,
 	 * device, so create one single skb packet containing complete
 	 * udp datagram
 	 */
-	if ((skb = skb_peek_tail(&sk->sk_write_queue)) == NULL) {
+	if ((skb = skb_peek_tail(queue)) == NULL) {
 		skb = sock_alloc_send_skb(sk,
 			hh_len + fragheaderlen + transhdrlen + 20,
 			(flags & MSG_DONTWAIT), &err);

@@ -771,35 +772,24 @@ static inline int ip_ufo_append_data(struct sock *sk,
 		/* specify the length of each IP datagram fragment */
 		skb_shinfo(skb)->gso_size = mtu - fragheaderlen;
 		skb_shinfo(skb)->gso_type = SKB_GSO_UDP;
-		__skb_queue_tail(&sk->sk_write_queue, skb);
+		__skb_queue_tail(queue, skb);
 	}
 
 	return skb_append_datato_frags(sk, skb, getfrag, from,
 				       (length - transhdrlen));
 }
 
-/*
- *	ip_append_data() and ip_append_page() can make one large IP datagram
- *	from many pieces of data. Each pieces will be holded on the socket
- *	until ip_push_pending_frames() is called. Each piece can be a page
- *	or non-page data.
- *
- *	Not only UDP, other transport protocols - e.g. raw sockets - can use
- *	this interface potentially.
- *
- *	LATER: length must be adjusted by pad at tail, when it is required.
- */
-int ip_append_data(struct sock *sk,
-		   int getfrag(void *from, char *to, int offset, int len,
-			       int odd, struct sk_buff *skb),
-		   void *from, int length, int transhdrlen,
-		   struct ipcm_cookie *ipc, struct rtable **rtp,
-		   unsigned int flags)
+static int __ip_append_data(struct sock *sk, struct sk_buff_head *queue,
+			    struct inet_cork *cork,
+			    int getfrag(void *from, char *to, int offset,
+					int len, int odd, struct sk_buff *skb),
+			    void *from, int length, int transhdrlen,
+			    unsigned int flags)
 {
 	struct inet_sock *inet = inet_sk(sk);
 	struct sk_buff *skb;
 
-	struct ip_options *opt = NULL;
+	struct ip_options *opt = inet->cork.opt;
 	int hh_len;
 	int exthdrlen;
 	int mtu;

@@ -808,58 +798,19 @@ int ip_append_data(struct sock *sk,
 	int offset = 0;
 	unsigned int maxfraglen, fragheaderlen;
 	int csummode = CHECKSUM_NONE;
-	struct rtable *rt;
-
-	if (flags&MSG_PROBE)
-		return 0;
+	struct rtable *rt = (struct rtable *)cork->dst;
 
-	if (skb_queue_empty(&sk->sk_write_queue)) {
-		/*
-		 * setup for corking.
-		 */
-		opt = ipc->opt;
-		if (opt) {
-			if (inet->cork.opt == NULL) {
-				inet->cork.opt = kmalloc(sizeof(struct ip_options) + 40, sk->sk_allocation);
-				if (unlikely(inet->cork.opt == NULL))
-					return -ENOBUFS;
-			}
-			memcpy(inet->cork.opt, opt, sizeof(struct ip_options)+opt->optlen);
-			inet->cork.flags |= IPCORK_OPT;
-			inet->cork.addr = ipc->addr;
-		}
-		rt = *rtp;
-		if (unlikely(!rt))
-			return -EFAULT;
-		/*
-		 * We steal reference to this route, caller should not release it
-		 */
-		*rtp = NULL;
-		inet->cork.fragsize = mtu = inet->pmtudisc == IP_PMTUDISC_PROBE ?
-					    rt->dst.dev->mtu :
-					    dst_mtu(rt->dst.path);
-		inet->cork.dst = &rt->dst;
-		inet->cork.length = 0;
-		sk->sk_sndmsg_page = NULL;
-		sk->sk_sndmsg_off = 0;
-		exthdrlen = rt->dst.header_len;
-		length += exthdrlen;
-		transhdrlen += exthdrlen;
-	} else {
-		rt = (struct rtable *)inet->cork.dst;
-		if (inet->cork.flags & IPCORK_OPT)
-			opt = inet->cork.opt;
+	exthdrlen = transhdrlen ? rt->dst.header_len : 0;
+	length += exthdrlen;
+	transhdrlen += exthdrlen;
+	mtu = inet->cork.fragsize;
 
-		transhdrlen = 0;
-		exthdrlen = 0;
-		mtu = inet->cork.fragsize;
-	}
 	hh_len = LL_RESERVED_SPACE(rt->dst.dev);
 
 	fragheaderlen = sizeof(struct iphdr) + (opt ? opt->optlen : 0);
 	maxfraglen = ((mtu - fragheaderlen) & ~7) + fragheaderlen;
 
-	if (inet->cork.length + length > 0xFFFF - fragheaderlen) {
+	if (cork->length + length > 0xFFFF - fragheaderlen) {
 		ip_local_error(sk, EMSGSIZE, rt->rt_dst, inet->inet_dport,
 			       mtu-exthdrlen);
 		return -EMSGSIZE;

@@ -875,15 +826,15 @@ int ip_append_data(struct sock *sk,
 	    !exthdrlen)
 		csummode = CHECKSUM_PARTIAL;
 
-	skb = skb_peek_tail(&sk->sk_write_queue);
+	skb = skb_peek_tail(queue);
 
-	inet->cork.length += length;
+	cork->length += length;
 	if (((length > mtu) || (skb && skb_is_gso(skb))) &&
 	    (sk->sk_protocol == IPPROTO_UDP) &&
 	    (rt->dst.dev->features & NETIF_F_UFO)) {
-		err = ip_ufo_append_data(sk, getfrag, from, length, hh_len,
-					 fragheaderlen, transhdrlen, mtu,
-					 flags);
+		err = ip_ufo_append_data(sk, queue, getfrag, from, length,
+					 hh_len, fragheaderlen, transhdrlen,
+					 mtu, flags);
 		if (err)
 			goto error;
 		return 0;

@@ -960,7 +911,7 @@ alloc_new_skb:
 				else
 					/* only the initial fragment is
 					   time stamped */
-					ipc->tx_flags = 0;
+					cork->tx_flags = 0;
 			}
 			if (skb == NULL)
 				goto error;

@@ -971,7 +922,7 @@ alloc_new_skb:
 			skb->ip_summed = csummode;
 			skb->csum = 0;
 			skb_reserve(skb, hh_len);
-			skb_shinfo(skb)->tx_flags = ipc->tx_flags;
+			skb_shinfo(skb)->tx_flags = cork->tx_flags;
 
 			/*
 			 *	Find where to start putting bytes.

@@ -1008,7 +959,7 @@ alloc_new_skb:
 			/*
 			 * Put the packet on the pending queue.
 			 */
-			__skb_queue_tail(&sk->sk_write_queue, skb);
+			__skb_queue_tail(queue, skb);
 			continue;
 		}
 

@@ -1028,8 +979,8 @@ alloc_new_skb:
 		} else {
 			int i = skb_shinfo(skb)->nr_frags;
 			skb_frag_t *frag = &skb_shinfo(skb)->frags[i-1];
-			struct page *page = sk->sk_sndmsg_page;
-			int off = sk->sk_sndmsg_off;
+			struct page *page = cork->page;
+			int off = cork->off;
 			unsigned int left;
 
 			if (page && (left = PAGE_SIZE - off) > 0) {

@@ -1041,7 +992,7 @@ alloc_new_skb:
 						goto error;
 					}
 					get_page(page);
-					skb_fill_page_desc(skb, i, page, sk->sk_sndmsg_off, 0);
+					skb_fill_page_desc(skb, i, page, off, 0);
 					frag = &skb_shinfo(skb)->frags[i];
 				}
 			} else if (i < MAX_SKB_FRAGS) {

@@ -1052,8 +1003,8 @@ alloc_new_skb:
 					err = -ENOMEM;
 					goto error;
 				}
-				sk->sk_sndmsg_page = page;
-				sk->sk_sndmsg_off = 0;
+				cork->page = page;
+				cork->off = 0;
 
 				skb_fill_page_desc(skb, i, page, 0, 0);
 				frag = &skb_shinfo(skb)->frags[i];

@@ -1065,7 +1016,7 @@ alloc_new_skb:
 				err = -EFAULT;
 				goto error;
 			}
-			sk->sk_sndmsg_off += copy;
+			cork->off += copy;
 			frag->size += copy;
 			skb->len += copy;
 			skb->data_len += copy;

@@ -1079,11 +1030,87 @@ alloc_new_skb:
 	return 0;
 
 error:
-	inet->cork.length -= length;
+	cork->length -= length;
 	IP_INC_STATS(sock_net(sk), IPSTATS_MIB_OUTDISCARDS);
 	return err;
 }
 
+static int ip_setup_cork(struct sock *sk, struct inet_cork *cork,
+			 struct ipcm_cookie *ipc, struct rtable **rtp)
+{
+	struct inet_sock *inet = inet_sk(sk);
+	struct ip_options *opt;
+	struct rtable *rt;
+
+	/*
+	 * setup for corking.
+	 */
+	opt = ipc->opt;
+	if (opt) {
+		if (cork->opt == NULL) {
+			cork->opt = kmalloc(sizeof(struct ip_options) + 40,
+					    sk->sk_allocation);
+			if (unlikely(cork->opt == NULL))
+				return -ENOBUFS;
+		}
+		memcpy(cork->opt, opt, sizeof(struct ip_options) + opt->optlen);
+		cork->flags |= IPCORK_OPT;
+		cork->addr = ipc->addr;
+	}
+	rt = *rtp;
+	if (unlikely(!rt))
+		return -EFAULT;
+	/*
+	 * We steal reference to this route, caller should not release it
+	 */
+	*rtp = NULL;
+	cork->fragsize = inet->pmtudisc == IP_PMTUDISC_PROBE ?
+			 rt->dst.dev->mtu : dst_mtu(rt->dst.path);
+	cork->dst = &rt->dst;
+	cork->length = 0;
+	cork->tx_flags = ipc->tx_flags;
+	cork->page = NULL;
+	cork->off = 0;
+
+	return 0;
+}
+
+/*
+ *	ip_append_data() and ip_append_page() can make one large IP datagram
+ *	from many pieces of data. Each pieces will be holded on the socket
+ *	until ip_push_pending_frames() is called. Each piece can be a page
+ *	or non-page data.
+ *
+ *	Not only UDP, other transport protocols - e.g. raw sockets - can use
+ *	this interface potentially.
+ *
+ *	LATER: length must be adjusted by pad at tail, when it is required.
+ */
+int ip_append_data(struct sock *sk,
+		   int getfrag(void *from, char *to, int offset, int len,
+			       int odd, struct sk_buff *skb),
+		   void *from, int length, int transhdrlen,
+		   struct ipcm_cookie *ipc, struct rtable **rtp,
+		   unsigned int flags)
+{
+	struct inet_sock *inet = inet_sk(sk);
+	int err;
+
+	if (flags&MSG_PROBE)
+		return 0;
+
+	if (skb_queue_empty(&sk->sk_write_queue)) {
+		err = ip_setup_cork(sk, &inet->cork, ipc, rtp);
+		if (err)
+			return err;
+	} else {
+		transhdrlen = 0;
+	}
+
+	return __ip_append_data(sk, &sk->sk_write_queue, &inet->cork, getfrag,
+				from, length, transhdrlen, flags);
+}
+
 ssize_t	ip_append_page(struct sock *sk, struct page *page,
 		       int offset, size_t size, int flags)
 {

@@ -1227,40 +1254,42 @@ error:
 	return err;
 }
 
-static void ip_cork_release(struct inet_sock *inet)
+static void ip_cork_release(struct inet_cork *cork)
 {
-	inet->cork.flags &= ~IPCORK_OPT;
-	kfree(inet->cork.opt);
-	inet->cork.opt = NULL;
-	dst_release(inet->cork.dst);
-	inet->cork.dst = NULL;
+	cork->flags &= ~IPCORK_OPT;
+	kfree(cork->opt);
+	cork->opt = NULL;
+	dst_release(cork->dst);
+	cork->dst = NULL;
 }
 
 /*
  *	Combined all pending IP fragments on the socket as one IP datagram
  *	and push them out.
  */
-int ip_push_pending_frames(struct sock *sk)
+static int __ip_push_pending_frames(struct sock *sk,
+				    struct sk_buff_head *queue,
+				    struct inet_cork *cork)
 {
 	struct sk_buff *skb, *tmp_skb;
 	struct sk_buff **tail_skb;
 	struct inet_sock *inet = inet_sk(sk);
 	struct net *net = sock_net(sk);
 	struct ip_options *opt = NULL;
-	struct rtable *rt = (struct rtable *)inet->cork.dst;
+	struct rtable *rt = (struct rtable *)cork->dst;
 	struct iphdr *iph;
 	__be16 df = 0;
 	__u8 ttl;
 	int err = 0;
 
-	if ((skb = __skb_dequeue(&sk->sk_write_queue)) == NULL)
+	if ((skb = __skb_dequeue(queue)) == NULL)
 		goto out;
 	tail_skb = &(skb_shinfo(skb)->frag_list);
 
 	/* move skb->data to ip header from ext header */
 	if (skb->data < skb_network_header(skb))
 		__skb_pull(skb, skb_network_offset(skb));
-	while ((tmp_skb = __skb_dequeue(&sk->sk_write_queue)) != NULL) {
+	while ((tmp_skb = __skb_dequeue(queue)) != NULL) {
 		__skb_pull(tmp_skb, skb_network_header_len(skb));
 		*tail_skb = tmp_skb;
 		tail_skb = &(tmp_skb->next);

@@ -1286,8 +1315,8 @@ int ip_push_pending_frames(struct sock *sk)
 	     ip_dont_fragment(sk, &rt->dst)))
 		df = htons(IP_DF);
 
-	if (inet->cork.flags & IPCORK_OPT)
-		opt = inet->cork.opt;
+	if (cork->flags & IPCORK_OPT)
+		opt = cork->opt;
 
 	if (rt->rt_type == RTN_MULTICAST)
 		ttl = inet->mc_ttl;

@@ -1299,7 +1328,7 @@ int ip_push_pending_frames(struct sock *sk)
 	iph->ihl = 5;
 	if (opt) {
 		iph->ihl += opt->optlen>>2;
-		ip_options_build(skb, opt, inet->cork.addr, rt, 0);
+		ip_options_build(skb, opt, cork->addr, rt, 0);
 	}
 	iph->tos = inet->tos;
 	iph->frag_off = df;