hesoyam/Xray-core
1
0
Fork 0
mirror of https://github.com/XTLS/Xray-core.git synced 2026-10-17 04:53:59 +00:00
Code Issues Packages Projects Releases Wiki Activity

WireGuard, MASQUE proxies: Use Xray-core's built-in lightweight TCP/IP stack instead of gVisor as well (#7156)

Browse Source 
https://github.com/XTLS/Xray-core/pull/7130#issuecomment-6116258188
https://github.com/XTLS/Xray-core/pull/7130#issuecomment-6120742739
https://github.com/XTLS/Xray-core/pull/7130#issuecomment-6137264458
https://github.com/XTLS/Xray-core/pull/7156#issuecomment-6202933664
https://github.com/XTLS/Xray-core/pull/7156#issuecomment-6203239736
This commit is contained in:
Artem Lytkin
2026-10-15 23:38:14 +00:00
committed by GitHub
parent 10879a4a8b
commit 8703fdb101
24 changed files with 3053 additions and 1430 deletions
Download Patch File Download Diff File Expand all files Collapse all files
proxy/tun/handler.go
+11 -10
View File
@@ -56,20 +56,21 @@ func (w *tunUDPStatsWriter) WriteMultiBuffer(mb buf.MultiBuffer) error {
return nil
}
// tunTCPWriter is the writing side of a TCP connection for the dispatcher. It has no Close: the outbound closes
// its writer when the other side has no more to send, and the system may still be sending
type tunTCPWriter struct {
writer buf.Writer
counter stats.Counter
// TCPWriter is the writing side of a TCP connection of the stack for the dispatcher. It has no Close: the outbound
// closes its writer when the other side has no more to send, and the system may still be sending
type TCPWriter struct {
Writer buf.Writer
// Counter counts what is written, if there is one
Counter stats.Counter
}
func (w *tunTCPWriter) WriteMultiBuffer(mb buf.MultiBuffer) error {
func (w *TCPWriter) WriteMultiBuffer(mb buf.MultiBuffer) error {
size := mb.Len()
if err := w.writer.WriteMultiBuffer(mb); err != nil {
if err := w.Writer.WriteMultiBuffer(mb); err != nil {
return err
}
if w.counter != nil {
w.counter.Add(int64(size))
if w.Counter != nil {
w.Counter.Add(int64(size))
}
return nil
}
@@ -243,7 +244,7 @@ func (t *Handler) HandleConnection(conn net.Conn, destination net.Destination) {
reader := &buf.TimeoutWrapperReader{Reader: buf.NewReader(conn), Counter: t.uplinkCounter}
writer := buf.NewWriter(conn)
if destination.Network != net.Network_UDP {
writer = &tunTCPWriter{writer: writer, counter: t.downlinkCounter}
writer = &TCPWriter{Writer: writer, Counter: t.downlinkCounter}
} else if t.downlinkCounter != nil {
writer = &tunUDPStatsWriter{writer: writer, counter: t.downlinkCounter}
}
proxy/tun/icmp/packet.go
-106
View File
@@ -1,106 +0,0 @@
package icmp
import (
"github.com/xtls/xray-core/common/errors"
"gvisor.dev/gvisor/pkg/tcpip"
"gvisor.dev/gvisor/pkg/tcpip/checksum"
"gvisor.dev/gvisor/pkg/tcpip/header"
)
func ProtocolLabel(netProto tcpip.NetworkProtocolNumber) string {
switch netProto {
case header.IPv4ProtocolNumber:
return "ipv4"
case header.IPv6ProtocolNumber:
return "ipv6"
default:
return "unknown"
}
}
func ParseEchoRequest(netProto tcpip.NetworkProtocolNumber, message []byte) (uint16, uint16, bool) {
switch netProto {
case header.IPv4ProtocolNumber:
if len(message) < header.ICMPv4MinimumSize {
return 0, 0, false
}
icmpHdr := header.ICMPv4(message)
if icmpHdr.Type() != header.ICMPv4Echo || icmpHdr.Code() != header.ICMPv4UnusedCode {
return 0, 0, false
}
return icmpHdr.Ident(), icmpHdr.Sequence(), true
case header.IPv6ProtocolNumber:
if len(message) < header.ICMPv6MinimumSize {
return 0, 0, false
}
icmpHdr := header.ICMPv6(message)
if icmpHdr.Type() != header.ICMPv6EchoRequest || icmpHdr.Code() != header.ICMPv6UnusedCode {
return 0, 0, false
}
return icmpHdr.Ident(), icmpHdr.Sequence(), true
default:
return 0, 0, false
}
}
func RewriteChecksum(netProto tcpip.NetworkProtocolNumber, message []byte, srcIP, dstIP tcpip.Address) error {
switch netProto {
case header.IPv4ProtocolNumber:
if len(message) < header.ICMPv4MinimumSize {
return errors.New("invalid icmpv4 packet")
}
icmpHdr := header.ICMPv4(message)
icmpHdr.SetChecksum(0)
icmpHdr.SetChecksum(header.ICMPv4Checksum(icmpHdr[:header.ICMPv4MinimumSize], checksum.Checksum(icmpHdr.Payload(), 0)))
return nil
case header.IPv6ProtocolNumber:
if len(message) < header.ICMPv6MinimumSize {
return errors.New("invalid icmpv6 packet")
}
icmpHdr := header.ICMPv6(message)
icmpHdr.SetChecksum(0)
icmpHdr.SetChecksum(header.ICMPv6Checksum(header.ICMPv6ChecksumParams{
Header: icmpHdr[:header.ICMPv6MinimumSize],
Src: srcIP,
Dst: dstIP,
PayloadCsum: checksum.Checksum(icmpHdr.Payload(), 0),
PayloadLen: len(icmpHdr.Payload()),
}))
return nil
default:
return errors.New("unsupported icmp network protocol")
}
}
func BuildLocalEchoReply(netProto tcpip.NetworkProtocolNumber, request []byte, srcIP, dstIP tcpip.Address) ([]byte, error) {
reply := append([]byte(nil), request...)
switch netProto {
case header.IPv4ProtocolNumber:
if len(reply) < header.ICMPv4MinimumSize {
return nil, errors.New("invalid icmpv4 echo packet")
}
icmpHdr := header.ICMPv4(reply)
if icmpHdr.Type() != header.ICMPv4Echo || icmpHdr.Code() != header.ICMPv4UnusedCode {
return nil, errors.New("not an icmpv4 echo request")
}
reply[0] = byte(header.ICMPv4EchoReply)
case header.IPv6ProtocolNumber:
if len(reply) < header.ICMPv6MinimumSize {
return nil, errors.New("invalid icmpv6 echo packet")
}
icmpHdr := header.ICMPv6(reply)
if icmpHdr.Type() != header.ICMPv6EchoRequest || icmpHdr.Code() != header.ICMPv6UnusedCode {
return nil, errors.New("not an icmpv6 echo request")
}
reply[0] = byte(header.ICMPv6EchoReply)
default:
return nil, errors.New("unsupported icmp network protocol")
}
if err := RewriteChecksum(netProto, reply, srcIP, dstIP); err != nil {
return nil, err
}
return reply, nil
}
proxy/tun/icmp/packet_test.go
-174
View File
@@ -1,174 +0,0 @@
package icmp
import (
"testing"
"gvisor.dev/gvisor/pkg/tcpip"
"gvisor.dev/gvisor/pkg/tcpip/checksum"
"gvisor.dev/gvisor/pkg/tcpip/header"
)
func TestParseEchoRequest(t *testing.T) {
t.Run("ipv4 echo", func(t *testing.T) {
var zero tcpip.Address
packet := []byte{
byte(header.ICMPv4Echo), 0,
0, 0,
0x12, 0x34,
0x56, 0x78,
0xaa, 0xbb,
}
if err := RewriteChecksum(header.IPv4ProtocolNumber, packet, zero, zero); err != nil {
t.Fatal(err)
}
ident, sequence, ok := ParseEchoRequest(header.IPv4ProtocolNumber, packet)
if !ok {
t.Fatal("expected ipv4 echo request to parse")
}
if ident != 0x1234 || sequence != 0x5678 {
t.Fatalf("unexpected ident/sequence: %x/%x", ident, sequence)
}
})
t.Run("ipv6 echo", func(t *testing.T) {
packet := []byte{
byte(header.ICMPv6EchoRequest), 0,
0, 0,
0xab, 0xcd,
0xef, 0x01,
0xaa, 0xbb,
}
src := tcpip.AddrFromSlice([]byte{0x20, 0x01, 0x0d, 0xb8, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 1})
dst := tcpip.AddrFromSlice([]byte{0x20, 0x01, 0x0d, 0xb8, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 2})
if err := RewriteChecksum(header.IPv6ProtocolNumber, packet, src, dst); err != nil {
t.Fatal(err)
}
ident, sequence, ok := ParseEchoRequest(header.IPv6ProtocolNumber, packet)
if !ok {
t.Fatal("expected ipv6 echo request to parse")
}
if ident != 0xabcd || sequence != 0xef01 {
t.Fatalf("unexpected ident/sequence: %x/%x", ident, sequence)
}
})
}
func TestRewriteChecksum(t *testing.T) {
t.Run("ipv4", func(t *testing.T) {
var zero tcpip.Address
packet := []byte{
byte(header.ICMPv4Echo), 0,
0xff, 0xff,
0x12, 0x34,
0x56, 0x78,
0xaa, 0xbb, 0xcc,
}
if err := RewriteChecksum(header.IPv4ProtocolNumber, packet, zero, zero); err != nil {
t.Fatal(err)
}
icmpHdr := header.ICMPv4(packet)
if got, want := icmpHdr.Checksum(), header.ICMPv4Checksum(icmpHdr[:header.ICMPv4MinimumSize], checksumPayloadV4(icmpHdr.Payload())); got != want {
t.Fatalf("unexpected ipv4 checksum: got %x want %x", got, want)
}
})
t.Run("ipv6", func(t *testing.T) {
packet := []byte{
byte(header.ICMPv6EchoReply), 0,
0xff, 0xff,
0x12, 0x34,
0x56, 0x78,
0xaa, 0xbb, 0xcc,
}
src := tcpip.AddrFromSlice([]byte{0x20, 0x01, 0x0d, 0xb8, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 1})
dst := tcpip.AddrFromSlice([]byte{0x20, 0x01, 0x0d, 0xb8, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 2})
if err := RewriteChecksum(header.IPv6ProtocolNumber, packet, src, dst); err != nil {
t.Fatal(err)
}
icmpHdr := header.ICMPv6(packet)
want := header.ICMPv6Checksum(header.ICMPv6ChecksumParams{
Header: icmpHdr[:header.ICMPv6MinimumSize],
Src: src,
Dst: dst,
PayloadLen: len(icmpHdr.Payload()),
PayloadCsum: checksumPayloadV6(icmpHdr.Payload()),
})
if got := icmpHdr.Checksum(); got != want {
t.Fatalf("unexpected ipv6 checksum: got %x want %x", got, want)
}
})
}
func TestBuildLocalEchoReply(t *testing.T) {
t.Run("ipv4", func(t *testing.T) {
request := []byte{
byte(header.ICMPv4Echo), 0,
0, 0,
0x12, 0x34,
0x56, 0x78,
0xaa, 0xbb, 0xcc,
}
src := tcpip.Address{}
dst := tcpip.Address{}
if err := RewriteChecksum(header.IPv4ProtocolNumber, request, src, dst); err != nil {
t.Fatal(err)
}
reply, err := BuildLocalEchoReply(header.IPv4ProtocolNumber, request, dst, src)
if err != nil {
t.Fatal(err)
}
if request[0] != byte(header.ICMPv4Echo) {
t.Fatal("request mutated")
}
icmpHdr := header.ICMPv4(reply)
if icmpHdr.Type() != header.ICMPv4EchoReply || icmpHdr.Code() != header.ICMPv4UnusedCode {
t.Fatalf("unexpected ipv4 reply type/code: %d/%d", icmpHdr.Type(), icmpHdr.Code())
}
if icmpHdr.Ident() != 0x1234 || icmpHdr.Sequence() != 0x5678 {
t.Fatalf("unexpected ipv4 ident/sequence: %x/%x", icmpHdr.Ident(), icmpHdr.Sequence())
}
})
t.Run("ipv6", func(t *testing.T) {
request := []byte{
byte(header.ICMPv6EchoRequest), 0,
0, 0,
0xab, 0xcd,
0xef, 0x01,
0xaa, 0xbb, 0xcc,
}
src := tcpip.AddrFromSlice([]byte{0x20, 0x01, 0x0d, 0xb8, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 1})
dst := tcpip.AddrFromSlice([]byte{0x20, 0x01, 0x0d, 0xb8, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 2})
if err := RewriteChecksum(header.IPv6ProtocolNumber, request, src, dst); err != nil {
t.Fatal(err)
}
reply, err := BuildLocalEchoReply(header.IPv6ProtocolNumber, request, dst, src)
if err != nil {
t.Fatal(err)
}
if request[0] != byte(header.ICMPv6EchoRequest) {
t.Fatal("request mutated")
}
icmpHdr := header.ICMPv6(reply)
if icmpHdr.Type() != header.ICMPv6EchoReply || icmpHdr.Code() != header.ICMPv6UnusedCode {
t.Fatalf("unexpected ipv6 reply type/code: %d/%d", icmpHdr.Type(), icmpHdr.Code())
}
if icmpHdr.Ident() != 0xabcd || icmpHdr.Sequence() != 0xef01 {
t.Fatalf("unexpected ipv6 ident/sequence: %x/%x", icmpHdr.Ident(), icmpHdr.Sequence())
}
})
}
func checksumPayloadV4(payload []byte) uint16 {
return checksum.Checksum(payload, 0)
}
func checksumPayloadV6(payload []byte) uint16 {
return checksum.Checksum(payload, 0)
}
proxy/tun/packet.go
+231 -2
View File
@@ -2,8 +2,14 @@ package tun
import (
"encoding/binary"
"math"
"math/bits"
"net/netip"
"slices"
"sync"
"time"
"github.com/xtls/xray-core/common/errors"
)
const (
@@ -16,6 +22,19 @@ const (
protoTCP = 6
protoUDP = 17
protoICMPv6 = 58
// the smallest that a link is believed to take whatever is said about it: what every IPv4 host takes,
// and the smallest link that IPv6 runs on
ipv4SmallestMTU = 576
ipv6SmallestMTU = 1280
// the header that an IPv6 packet has in front of each of its fragments, and its number
fragmentHeaderSize = 8
protoFragment = 44
// fragmentsMost is how many packets can be waiting for the rest of their fragments,
// fragmentsTime how long they are waited for at most
fragmentsMost = 64
fragmentsTime = 30 * time.Second
)
// packet is an IP packet with its header read
@@ -24,9 +43,15 @@ type packet struct {
src, dst netip.Addr
// payload is what follows the IP header
payload []byte
// fragment is set if payload is a part of what follows the IP header: the one at offset of the packet
// with the number id, which has more after it or not
fragment bool
more bool
offset int
id uint32
}
// parsePacket reads the IP header of b. Fragments are not accepted: what the system sends to the tun fits its MTU
// parsePacket reads the IP header of b
func parsePacket(b []byte) (p packet, ok bool) {
if len(b) < ipv4HeaderSize {
return p, false
@@ -35,9 +60,12 @@ func parsePacket(b []byte) (p packet, ok bool) {
case 4:
size := int(b[0]&0xf) * 4
total := int(binary.BigEndian.Uint16(b[2:]))
if size < ipv4HeaderSize || total < size || total > len(b) || binary.BigEndian.Uint16(b[6:])&0x3fff != 0 {
if size < ipv4HeaderSize || total < size || total > len(b) {
return p, false
}
if f := binary.BigEndian.Uint16(b[6:]); f&0x3fff != 0 {
p.fragment, p.more, p.offset, p.id = true, f&0x2000 != 0, int(f&0x1fff)*8, uint32(binary.BigEndian.Uint16(b[4:]))
}
p.proto = b[9]
p.src = netip.AddrFrom4([4]byte(b[12:16]))
p.dst = netip.AddrFrom4([4]byte(b[16:20]))
@@ -65,12 +93,55 @@ func parsePacket(b []byte) (p packet, ok bool) {
}
p.proto, p.payload = p.payload[0], p.payload[size:]
}
if p.proto == protoFragment {
f := p.payload
if len(f) < fragmentHeaderSize {
return p, false
}
p.fragment, p.more, p.offset, p.id = true, f[3]&1 != 0, int(binary.BigEndian.Uint16(f[2:])&^7), binary.BigEndian.Uint32(f[4:])
p.proto, p.payload = f[0], f[fragmentHeaderSize:]
}
default:
return p, false
}
return p, true
}
// verified reports whether the checksum of what the packet carries is right
func (p packet) verified() bool {
data := p.payload
switch p.proto {
case protoICMPv4:
return checksum(sum(data, 0)) == 0
case protoUDP:
if len(data) < udpHeaderSize {
return false
}
// the packet may be longer than what it carries
size := int(binary.BigEndian.Uint16(data[4:]))
if size < udpHeaderSize || size > len(data) {
return false
}
if p.src.Is4() && data[6]|data[7] == 0 {
// zero stands for no checksum
return true
}
data = data[:size]
case protoTCP, protoICMPv6:
default:
return true
}
pseudo := uint64(p.proto) + uint64(len(data))
if p.src.Is4() {
src, dst := p.src.As4(), p.dst.As4()
pseudo = sum(dst[:], sum(src[:], pseudo))
} else {
src, dst := p.src.As16(), p.dst.As16()
pseudo = sum(dst[:], sum(src[:], pseudo))
}
return checksum(sum(data, pseudo)) == 0
}
// putIPHeader writes the header of a packet from src to dst into b, without its length, and returns the size of the header
func putIPHeader(b []byte, proto byte, src, dst netip.Addr) int {
if src.Is4() {
@@ -159,3 +230,161 @@ func checksum(s uint64) uint16 {
s = s>>16 + s&0xffff
return ^uint16(s)
}
// writeFragments writes a packet that is larger than mtu to tun in fragments that are not. header is its IP header
// with what follows that up to payload, which is no more than the header of a UDP packet. id is the number
// of the packet if it is an IPv6 one
func writeFragments(tun Tun, mtu int, id uint32, header, payload []byte) error {
var b [ipv6HeaderSize + fragmentHeaderSize + udpHeaderSize]byte
size := ipv4HeaderSize
if header[0]>>4 == 6 {
size = ipv6HeaderSize
}
n := copy(b[:], header[:size])
if size == ipv6HeaderSize {
// the number goes into a header of its own, behind the IP header. That of IPv4 has it
b[6], b[n] = protoFragment, header[6]
binary.BigEndian.PutUint32(b[n+4:], id)
n += fragmentHeaderSize
}
// first goes in front of payload, in the first fragment
first := header[size:]
if n+len(first) > len(b) {
return errors.New("no room for a header of ", len(first), " bytes in a fragment")
}
// every fragment but the last is a number of 8 bytes
room, total := (mtu-n)&^7, len(first)+len(payload)
if room < 8 {
return errors.New("no room for a fragment in an MTU of ", mtu)
}
for offset := 0; offset < total; {
h, part := b[:n], min(room, total-offset)
last := offset+part == total
if size == ipv6HeaderSize {
f := uint16(offset)
if !last {
f |= 1
}
binary.BigEndian.PutUint16(b[size+2:], f)
} else {
f := uint16(offset / 8)
if !last {
f |= 0x2000
}
binary.BigEndian.PutUint16(b[6:], f)
}
setIPLength(b[:size], n+part)
data := payload[max(offset-len(first), 0):min(offset+part-len(first), len(payload))]
if offset == 0 {
h = b[:n+copy(b[n:], first)]
}
if err := tun.WritePacket(h, data); err != nil {
return err
}
offset += part
}
return nil
}
// fragmented is a packet of which not every fragment has come yet
type fragmented struct {
// data is what follows its IP header, as far as its last fragment that has come goes
data []byte
// parts are what of data has come, in order and apart from each other
parts [][2]int
// size is that of data when all of it has come, known from the last fragment
size int
known bool
since time.Time
}
// reassembly puts the packets that come in fragments together
type reassembly struct {
mu sync.Mutex
packets map[fragments]*fragmented
}
// fragments is what the fragments of a packet have in common
type fragments struct {
src, dst netip.Addr
proto byte
id uint32
}
// add takes the fragment p of a packet. With the last one that was missing, it returns the packet.
// A packet whose fragments do not fit each other is given up
func (r *reassembly) add(p packet) (packet, bool) {
fragment, offset, end := p.payload, p.offset, p.offset+len(p.payload)
// what an IP packet can carry
most := math.MaxUint16
if p.src.Is4() {
most -= ipv4HeaderSize
}
if len(fragment) == 0 || end > most || p.more && len(fragment)%8 != 0 {
return p, false
}
key, now := fragments{p.src, p.dst, p.proto, p.id}, time.Now()
r.mu.Lock()
defer r.mu.Unlock()
f := r.packets[key]
if f != nil && now.Sub(f.since) > fragmentsTime {
// what is left of a packet whose other fragments were lost: this is one of another packet with its number
f = nil
}
if f == nil {
if _, stale := r.packets[key]; !stale && len(r.packets) >= fragmentsMost {
// the one that is waited for the longest makes room, not the one that comes now
var longest fragments
var since time.Time
for other, old := range r.packets {
if since.IsZero() || old.since.Before(since) {
longest, since = other, old.since
}
}
delete(r.packets, longest)
}
if r.packets == nil {
r.packets = make(map[fragments]*fragmented)
}
f = &fragmented{since: now}
r.packets[key] = f
}
for _, part := range f.parts {
if offset >= part[0] && end <= part[1] {
// came already
return p, false
}
if offset < part[1] && part[0] < end {
// fragments of one packet do not overlap: these are of two, or made to be taken for something else
delete(r.packets, key)
return p, false
}
}
if f.known && end > f.size || !p.more && len(f.data) > end {
// past its end, or an end before what came
delete(r.packets, key)
return p, false
}
if !p.more {
f.size, f.known = end, true
}
if len(f.data) < end {
f.data = append(f.data, make([]byte, end-len(f.data))...)
}
copy(f.data[offset:], fragment)
i := 0
for i < len(f.parts) && f.parts[i][1] < offset {
i++
}
j := i
for ; j < len(f.parts) && f.parts[j][0] <= end; j++ {
offset, end = min(offset, f.parts[j][0]), max(end, f.parts[j][1])
}
f.parts = slices.Replace(f.parts, i, j, [2]int{offset, end})
if !f.known || f.parts[0] != [2]int{0, f.size} {
return p, false
}
delete(r.packets, key)
p.payload, p.fragment = f.data[:f.size], false
return p, true
}
proxy/tun/stack.go
+306 -42
View File
@@ -3,7 +3,12 @@ package tun
import (
"context"
"encoding/binary"
"io"
"math"
"math/rand/v2"
gonet "net"
"net/netip"
"slices"
"sync"
"sync/atomic"
"time"
@@ -15,8 +20,16 @@ import (
// Stack interface implement ip protocol stack, bridging raw network packets and data streams
type Stack interface {
// Start makes the stack read the packets of its tun
Start() error
Close() error
// Input takes a packet of a tun that hands them over instead of being read. The packet is not kept,
// but it may be written to. The packets of one connection have to come one after another
Input(packet []byte)
// DialTCP opens a connection to address from the one of StackOptions.Addresses that is of its family
DialTCP(ctx context.Context, address netip.AddrPort) (net.Conn, error)
// DialUDP opens a port on StackOptions.Addresses. What anyone sends to the port is read from it
DialUDP() (net.PacketConn, error)
}
// StackOptions for the stack implementation
@@ -24,6 +37,12 @@ type StackOptions struct {
Tun Tun
MTU uint32
IdleTimeout time.Duration
// Addresses are what the stack opens connections from
Addresses []netip.Addr
// Remote is set if the tun is a tunnel to another system and not the tun of this one: what comes through it
// has come a long way, so its checksums are looked at, nothing but that way limits the windows, and the
// segments of TCP carry the times of both sides if the other side has them
Remote bool
}
// flow is what the packets of one connection have in common, src is the side of the system
@@ -33,7 +52,8 @@ type flow struct {
// stack turns the packets that the system sends to the tun into connections and back.
// It has only what the two need to talk through a tun, and a local link when the system passes on
// what others send, see tcpConn
// what others send, see tcpConn. The tun may also be a tunnel to a system that is far away, and the stack
// the one that opens the connections
type stack struct {
ctx context.Context
tun Tun
@@ -42,32 +62,58 @@ type stack struct {
segmenter segmenter
// window is the window of every connection
window int32
// handle is given every new connection
// verify is set if the checksums of what comes are looked at
verify bool
// stamps is set if connections carry the times of RFC 7323 when the other side does
stamps bool
// handle is given every new connection that the other side opens. Without it those are refused
handle func(conn net.Conn, destination net.Destination)
udp *udpConnectionHandler
// addresses are what connections are opened from
addresses []netip.Addr
mu sync.Mutex
conns map[flow]*tcpConn
closed bool
// opening is how many of conns wait for the answer to their SYN-ACK
mu sync.Mutex
conns map[flow]*tcpConn
// sockets are the ports that DialUDP has opened
sockets map[netip.AddrPort]*udpSocket
closed bool
// opening is how many of conns wait for the answer to their SYN or SYN-ACK
opening atomic.Int32
// fragments are the packets that are larger than a link on their way and come in parts.
// identification numbers the ones that leave like that: a counter for each of a few groups of destinations,
// each from a start of its own, so that no two of them that go one way meet, and no destination learns how much
// goes to the others
fragments reassembly
identification [64]atomic.Uint32
}
// NewStack builds new ip stack
func NewStack(ctx context.Context, options StackOptions, handler *Handler) (Stack, error) {
// NewStack builds new ip stack. Without a handler it takes no connections, only opens them
func NewStack(ctx context.Context, options StackOptions, handler ConnectionHandler) (Stack, error) {
s := &stack{
ctx: ctx,
tun: options.Tun,
mtu: int(options.MTU),
window: tcpWindow,
handle: handler.HandleConnection,
conns: make(map[flow]*tcpConn),
ctx: ctx,
tun: options.Tun,
mtu: int(options.MTU),
window: tcpWindow,
addresses: options.Addresses,
verify: options.Remote,
stamps: options.Remote,
conns: make(map[flow]*tcpConn),
sockets: make(map[netip.AddrPort]*udpSocket),
}
if t, ok := options.Tun.(segmenter); ok && t.Segments() {
s.segmenter = t
s.window = tcpLargeWindow
}
s.udp = newUdpConnectionHandler(s.handle, s.writeUDP)
if options.Remote {
s.window = tcpLargeWindow
}
for i := range s.identification {
s.identification[i].Store(rand.Uint32())
}
if handler != nil {
s.handle = handler.HandleConnection
s.udp = newUdpConnectionHandler(s.handle, s.writeUDP)
}
return s, nil
}
@@ -85,6 +131,8 @@ func (s *stack) Close() error {
for _, c := range s.conns {
conns = append(conns, c)
}
sockets := s.sockets
s.sockets = nil
s.mu.Unlock()
for _, c := range conns {
c.mu.Lock()
@@ -93,6 +141,9 @@ func (s *stack) Close() error {
}
c.mu.Unlock()
}
for _, u := range sockets {
u.queue.close()
}
return nil
}
@@ -111,19 +162,59 @@ func (s *stack) read() {
}
return
}
p, ok := parsePacket(b)
if !ok {
continue
}
switch p.proto {
case protoTCP:
s.handleTCP(p)
case protoUDP:
s.handleUDP(p)
case protoICMPv4, protoICMPv6:
s.handleICMP(p)
s.Input(b)
}
}
// Input implements Stack. Several tunnels may hand over their packets at once
func (s *stack) Input(b []byte) {
p, ok := parsePacket(b)
if !ok || p.src.Is4In6() || p.dst.Is4In6() || p.src.IsLoopback() || p.dst.IsLoopback() || p.dst.IsUnspecified() || p.src.IsMulticast() {
// nothing comes from many at once, and an address that stands for this machine itself is not one
// that the other side of a tun can ask for. Nor is an IPv6 address that stands for an IPv4 one, which
// would be dialed as that: ::ffff:0.0.0.0 is this machine as well
return
}
if s.verify && b[0]>>4 == 4 && checksum(sum(b[:int(b[0]&0xf)*4], 0)) != 0 {
return
}
// what was put together may be of two packets with one number, its checksum tells
verify := s.verify || p.fragment
if p.fragment {
p, ok = s.fragments.add(p)
}
if !ok || verify && !p.verified() {
return
}
switch p.proto {
case protoTCP:
s.handleTCP(p)
case protoUDP:
s.handleUDP(p)
case protoICMPv4, protoICMPv6:
s.handleICMP(p)
}
}
// local returns the address that packets to remote are sent from
func (s *stack) local(remote netip.Addr) (netip.Addr, error) {
for _, a := range s.addresses {
if a.Is4() == remote.Is4() {
return a, nil
}
}
return netip.Addr{}, errors.New("no address to reach ", remote, " from")
}
// ephemeral returns a port for what is opened from here: the first that free takes, from a random one on
func ephemeral(free func(port uint16) bool) (uint16, bool) {
start := rand.Uint32()
for i := range uint32(1 << 15) {
if port := uint16(1<<15 | (start+i)&(1<<15-1)); free(port) {
return port, true
}
}
return 0, false
}
func (s *stack) handleTCP(p packet) {
@@ -154,7 +245,7 @@ func (s *stack) handleTCP(p packet) {
if c != nil && c.input(seg) {
return
}
if seg.flags&(tcpSyn|tcpAck|tcpRst) != tcpSyn {
if seg.flags&(tcpSyn|tcpAck|tcpRst) != tcpSyn || s.handle == nil {
if seg.flags&tcpRst == 0 {
s.refuse(f, seg)
}
@@ -164,7 +255,8 @@ func (s *stack) handleTCP(p packet) {
if s.opening.Load() >= tcpOpening {
return
}
c = newTCPConn(s, f, seg)
c = newTCPConn(s, f)
c.synchronize(seg)
s.mu.Lock()
if s.closed {
s.mu.Unlock()
@@ -174,7 +266,7 @@ func (s *stack) handleTCP(p packet) {
s.mu.Unlock()
s.opening.Add(1)
c.mu.Lock()
c.sendSynAck()
c.sendSyn()
c.arm(tcpTick)
c.mu.Unlock()
}
@@ -224,51 +316,189 @@ func (s *stack) handleUDP(p packet) {
if size < udpHeaderSize || size > len(u) {
return
}
var socket *udpSocket
if s.handle == nil {
// only for a port that was opened from here
s.mu.Lock()
socket = s.sockets[netip.AddrPortFrom(p.dst, binary.BigEndian.Uint16(u[2:]))]
s.mu.Unlock()
if socket == nil {
return
}
}
// the packet waits for its reader, longer than read keeps it
b := buf.NewWithSize(int32(size - udpHeaderSize))
b.Write(u[udpHeaderSize:size])
src := net.UDPDestination(net.IPAddress(p.src.AsSlice()), net.Port(binary.BigEndian.Uint16(u)))
dst := net.UDPDestination(net.IPAddress(p.dst.AsSlice()), net.Port(binary.BigEndian.Uint16(u[2:])))
s.udp.HandlePacket(src, dst, b)
if socket == nil {
s.udp.HandlePacket(src, net.UDPDestination(net.IPAddress(p.dst.AsSlice()), net.Port(binary.BigEndian.Uint16(u[2:]))), b)
} else if b.UDP = &src; !socket.queue.push(b) {
b.Release()
}
}
// writeUDP writes a packet with the payload from src to dst to the tun
func (s *stack) writeUDP(payload []byte, src net.Destination, dst net.Destination) error {
srcIP, _ := netip.AddrFromSlice(src.Address.IP())
dstIP, _ := netip.AddrFromSlice(dst.Address.IP())
srcIP, dstIP = srcIP.Unmap(), dstIP.Unmap()
if !srcIP.IsValid() || srcIP.Is4() != dstIP.Is4() {
return errors.New("failed to write udp packet from ", src, " to ", dst)
return s.sendUDP(payload, netip.AddrPortFrom(srcIP, uint16(src.Port)), netip.AddrPortFrom(dstIP, uint16(dst.Port)))
}
func (s *stack) sendUDP(payload []byte, src, dst netip.AddrPort) error {
srcIP, dstIP := src.Addr().Unmap(), dst.Addr().Unmap()
if !srcIP.IsValid() || srcIP.Is4() != dstIP.Is4() || len(payload) > math.MaxUint16-udpHeaderSize {
return errors.New("failed to write udp packet of ", len(payload), " bytes from ", src, " to ", dst)
}
var b [ipv6HeaderSize + udpHeaderSize]byte
n := putIPHeader(b[:], protoUDP, srcIP, dstIP)
u := b[n : n+udpHeaderSize]
binary.BigEndian.PutUint16(u, uint16(src.Port))
binary.BigEndian.PutUint16(u[2:], uint16(dst.Port))
binary.BigEndian.PutUint16(u, src.Port())
binary.BigEndian.PutUint16(u[2:], dst.Port())
binary.BigEndian.PutUint16(u[4:], uint16(udpHeaderSize+len(payload)))
setIPLength(b[:n], n+udpHeaderSize+len(payload))
check := transportChecksum(b[:n], protoUDP, u, payload)
if check == 0 {
// zero stands for no checksum
check = 0xffff
}
binary.BigEndian.PutUint16(u[6:], check)
return s.tun.WritePacket(b[:n+udpHeaderSize], payload)
return s.write(b[:n+udpHeaderSize], payload)
}
// handleICMP answers an echo request as its destination would
// write completes the IP header of a packet that is in two parts and writes it to the tun, in fragments if it is
// larger than the MTU. A segment of a connection never is, and is not written by this
func (s *stack) write(header, payload []byte) error {
size, last := ipv6HeaderSize, 39
if header[0]>>4 == 4 {
size, last = ipv4HeaderSize, 19
}
// the group is chosen by the last bytes of the destination
id := s.identification[(header[last]^header[last-1]<<3)%byte(len(s.identification))].Add(1)
if header[0]>>4 == 4 {
// a link on the way may split it as well, and what is put together there is told apart by its number
binary.BigEndian.PutUint16(header[4:], uint16(id))
}
if len(header)+len(payload) > s.mtu {
return writeFragments(s.tun, s.mtu, id, header, payload)
}
setIPLength(header[:size], len(header)+len(payload))
return s.tun.WritePacket(header, payload)
}
// DialUDP implements Stack
func (s *stack) DialUDP() (net.PacketConn, error) {
u := &udpSocket{stack: s, queue: newPacketQueue(udpQueueLimit)}
s.mu.Lock()
defer s.mu.Unlock()
if s.closed {
return nil, io.ErrClosedPipe
}
port, ok := ephemeral(func(port uint16) bool {
for _, a := range s.addresses {
if s.sockets[netip.AddrPortFrom(a, port)] != nil {
return false
}
}
return true
})
if !ok || len(s.addresses) == 0 {
return nil, errors.New("no port left on ", s.addresses)
}
u.port = port
for _, a := range s.addresses {
s.sockets[netip.AddrPortFrom(a, port)] = u
}
return u, nil
}
// udpSocket is a UDP port that was opened from here, on every address of the stack. It is not bound to one
// other side: what anyone sends to it is read from it, with who sent it
type udpSocket struct {
stack *stack
port uint16
queue *packetQueue
}
func (u *udpSocket) ReadFrom(p []byte) (int, net.Addr, error) {
b, ok := u.queue.pop()
if !ok {
return 0, nil, gonet.ErrClosed
}
defer b.Release()
return copy(p, b.Bytes()), b.UDP.RawNetAddr(), nil
}
func (u *udpSocket) WriteTo(p []byte, addr net.Addr) (int, error) {
to, ok := addr.(*net.UDPAddr)
if !ok {
return 0, errors.New("not a udp address: ", addr)
}
local, err := u.stack.local(to.AddrPort().Addr().Unmap())
if err != nil {
return 0, err
}
if err := u.stack.sendUDP(p, netip.AddrPortFrom(local, u.port), to.AddrPort()); err != nil {
return 0, err
}
return len(p), nil
}
func (u *udpSocket) Close() error {
u.stack.mu.Lock()
for _, a := range u.stack.addresses {
if local := netip.AddrPortFrom(a, u.port); u.stack.sockets[local] == u {
delete(u.stack.sockets, local)
}
}
u.stack.mu.Unlock()
u.queue.close()
return nil
}
// LocalAddr implements net.PacketConn with the first address of the stack
func (u *udpSocket) LocalAddr() net.Addr {
return gonet.UDPAddrFromAddrPort(netip.AddrPortFrom(u.stack.addresses[0], u.port))
}
func (u *udpSocket) SetDeadline(time.Time) error {
return nil
}
func (u *udpSocket) SetReadDeadline(time.Time) error {
return nil
}
func (u *udpSocket) SetWriteDeadline(time.Time) error {
return nil
}
// handleICMP answers an echo request as its destination would, and takes the word that a packet of a connection
// was too large for a link on its way or found no way at all
func (s *stack) handleICMP(p packet) {
m := p.payload
request, reply := byte(8), byte(0)
if p.proto == protoICMPv6 {
request, reply = 128, 129
}
if len(m) < 8 || m[0] != request || m[1] != 0 {
if len(m) < 8 {
return
}
switch {
case p.proto == protoICMPv4 && m[0] == 3 && m[1] == 4:
s.undelivered(m[8:], max(int(binary.BigEndian.Uint16(m[6:])), ipv4SmallestMTU))
return
case p.proto == protoICMPv6 && m[0] == 2:
s.undelivered(m[8:], max(int(binary.BigEndian.Uint32(m[4:])), ipv6SmallestMTU))
return
case p.proto == protoICMPv4 && m[0] == 3, p.proto == protoICMPv6 && m[0] == 1:
s.undelivered(m[8:], 0)
return
}
if m[0] != request || m[1] != 0 || s.handle == nil && !slices.Contains(s.addresses, p.dst) {
// a stack that takes no connections is one host, not all of them
return
}
var b [ipv6HeaderSize]byte
n := putIPHeader(b[:], p.proto, p.dst, p.src)
setIPLength(b[:n], n+len(m))
// the reply is the request with another type
m[0], m[2], m[3] = reply, 0, 0
if p.proto == protoICMPv6 {
@@ -276,5 +506,39 @@ func (s *stack) handleICMP(p packet) {
} else {
binary.BigEndian.PutUint16(m[2:], checksum(sum(m, 0)))
}
_ = s.tun.WritePacket(b[:n], m)
_ = s.write(b[:n], m)
}
// undelivered takes the start of a packet that was sent from here and did not get to where it goes. If it did not fit
// a link on its way, which takes packets of mtu bytes, the connection that sent it goes on with smaller segments.
// Without an mtu there is no way at all, and a connection that is being opened is not waited for
func (s *stack) undelivered(b []byte, mtu int) {
var proto byte
var local, remote netip.Addr
switch {
case len(b) >= ipv4HeaderSize && b[0]>>4 == 4:
proto, local, remote = b[9], netip.AddrFrom4([4]byte(b[12:16])), netip.AddrFrom4([4]byte(b[16:20]))
b = b[min(int(b[0]&0xf)*4, len(b)):]
case len(b) >= ipv6HeaderSize && b[0]>>4 == 6:
proto, local, remote = b[6], netip.AddrFrom16([16]byte(b[8:24])), netip.AddrFrom16([16]byte(b[24:40]))
b = b[ipv6HeaderSize:]
}
// of the segment there are the ports and the sequence number at least
if proto != protoTCP || len(b) < 8 {
return
}
s.mu.Lock()
c := s.conns[flow{
src: netip.AddrPortFrom(remote, binary.BigEndian.Uint16(b[2:])),
dst: netip.AddrPortFrom(local, binary.BigEndian.Uint16(b)),
}]
s.mu.Unlock()
if c == nil {
return
}
if seq := binary.BigEndian.Uint32(b[4:]); mtu != 0 {
c.shrink(seq, mtu)
} else {
c.unreachable(seq)
}
}
proxy/tun/stack_test.go
+1453 -129
View File
File diff suppressed because it is too large Load Diff
proxy/tun/tcp_lightweight.go
+287 -41
View File
@@ -1,10 +1,12 @@
package tun
import (
"context"
"encoding/binary"
"io"
"math"
"math/rand/v2"
"net/netip"
"slices"
"sync"
"time"
@@ -37,7 +39,9 @@ const (
tcpHeld = 1024
// tcpSmallest is the smallest mss that the system is believed, the one of Linux
tcpSmallest = 48
// tcpOpening is how many connections can be waiting for the answer to their SYN-ACK
// tcpStamps is what the times of both sides take of the options of every segment that carries them (RFC 7323)
tcpStamps = 12
// tcpOpening is how many connections that the other side opens can be waiting for the answer to their SYN-ACK
tcpOpening = 65535
// tcpStart is how many segments are sent before any of them is acknowledged
tcpStart = 10
@@ -48,12 +52,18 @@ const (
// tcpTick is how often a connection that waits for something looks at what it waits for
tcpTick = 100 * time.Millisecond
// ticks without anything acknowledged before it is sent again, twice as many every next time, a minute at most
// ticks without anything acknowledged before it is sent again, twice as many every next time, a minute at most,
// while an open connection that waits sends again every two seconds at most or as often as an answer takes,
// twice as seldom every next minute: when its way was away for a while it goes on soon after the way is back,
// and the next segment still waits the doubled time, so that its answer can be timed
tcpRetransmitTicks = 3
tcpRetransmitTicksOpen = 20
tcpRetransmitTicksMost = 600
// how many times it is sent again before the connection is given up: that is after four minutes, as what is
// behind a wireless link may be away for a while, and after twenty seconds for one that never answered
tcpRetransmits = 10
tcpMinuteTicks = int(time.Minute / tcpTick)
// an open connection is given up after four minutes of ticks without an acknowledgment since the timer first sent
// it again, as what is behind a wireless link may be away for a while (ticks, so that the time a computer sleeps
// is not counted), and one that never answered after it was sent again that many times, after twenty seconds
tcpGiveUpTicks = 4 * tcpMinuteTicks
tcpSynRetransmits = 5
// ticks between the probes of a window that stays closed
tcpProbeTicks = 10
@@ -62,10 +72,15 @@ const (
)
var (
errTCPReset = errors.New("connection reset")
errTCPTimeout = errors.New("connection timed out")
errTCPReset = errors.New("connection reset")
errTCPTimeout = errors.New("connection timed out")
errTCPRefused = errors.New("connection refused")
errTCPUnreachable = errors.New("destination unreachable")
)
// tcpStarted is where the times that segments carry count from
var tcpStarted = time.Now()
// segment is a TCP segment that came from the tun
type segment struct {
seq, ack uint32
@@ -92,7 +107,7 @@ type held struct {
data *buf.Buffer
}
// tcpConn is a TCP connection that the system opened through the tun.
// tcpConn is a TCP connection that the system opened through the tun, or one that was opened to it by DialTCP.
// Mostly the two are on the same machine, with nothing but the queue of the tun in between, which drops what
// does not fit. But the system may also pass on what others send, as a router does, and then the link to them
// is in between. So a connection has windows that are small enough, keeps what came after a lost segment,
@@ -114,12 +129,19 @@ type tcpConn struct {
sndShift, rcvShift uint8
// sack is set if the system takes selective acknowledgments
sack bool
// stamps is set if every segment carries the time of this side and the last one that came of the other side,
// which sees by it how long its segments take, also the ones that it sent again (RFC 7323).
// stampBase is what the time of this side starts from, stampSent the one that the last segment carried
stamps bool
stampBase, stampSent uint32
mu sync.Mutex
readable sync.Cond
writable sync.Cond
established bool
// opened is there while DialTCP waits for the answer to its SYN, and closed when that wait is over
opened chan struct{}
// closed is set when Close is called
closed bool
// err is why the connection is gone
@@ -140,6 +162,8 @@ type tcpConn struct {
rcvHeld []held
// rcvLast is the seq of the last one that was put into rcvHeld
rcvLast uint32
// stamp is the time of the other side that is sent back to it
stamp uint32
sndUna, sndNxt uint32
sndWnd uint32
@@ -165,6 +189,9 @@ type tcpConn struct {
recover uint32
// lost is the end of what is lost past the last of sacked, resent the end of what was sent again since then
lost, resent uint32
// headAt is when the segment at sndUna was sent again, or took that place while something was lost.
// It is zero when that segment is not sent once more for the system still asking for it: it was already
headAt time.Time
// timed is the end of a segment that was sent at timedAt and not again, to see how long its acknowledgment takes.
// The first one is the SYN-ACK
timed uint32
@@ -176,17 +203,17 @@ type tcpConn struct {
armed bool
// stalled counts ticks without an acknowledgment of what waits for one
stalled, retransmitTicks, retransmits int
lingered bool
// lostTicks counts the ticks since the timer sent again the first time since something was acknowledged
lostTicks int
lingered bool
}
func newTCPConn(s *stack, f flow, syn segment) *tcpConn {
func newTCPConn(s *stack, f flow) *tcpConn {
c := &tcpConn{
stack: s,
flow: f,
src: net.TCPDestination(net.IPAddress(f.src.Addr().AsSlice()), net.Port(f.src.Port())),
dst: net.TCPDestination(net.IPAddress(f.dst.Addr().AsSlice()), net.Port(f.dst.Port())),
rcvNxt: syn.seq + 1,
sndWnd: uint32(syn.window),
window: s.window,
ssthresh: uint32(s.window),
timedAt: time.Now(),
@@ -196,6 +223,17 @@ func newTCPConn(s *stack, f flow, syn segment) *tcpConn {
c.writable.L = &c.mu
c.ipSize = putIPHeader(c.ip[:], protoTCP, f.dst.Addr(), f.src.Addr())
c.mss = s.mtu - c.ipSize - tcpHeaderSize
c.sndUna = rand.Uint32()
c.sndNxt = c.sndUna + 1
c.stampBase = rand.Uint32()
return c
}
// synchronize takes what the SYN of the other side says. c.mu is held, or nobody else has c yet
func (c *tcpConn) synchronize(syn segment) {
c.rcvNxt = syn.seq + 1
c.sndWnd = uint32(syn.window)
c.sack, c.stamps, c.sndShift, c.rcvShift = false, false, 0, 0
for kind, value, o := nextOption(syn.options); kind != 0; kind, value, o = nextOption(o) {
switch {
case kind == 2 && len(value) == 2:
@@ -205,20 +243,77 @@ func newTCPConn(s *stack, f flow, syn segment) *tcpConn {
c.rcvShift = tcpWindowShift
case kind == 4:
c.sack = true
case kind == 8 && len(value) == 8 && c.stack.stamps:
c.stamps, c.stamp = true, binary.BigEndian.Uint32(value)
}
}
if c.stamps {
c.mss -= tcpStamps
}
c.mss = max(c.mss, tcpSmallest)
c.cwnd = tcpStart * uint32(c.mss)
c.sndUna = rand.Uint32()
c.sndNxt = c.sndUna + 1
return c
}
// DialTCP implements Stack
func (s *stack) DialTCP(ctx context.Context, remote netip.AddrPort) (net.Conn, error) {
remote = netip.AddrPortFrom(remote.Addr().Unmap(), remote.Port())
local, err := s.local(remote.Addr())
if err != nil {
return nil, err
}
opened := make(chan struct{})
s.mu.Lock()
if s.closed {
s.mu.Unlock()
return nil, io.ErrClosedPipe
}
f := flow{src: remote}
port, ok := ephemeral(func(port uint16) bool {
f.dst = netip.AddrPortFrom(local, port)
return s.conns[f] == nil
})
if !ok {
s.mu.Unlock()
return nil, errors.New("no port left to reach ", remote, " from")
}
f.dst = netip.AddrPortFrom(local, port)
c := newTCPConn(s, f)
// what the SYN offers, until the answer says what of it is taken
c.sack, c.stamps, c.rcvShift, c.opened = true, s.stamps, tcpWindowShift, opened
s.conns[f] = c
s.mu.Unlock()
s.opening.Add(1)
c.mu.Lock()
if c.err == nil {
c.sendSyn()
c.arm(tcpTick)
}
c.mu.Unlock()
select {
case <-opened:
case <-ctx.Done():
}
c.mu.Lock()
defer c.mu.Unlock()
if !c.established {
if c.err == nil {
c.abort(ctx.Err())
}
return nil, c.err
}
return c, nil
}
// header writes the headers of a segment with size bytes of payload into b and returns their size.
// The checksum of the segment is left to write. c.mu is held
func (c *tcpConn) header(b []byte, seq uint32, flags byte, options []byte, size int) int {
n := copy(b, c.ip[:c.ipSize])
t := b[n : n+tcpHeaderSize+len(options)]
stamps := 0
if c.stamps {
stamps = tcpStamps
}
t := b[n : n+tcpHeaderSize+stamps+len(options)]
binary.BigEndian.PutUint16(t, c.flow.dst.Port())
binary.BigEndian.PutUint16(t[2:], c.flow.src.Port())
binary.BigEndian.PutUint32(t[4:], seq)
@@ -233,7 +328,18 @@ func (c *tcpConn) header(b []byte, seq uint32, flags byte, options []byte, size
binary.BigEndian.PutUint16(t[14:], uint16(min(window>>c.rcvShift, math.MaxUint16)))
}
t[16], t[17], t[18], t[19] = 0, 0, 0, 0
copy(t[tcpHeaderSize:], options)
if c.stamps {
o := t[tcpHeaderSize:]
o[0], o[1], o[2], o[3] = 1, 1, 8, 10
if !c.sending {
// what is sent while WriteMultiBuffer writes a segment has the time of that one: it may get to the tun
// first, and the other side drops a segment with a time before the last one that it took
c.stampSent = c.stampBase + uint32(time.Since(tcpStarted)/time.Millisecond)
}
binary.BigEndian.PutUint32(o[4:], c.stampSent)
binary.BigEndian.PutUint32(o[8:], c.stamp)
}
copy(t[tcpHeaderSize+stamps:], options)
setIPLength(b[:n], n+len(t)+size)
c.rcvAcked = c.rcvNxt
c.rcvAdvertised = window
@@ -254,7 +360,8 @@ func (c *tcpConn) send(seq uint32, flags byte, options, payload []byte) {
c.write(b[:c.header(b[:], seq, flags, options, len(payload))], payload)
}
func (c *tcpConn) sendSynAck() {
// sendSyn sends the answer to the SYN of the other side, or the SYN of a connection that is opened from here
func (c *tcpConn) sendSyn() {
options := [12]byte{2, 4}
binary.BigEndian.PutUint16(options[2:], uint16(c.stack.mtu-c.ipSize-tcpHeaderSize))
n := 4
@@ -264,7 +371,11 @@ func (c *tcpConn) sendSynAck() {
if c.rcvShift != 0 {
n += copy(options[n:], []byte{1, 3, 3, c.rcvShift})
}
c.send(c.sndUna, tcpSyn|tcpAck, options[:n], nil)
flags := byte(tcpSyn | tcpAck)
if c.opened != nil {
flags = tcpSyn
}
c.send(c.sndUna, flags, options[:n], nil)
}
// sendAck acknowledges what was received, and tells what of rcvHeld waits for a lost segment
@@ -278,8 +389,12 @@ func (c *tcpConn) sendAck() {
c.send(c.sndNxt, tcpAck, nil, nil)
return
}
// the block with the segment that came last goes first
var blocks [4][2]uint32
// the block with the segment that came last goes first. Next to the times there is room for one less
var room [4][2]uint32
blocks := room[:]
if c.stamps {
blocks = room[:3]
}
n, last := 1, false
for i := 0; i < len(c.rcvHeld); {
start, end, recent := c.rcvHeld[i].seq, c.rcvHeld[i].seq, false
@@ -306,6 +421,22 @@ func (c *tcpConn) sendAck() {
c.send(c.sndNxt, tcpAck, options[:4+8*len(list)], nil)
}
// stamped keeps the time that a segment of the other side carries, to send it back. It is the one of the segment
// that the stream goes on with, so that the other side sees how long that one took, and not one of what came
// before it. c.mu is held
func (c *tcpConn) stamped(s segment) {
if int32(s.seq-c.rcvAcked) > 0 {
return
}
for kind, value, o := nextOption(s.options); kind != 0; kind, value, o = nextOption(o) {
if kind == 8 && len(value) == 8 {
if stamp := binary.BigEndian.Uint32(value); int32(stamp-c.stamp) > 0 {
c.stamp = stamp
}
}
}
}
// input takes a segment of the connection. It returns false if the connection is gone
// and the segment is the start of another one
func (c *tcpConn) input(s segment) bool {
@@ -314,6 +445,32 @@ func (c *tcpConn) input(s segment) bool {
if c.err != nil {
return s.flags&(tcpSyn|tcpAck|tcpRst) != tcpSyn
}
if c.opened != nil {
// only the answer to the SYN is waited for, or the word that nobody takes it
switch {
case s.flags&tcpAck == 0:
case s.ack != c.sndNxt:
// of a connection that the other side still has from before: it is told that there is none
if s.flags&tcpRst == 0 {
c.stack.refuse(c.flow, s)
}
case s.flags&tcpRst != 0:
c.abort(errTCPRefused)
case s.flags&tcpSyn != 0:
c.synchronize(s)
c.established = true
c.sndUna, c.sndBase = s.ack, s.ack
c.stalled, c.retransmits = 0, 0
if !c.timedAt.IsZero() {
c.clocked()
}
c.stack.opening.Add(-1)
close(c.opened)
c.opened = nil
c.sendAck()
}
return true
}
if s.flags&tcpRst != 0 {
// from where the system is, or was before its FIN, not from someone who knows no more than the addresses
if s.seq-c.rcvNxt <= uint32(c.window) || c.rcvFin && s.seq+1 == c.rcvNxt {
@@ -324,9 +481,12 @@ func (c *tcpConn) input(s segment) bool {
if s.flags&tcpSyn != 0 {
switch {
case !c.established && s.seq+1 == c.rcvNxt:
c.sendSynAck()
c.sendSyn()
c.timedAt = time.Time{}
case c.closed && s.flags&tcpAck == 0:
case c.closed && s.flags&tcpAck == 0 && !c.stack.verify:
// the system of this machine opens the same ports again: it has let the old connection go.
// What comes through a tunnel is answered with what is expected instead, as anyone who knows
// the addresses could send it, and the other side answers that with a reset if it is gone
c.abort(errTCPReset)
return false
default:
@@ -337,6 +497,9 @@ func (c *tcpConn) input(s segment) bool {
if s.flags&tcpAck == 0 {
return true
}
if c.stamps {
c.stamped(s)
}
if !c.established {
if s.ack != c.sndNxt {
return true
@@ -385,6 +548,7 @@ func (c *tcpConn) acknowledge(s segment) {
if !c.timedAt.IsZero() && int32(c.sndUna-c.timed) >= 0 {
c.clocked()
}
recovering := c.recovering
switch {
case c.recovering && int32(c.sndUna-c.recover) >= 0:
c.recovering = false
@@ -393,11 +557,18 @@ func (c *tcpConn) acknowledge(s segment) {
if end := c.sndUna + min(uint32(c.mss), c.recover-c.sndUna); int32(end-c.lost) > 0 {
c.lost = end
}
c.headAt = time.Now()
}
// the way takes more if what it was given has arrived, as long as it was given about as much as it takes
if sent >= c.cwnd/2 {
if c.cwnd < c.ssthresh {
c.cwnd = min(c.cwnd+acked, c.ssthresh)
grow := acked
if recovering {
// after a timeout: one segment that is sent again can acknowledge all that came after it,
// which says nothing about how much the way takes at once
grow = min(grow, 2*uint32(c.mss))
}
c.cwnd = min(c.cwnd+grow, c.ssthresh)
} else if !c.recovering {
// a segment more for every cwnd that arrives
c.cwnd = min(c.cwnd+max(uint32(uint64(c.mss)*uint64(acked)/uint64(c.cwnd)), 1), uint32(c.window))
@@ -417,12 +588,22 @@ func (c *tcpConn) acknowledge(s segment) {
}
}
}
if !c.recovering && c.sndUna != c.sndNxt && (len(c.sacked) > 0 || c.dupAcks >= 3) {
// the system has something past sndUna, or says the same for the third time: the segment at sndUna is lost
if !c.recovering && c.sndUna != c.sndNxt && (len(c.sacked) > 0 || !c.sack && c.dupAcks >= 3) {
// the system has something past sndUna, or says the same for the third time when it cannot say what it has:
// the segment at sndUna is lost. One that can say it repeats itself also for what it got twice
c.congested()
c.lost = c.sndUna + min(uint32(c.mss), c.sndNxt-c.sndUna)
}
if c.recovering {
if acked == 0 && c.srtt != 0 && !c.headAt.IsZero() && time.Since(c.headAt) > max(c.srtt+max(c.srtt/4, 4*c.rttvar), tcpTick/4) {
// the segment that the system waits for was sent again longer ago than an answer takes, and the system
// still answers what comes after it: that one was lost as well. It is sent once more, and only once:
// the timer would find it too, but late, and then nothing is known anymore of how much the way takes
if start, end := c.hole(c.sndUna); start == c.sndUna && start != end {
c.retransmit(start, start+min(end-start, uint32(c.mss)))
c.headAt, c.stalled = time.Time{}, 0
}
}
c.resend()
}
c.sndWnd = window
@@ -451,7 +632,12 @@ func (c *tcpConn) clocked() {
c.rttvar += ((c.srtt - d).Abs() - c.rttvar) / 4
c.srtt += (d - c.srtt) / 8
}
c.retransmitTicks = max(tcpRetransmitTicks, int((c.srtt+4*c.rttvar)/tcpTick)+2)
c.retransmitTicks = c.answerTicks()
}
// answerTicks is how many ticks an acknowledgment is waited for as long as nothing was sent again. c.mu is held
func (c *tcpConn) answerTicks() int {
return max(tcpRetransmitTicks, int((c.srtt+4*c.rttvar)/tcpTick)+2)
}
// sackAdd puts what the system says it has into sacked. c.mu is held
@@ -481,7 +667,7 @@ func (c *tcpConn) congested() {
c.ssthresh = max(min(c.cwnd, c.sndNxt-c.sndUna)/2, 2*uint32(c.mss))
c.cwnd = c.ssthresh
c.recovering, c.recover, c.resent = true, c.sndNxt, c.sndUna
c.timedAt = time.Time{}
c.timedAt, c.headAt = time.Time{}, time.Time{}
}
// hole returns the first part at or after seq of what is lost: what the system does not have
@@ -543,6 +729,10 @@ func (c *tcpConn) resend() {
room = c.cwnd - flight
}
end = start + min(end-start, room)
if start == c.sndUna {
// its answer is waited for from now on
c.headAt, c.stalled = time.Now(), 0
}
c.retransmit(start, end)
c.resent = end
}
@@ -642,9 +832,9 @@ func (c *tcpConn) release() {
// hold keeps a segment that is not the next one. c.mu is held
func (c *tcpConn) hold(seq uint32, payload []byte) {
i := 0
for i < len(c.rcvHeld) && int32(c.rcvHeld[i].seq-seq) < 0 {
i++
i := len(c.rcvHeld)
for i > 0 && int32(c.rcvHeld[i-1].seq-seq) >= 0 {
i--
}
// without what the one before and the one after have
if i > 0 {
@@ -666,8 +856,9 @@ func (c *tcpConn) hold(seq uint32, payload []byte) {
if len(payload) == 0 || len(c.rcvHeld) == tcpHeld {
return
}
b := buf.NewWithSize(int32(len(payload)))
b.Write(payload)
// as much memory as it carries: the room that is left counts bytes, and a thousand tiny segments in pooled
// buffers would take two megabytes
b := buf.FromBytes(slices.Clone(payload))
c.rcvHeld = slices.Insert(c.rcvHeld, i, held{seq, b})
c.rcvLast = seq
}
@@ -716,6 +907,38 @@ func (c *tcpConn) queued(parts [][]byte, seq uint32, size int) ([][]byte, int) {
return parts, n
}
// shrink makes the segments fit a link on the way that takes packets of mtu bytes, when the one at seq did not.
// None of what is on its way fits: it is sent again, the first segment at once and the others as they are
// acknowledged. Only what is said about a segment that was sent and is not acknowledged is believed
func (c *tcpConn) shrink(seq uint32, mtu int) {
c.mu.Lock()
defer c.mu.Unlock()
mss := mtu - c.ipSize - tcpHeaderSize
if c.stamps {
mss -= tcpStamps
}
mss = max(mss, tcpSmallest)
if c.err != nil || !c.established || seq-c.sndUna >= c.sndNxt-c.sndUna || mss >= c.mss {
return
}
c.mss = mss
// as what is lost, without the way taking less for it: it is the size that it did not take
c.recovering, c.recover, c.lost = true, c.sndNxt, c.sndNxt
c.timedAt = time.Time{}
c.resent = c.sndUna + min(uint32(mss), c.sndNxt-c.sndUna)
c.headAt = time.Now()
c.retransmit(c.sndUna, c.resent)
}
// unreachable ends the wait for the answer to the SYN at seq, which a router on the way could not deliver
func (c *tcpConn) unreachable(seq uint32) {
c.mu.Lock()
defer c.mu.Unlock()
if c.err == nil && c.opened != nil && seq == c.sndUna {
c.abort(errTCPUnreachable)
}
}
// arm makes tick run after d, unless it is going to run already. c.mu is held
func (c *tcpConn) arm(d time.Duration) {
if c.armed {
@@ -742,19 +965,34 @@ func (c *tcpConn) tick() {
}
switch {
case c.sndUna != c.sndNxt:
if c.stalled++; c.stalled >= c.retransmitTicks {
if c.retransmits++; c.retransmits > tcpRetransmits || !c.established && c.retransmits > tcpSynRetransmits {
c.stalled++
due := c.stalled >= c.retransmitTicks
if c.established && c.retransmits > 0 {
due = due || c.stalled >= max(tcpRetransmitTicksOpen, c.answerTicks())<<(c.lostTicks/tcpMinuteTicks)
}
if due {
if c.retransmits++; c.retransmits == 1 {
c.lostTicks = 0
} else {
c.lostTicks += c.stalled
}
if c.established && c.lostTicks > tcpGiveUpTicks || !c.established && c.retransmits > tcpSynRetransmits {
c.abort(errTCPTimeout)
return
}
c.stalled = 0
c.retransmitTicks = min(c.retransmitTicks*2, tcpRetransmitTicksMost)
if !c.established {
c.sendSynAck()
c.sendSyn()
c.timedAt = time.Time{}
} else {
// what the system does not say it has is lost, and the way to it is tried with a segment again
// what the system does not say it has is lost, and the way to it is tried with a segment again.
// It takes half of what it took before the loss, not half of that again if the loss was seen already
ssthresh, recovering := c.ssthresh, c.recovering
c.congested()
if recovering {
c.ssthresh = ssthresh
}
c.cwnd, c.lost = uint32(c.mss), c.sndNxt
c.resend()
}
@@ -792,6 +1030,10 @@ func (c *tcpConn) abort(err error) {
if c.timer != nil {
c.timer.Stop()
}
if c.opened != nil {
close(c.opened)
c.opened = nil
}
for _, h := range c.rcvHeld {
h.data.Release()
}
@@ -879,9 +1121,9 @@ func (c *tcpConn) WriteMultiBuffer(mb buf.MultiBuffer) error {
c.writing = len(mb)
// a segment is one part of a Buffer, or as much as there is to send if the tun splits it by itself
var room [tcpParts][]byte
limit, parts := c.mss, room[:0:1]
parts := room[:0:1]
if c.stack.segmenter != nil {
limit, parts = tcpSegment, room[:0]
parts = room[:0]
}
for end := c.sndNxt + size; c.sndNxt != end; {
// not a part of a segment when all of it can go later, with what is acknowledged by then
@@ -897,7 +1139,11 @@ func (c *tcpConn) WriteMultiBuffer(mb buf.MultiBuffer) error {
if c.err != nil {
break
}
most := min(limit, int(c.space()))
// the mss may have become smaller while the last segment was written
most := min(c.mss, int(c.space()))
if c.stack.segmenter != nil {
most = min(tcpSegment, int(c.space()))
}
if most > c.mss {
// the same for what the tun splits
most -= most % c.mss
@@ -907,7 +1153,7 @@ func (c *tcpConn) WriteMultiBuffer(mb buf.MultiBuffer) error {
if c.sndNxt+uint32(n) == end {
flags |= tcpPsh
}
var b [ipv6HeaderSize + tcpHeaderSize]byte
var b [ipv6HeaderSize + tcpHeaderSize + tcpStamps]byte
h := b[:c.header(b[:], c.sndNxt, flags, nil, n)]
if c.sndUna == c.sndNxt {
// the ticks without an acknowledgment are counted from the first segment that waits for one
proxy/tun/udp_fullcone.go
+76 -16
View File
@@ -39,9 +39,7 @@ func (u *udpConnectionHandler) HandlePacket(src net.Destination, dst net.Destina
u.RLock()
conn, found := u.udpConns[src]
if found {
select {
case conn.egress <- data:
default:
if !conn.queue.push(data) {
errors.LogDebug(context.Background(), "drop udp with size ", data.Len(), " to ", dst.NetAddr(), " original ", conn.dst.NetAddr(), " > queue full")
data.Release()
}
@@ -55,17 +53,14 @@ func (u *udpConnectionHandler) HandlePacket(src net.Destination, dst net.Destina
conn, found = u.udpConns[src]
if !found {
egress := make(chan *buf.Buffer, 1024)
conn = &udpConn{handler: u, egress: egress, src: src, dst: dst}
conn = &udpConn{handler: u, queue: newPacketQueue(udpQueueLimit), src: src, dst: dst}
u.udpConns[src] = conn
go u.handleConnection(conn, dst)
}
// send packet data to the egress channel, if it has buffer, or discard
select {
case conn.egress <- data:
default:
// queue the packet for the reader of the connection, or discard it
if !conn.queue.push(data) {
errors.LogDebug(context.Background(), "drop udp with size ", data.Len(), " to ", dst.NetAddr(), " original ", conn.dst.NetAddr(), " > queue full 2")
data.Release()
}
@@ -76,7 +71,7 @@ func (u *udpConnectionHandler) connectionFinished(conn *udpConn) {
// Close runs twice per flow; a newer conn may already own this src.
if u.udpConns[conn.src] == conn {
delete(u.udpConns, conn.src)
close(conn.egress)
conn.queue.close()
}
u.Unlock()
}
@@ -85,13 +80,13 @@ func (u *udpConnectionHandler) connectionFinished(conn *udpConn) {
type udpConn struct {
handler *udpConnectionHandler
egress chan *buf.Buffer
src net.Destination
dst net.Destination
queue *packetQueue
src net.Destination
dst net.Destination
}
func (c *udpConn) ReadMultiBuffer() (buf.MultiBuffer, error) {
b, ok := <-c.egress
b, ok := c.queue.pop()
if !ok {
return nil, io.EOF
}
@@ -100,7 +95,7 @@ func (c *udpConn) ReadMultiBuffer() (buf.MultiBuffer, error) {
// Read packets from the connection
func (c *udpConn) Read(p []byte) (int, error) {
b, ok := <-c.egress
b, ok := c.queue.pop()
if !ok {
return 0, io.EOF
}
@@ -137,7 +132,7 @@ func (c *udpConn) Write(p []byte) (int, error) {
// sending packets back mean sending payload with source/destination reversed
err := c.handler.writePacket(p, c.dst, c.src)
if err != nil {
return 0, nil
return 0, err
}
return len(p), nil
@@ -168,3 +163,68 @@ func (c *udpConn) SetReadDeadline(t time.Time) error {
func (c *udpConn) SetWriteDeadline(t time.Time) error {
return nil
}
// udpQueueLimit bounds the packets waiting for one UDP flow; more are dropped.
const udpQueueLimit = 1024
// packetQueue holds the packets waiting for one UDP flow. Unlike a buffered
// channel of the same bound it only allocates for packets actually queued, so
// the many idle flows kept until the idle timeout cost next to nothing.
type packetQueue struct {
mu sync.Mutex
items []*buf.Buffer
limit int
notify chan struct{}
closed bool
}
func newPacketQueue(limit int) *packetQueue {
return &packetQueue{limit: limit, notify: make(chan struct{}, 1)}
}
// push queues b and reports whether it was accepted.
func (q *packetQueue) push(b *buf.Buffer) bool {
q.mu.Lock()
defer q.mu.Unlock()
if q.closed || len(q.items) >= q.limit {
return false
}
q.items = append(q.items, b)
select {
case q.notify <- struct{}{}:
default:
}
return true
}
// pop blocks until a packet is queued or the queue is closed and drained.
func (q *packetQueue) pop() (*buf.Buffer, bool) {
for {
q.mu.Lock()
if len(q.items) > 0 {
b := q.items[0]
q.items[0] = nil
q.items = q.items[1:]
if len(q.items) == 0 {
q.items = nil
}
q.mu.Unlock()
return b, true
}
if q.closed {
q.mu.Unlock()
return nil, false
}
q.mu.Unlock()
<-q.notify
}
}
func (q *packetQueue) close() {
q.mu.Lock()
defer q.mu.Unlock()
if !q.closed {
q.closed = true
close(q.notify)
}
}
proxy/tun/udp_fullcone_test.go
+91
View File
@@ -0,0 +1,91 @@
package tun
import (
"testing"
"time"
"github.com/xtls/xray-core/common/buf"
"github.com/xtls/xray-core/common/net"
)
// BenchmarkUDPNewSession measures what one new UDP flow costs the
// inbound while it stays open: QUIC and DNS open many short flows, and each
// one lives until the connection idle timeout.
func BenchmarkUDPNewSession(b *testing.B) {
u := newUdpConnectionHandler(func(conn net.Conn, dest net.Destination) {}, nil)
dst := net.UDPDestination(net.ParseAddress("1.1.1.1"), 443)
b.ReportAllocs()
b.ResetTimer()
for i := 0; i < b.N; i++ {
src := net.UDPDestination(net.IPAddress([]byte{10, byte(i >> 16), byte(i >> 8), byte(i)}), net.Port(1024+i%60000))
u.HandlePacket(src, dst, &buf.Buffer{})
}
}
func TestPacketQueueOrderAndClose(t *testing.T) {
q := newPacketQueue(udpQueueLimit)
for i := 0; i < 3; i++ {
b := buf.New()
b.WriteByte(byte(i))
if !q.push(b) {
t.Fatalf("push %d rejected", i)
}
}
for i := 0; i < 3; i++ {
b, ok := q.pop()
if !ok || b.Byte(0) != byte(i) {
t.Fatalf("pop %d: got %v, %v", i, b, ok)
}
b.Release()
}
q.close()
if _, ok := q.pop(); ok {
t.Fatal("pop after close returned a packet")
}
if q.push(&buf.Buffer{}) {
t.Fatal("push after close accepted")
}
}
func TestPacketQueueLimit(t *testing.T) {
q := newPacketQueue(udpQueueLimit)
for i := 0; i < udpQueueLimit; i++ {
if !q.push(&buf.Buffer{}) {
t.Fatalf("push %d rejected below the limit", i)
}
}
if q.push(&buf.Buffer{}) {
t.Fatal("push above the limit accepted")
}
}
func TestPacketQueueCloseUnblocksReader(t *testing.T) {
q := newPacketQueue(udpQueueLimit)
done := make(chan bool)
go func() {
_, ok := q.pop()
done <- ok
}()
q.close()
select {
case ok := <-done:
if ok {
t.Fatal("blocked pop returned a packet after close")
}
case <-time.After(time.Second):
t.Fatal("close did not wake the reader")
}
}
func TestPacketQueueDropsDrainedStorage(t *testing.T) {
q := newPacketQueue(udpQueueLimit)
for i := 0; i < 100; i++ {
q.push(&buf.Buffer{})
}
for i := 0; i < 100; i++ {
q.pop()
}
if q.items != nil {
t.Fatalf("drained queue still holds %d slots", cap(q.items))
}
}
proxy/tun/udp_queue_bench_test.go
+82
View File
@@ -0,0 +1,82 @@
package tun
import (
"runtime"
"testing"
"github.com/xtls/xray-core/common/buf"
)
const benchBatch = 64
// Raw cost of queueing and draining a small burst, as one flow's reader does.
func BenchmarkQueueBurstChan(b *testing.B) {
ch := make(chan *buf.Buffer, udpQueueLimit)
p := &buf.Buffer{}
b.ReportAllocs()
for i := 0; i < b.N; i++ {
for j := 0; j < benchBatch; j++ {
ch <- p
}
for j := 0; j < benchBatch; j++ {
<-ch
}
}
}
func BenchmarkQueueBurstPacketQueue(b *testing.B) {
q := newPacketQueue(udpQueueLimit)
p := &buf.Buffer{}
b.ReportAllocs()
for i := 0; i < b.N; i++ {
for j := 0; j < benchBatch; j++ {
q.push(p)
}
for j := 0; j < benchBatch; j++ {
q.pop()
}
}
}
// Producer and consumer on different goroutines; the producer yields when the
// queue is full instead of spinning, like a blocking channel send would.
func BenchmarkQueueStreamChan(b *testing.B) {
ch := make(chan *buf.Buffer, udpQueueLimit)
p := &buf.Buffer{}
done := make(chan struct{})
go func() {
for range ch {
}
close(done)
}()
b.ReportAllocs()
b.ResetTimer()
for i := 0; i < b.N; i++ {
ch <- p
}
close(ch)
<-done
}
func BenchmarkQueueStreamPacketQueue(b *testing.B) {
q := newPacketQueue(udpQueueLimit)
p := &buf.Buffer{}
done := make(chan struct{})
go func() {
for {
if _, ok := q.pop(); !ok {
break
}
}
close(done)
}()
b.ReportAllocs()
b.ResetTimer()
for i := 0; i < b.N; i++ {
for !q.push(p) {
runtime.Gosched()
}
}
q.close()
<-done
}