package tun import ( "encoding/binary" "io" "math" "math/rand/v2" "slices" "sync" "time" "github.com/xtls/xray-core/common/buf" "github.com/xtls/xray-core/common/errors" "github.com/xtls/xray-core/common/net" ) const ( tcpFin = 1 << iota tcpSyn tcpRst tcpPsh tcpAck ) const ( // tcpWindow is what a connection takes from the system before its reader has to take it, // and the most that it sends to the system before that is acknowledged. More than that, for several // connections at once, is more packets than the queue of a tun holds (500 on Linux) tcpWindow = 256 << 10 // tcpLargeWindow is the same for a tun that takes and gives segments that are larger than its MTU: those are // few, and a way with a longer round trip, as a router has to what is behind it, is not held back by it tcpLargeWindow = 1 << 20 // tcpWindowShift makes them fit the 16 bits of a window tcpWindowShift = 5 // tcpHeld is how many segments wait for one that was lost before them, // and of how many parts what the system says it has is known 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 tcpOpening = 65535 // tcpStart is how many segments are sent before any of them is acknowledged tcpStart = 10 // tcpSegment is the largest payload of a segment that the tun splits by itself, // tcpParts of how many parts of what was written it is made at most tcpSegment = math.MaxUint16 - ipv6HeaderSize - tcpHeaderSize tcpParts = 7 // 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 tcpRetransmitTicks = 3 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 tcpSynRetransmits = 5 // ticks between the probes of a window that stays closed tcpProbeTicks = 10 // tcpLinger is how long a closed connection waits for the system to close its side tcpLinger = 30 * time.Second ) var ( errTCPReset = errors.New("connection reset") errTCPTimeout = errors.New("connection timed out") ) // segment is a TCP segment that came from the tun type segment struct { seq, ack uint32 flags byte window uint16 options []byte payload []byte } // nextOption returns the first option of o and what follows it, kind is zero when there is none func nextOption(o []byte) (kind byte, value, rest []byte) { for len(o) > 0 && o[0] == 1 { o = o[1:] } if len(o) < 2 || int(o[1]) < 2 || int(o[1]) > len(o) { return 0, nil, nil } return o[0], o[2:o[1]], o[o[1]:] } // held is a segment that came before the one in front of it type held struct { seq uint32 data *buf.Buffer } // tcpConn is a TCP connection that the system opened through the tun. // 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, // tells the system what it has (SACK), sends again what the system misses, and sends no more at once // than what has arrived before (slow start and congestion avoidance as in RFC 5681) type tcpConn struct { stack *stack flow flow src, dst net.Destination // ip is the IP header of what is sent, without the length ip [ipv6HeaderSize]byte // ipSize is the size of that header ipSize int // mss is the largest payload of a segment that is sent mss int // window is tcpWindow or tcpLargeWindow window int32 // sndShift and rcvShift are the window scale of what the system and we advertise sndShift, rcvShift uint8 // sack is set if the system takes selective acknowledgments sack bool mu sync.Mutex readable sync.Cond writable sync.Cond established bool // closed is set when Close is called closed bool // err is why the connection is gone err error rcvNxt uint32 // rcvAcked is the last acknowledgment number sent rcvAcked uint32 // rcvAdvertised is the last window sent, in bytes rcvAdvertised int32 rcvQueue buf.MultiBuffer rcvLen int32 rcvFin bool // rcvEnd is where what the system sends ends, if rcvEnds: its FIN came with a segment that waits in rcvHeld rcvEnd uint32 rcvEnds bool // rcvHeld is what came after something that did not, in the order of seq and without overlaps rcvHeld []held // rcvLast is the seq of the last one that was put into rcvHeld rcvLast uint32 sndUna, sndNxt uint32 sndWnd uint32 // sndQueue is what was sent and is not acknowledged sndQueue buf.MultiBuffer // sndBase is the sequence number of the first byte of sndQueue sndBase uint32 sndFin bool // writing is how many of the last of sndQueue WriteMultiBuffer is sending, it is the only one to release them writing int // blocked is set while WriteMultiBuffer waits for the window blocked bool // sending is set while WriteMultiBuffer writes a segment without mu, ackWanted if sendAck was called meanwhile sending, ackWanted bool dupAcks int // cwnd is how much the way to the system takes at once, as far as that is known: it grows while what is sent // arrives, by as much as arrives below ssthresh, and is cut when something is lost cwnd, ssthresh uint32 // sacked is what the system says it has past sndUna, in order and apart from each other sacked [][2]uint32 // recovering is set from a sign of loss until what was sent by then, up to recover, is acknowledged recovering bool 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 // 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 timedAt time.Time // srtt is how long an acknowledgment takes, rttvar how much that varies (RFC 6298) srtt, rttvar time.Duration timer *time.Timer armed bool // stalled counts ticks without an acknowledgment of what waits for one stalled, retransmitTicks, retransmits int lingered bool } func newTCPConn(s *stack, f flow, syn segment) *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(), retransmitTicks: tcpRetransmitTicks, } c.readable.L = &c.mu c.writable.L = &c.mu c.ipSize = putIPHeader(c.ip[:], protoTCP, f.dst.Addr(), f.src.Addr()) c.mss = s.mtu - c.ipSize - tcpHeaderSize for kind, value, o := nextOption(syn.options); kind != 0; kind, value, o = nextOption(o) { switch { case kind == 2 && len(value) == 2: c.mss = min(c.mss, int(binary.BigEndian.Uint16(value))) case kind == 3 && len(value) == 1: c.sndShift = min(value[0], 14) c.rcvShift = tcpWindowShift case kind == 4: c.sack = true } } c.mss = max(c.mss, tcpSmallest) c.cwnd = tcpStart * uint32(c.mss) c.sndUna = rand.Uint32() c.sndNxt = c.sndUna + 1 return c } // 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)] binary.BigEndian.PutUint16(t, c.flow.dst.Port()) binary.BigEndian.PutUint16(t[2:], c.flow.src.Port()) binary.BigEndian.PutUint32(t[4:], seq) binary.BigEndian.PutUint32(t[8:], c.rcvNxt) t[12] = byte(len(t)/4) << 4 t[13] = flags window := max(c.window-c.rcvLen, 0) if flags&tcpSyn != 0 { // the window of a SYN is not scaled binary.BigEndian.PutUint16(t[14:], uint16(min(window, math.MaxUint16))) } else { 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) setIPLength(b[:n], n+len(t)+size) c.rcvAcked = c.rcvNxt c.rcvAdvertised = window return n + len(t) } // write completes the headers h with the checksum of the segment and writes it to the tun. // A segment that is not written is one that is lost: it is sent again when it is not acknowledged func (c *tcpConn) write(h, payload []byte) { t := h[c.ipSize:] binary.BigEndian.PutUint16(t[16:], transportChecksum(h[:c.ipSize], protoTCP, t, payload)) _ = c.stack.tun.WritePacket(h, payload) } // send writes a segment to the tun. c.mu is held func (c *tcpConn) send(seq uint32, flags byte, options, payload []byte) { var b [ipv6HeaderSize + tcpHeaderSize + 40]byte c.write(b[:c.header(b[:], seq, flags, options, len(payload))], payload) } func (c *tcpConn) sendSynAck() { options := [12]byte{2, 4} binary.BigEndian.PutUint16(options[2:], uint16(c.stack.mtu-c.ipSize-tcpHeaderSize)) n := 4 if c.sack { n += copy(options[n:], []byte{1, 1, 4, 2}) } if c.rcvShift != 0 { n += copy(options[n:], []byte{1, 3, 3, c.rcvShift}) } c.send(c.sndUna, tcpSyn|tcpAck, options[:n], nil) } // sendAck acknowledges what was received, and tells what of rcvHeld waits for a lost segment func (c *tcpConn) sendAck() { if c.sending { // after the segment that is being written: it was made before, and what the system is told goes in that order c.ackWanted = true return } if !c.sack || len(c.rcvHeld) == 0 { c.send(c.sndNxt, tcpAck, nil, nil) return } // the block with the segment that came last goes first var blocks [4][2]uint32 n, last := 1, false for i := 0; i < len(c.rcvHeld); { start, end, recent := c.rcvHeld[i].seq, c.rcvHeld[i].seq, false for ; i < len(c.rcvHeld) && c.rcvHeld[i].seq == end; i++ { recent = recent || end == c.rcvLast end += uint32(c.rcvHeld[i].data.Len()) } if recent { blocks[0], last = [2]uint32{start, end}, true } else if n < len(blocks) { blocks[n] = [2]uint32{start, end} n++ } } list := blocks[:n] if !last { list = list[1:] } options := [36]byte{1, 1, 5, byte(2 + 8*len(list))} for i, block := range list { binary.BigEndian.PutUint32(options[4+8*i:], block[0]) binary.BigEndian.PutUint32(options[8+8*i:], block[1]) } c.send(c.sndNxt, tcpAck, options[:4+8*len(list)], nil) } // 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 { c.mu.Lock() defer c.mu.Unlock() if c.err != nil { return s.flags&(tcpSyn|tcpAck|tcpRst) != tcpSyn } 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 { c.abort(errTCPReset) } return true } if s.flags&tcpSyn != 0 { switch { case !c.established && s.seq+1 == c.rcvNxt: c.sendSynAck() c.timedAt = time.Time{} case c.closed && s.flags&tcpAck == 0: c.abort(errTCPReset) return false default: c.sendAck() } return true } if s.flags&tcpAck == 0 { return true } if !c.established { if s.ack != c.sndNxt { return true } c.established = true c.sndUna = s.ack c.sndBase = s.ack c.sndWnd = uint32(s.window) << c.sndShift c.stalled, c.retransmits = 0, 0 if !c.timedAt.IsZero() { c.clocked() } c.stack.opening.Add(-1) go c.stack.handle(c, c.dst) } else { c.acknowledge(s) } c.receive(s) if c.sndFin && c.rcvFin && c.sndUna == c.sndNxt { // both sides have closed it and nothing waits to be acknowledged c.abort(io.ErrClosedPipe) } return true } // acknowledge takes what a segment says about what was sent. c.mu is held func (c *tcpConn) acknowledge(s segment) { acked, sent := s.ack-c.sndUna, c.sndNxt-c.sndUna if acked > sent { // older than the last one, or of what was not sent return } window := uint32(s.window) << c.sndShift if acked != 0 { c.sndUna = s.ack c.dupAcks, c.stalled, c.retransmits = 0, 0, 0 c.release() n := 0 for n < len(c.sacked) && int32(c.sacked[n][0]-c.sndUna) <= 0 { n++ } c.sacked = slices.Delete(c.sacked, 0, n) if int32(c.resent-c.sndUna) < 0 { c.resent = c.sndUna } if !c.timedAt.IsZero() && int32(c.sndUna-c.timed) >= 0 { c.clocked() } switch { case c.recovering && int32(c.sndUna-c.recover) >= 0: c.recovering = false case c.recovering: // what was sent again has arrived, and what was sent before it has not: the next segment is lost as well if end := c.sndUna + min(uint32(c.mss), c.recover-c.sndUna); int32(end-c.lost) > 0 { c.lost = end } } // 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) } 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)) } } } else if sent != 0 && window == c.sndWnd && len(s.payload) == 0 && s.flags&tcpFin == 0 { c.dupAcks++ } if c.sack { for kind, value, o := nextOption(s.options); kind != 0; kind, value, o = nextOption(o) { for ; kind == 5 && len(value) >= 8; value = value[8:] { // of what was sent and is not acknowledged start, end := binary.BigEndian.Uint32(value)-c.sndUna, binary.BigEndian.Uint32(value[4:])-c.sndUna if start != 0 && start < end && end <= c.sndNxt-c.sndUna { c.sackAdd(c.sndUna+start, c.sndUna+end) } } } } 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 c.congested() c.lost = c.sndUna + min(uint32(c.mss), c.sndNxt-c.sndUna) } if c.recovering { c.resend() } c.sndWnd = window // the writer goes on when it can write a good part of the window, not for every segment that is acknowledged. // Unless the way to the system is what holds it back: that takes what the acknowledgments say has left it, // and what is sent as they come does not come to it all at once limit := min(c.sndWnd, uint32(c.window)) enough := limit / 2 if c.cwnd < limit { enough = uint32(c.mss) } if c.blocked && (c.sndUna == c.sndNxt || c.space() >= enough) { c.writable.Broadcast() } } // clocked takes the time since timedAt as how long an acknowledgment takes. The next ones are waited for as long // as that says and no less than a few ticks. Until there is such a time again, after something was sent again, // they are waited for as long as the last time. c.mu is held func (c *tcpConn) clocked() { d := time.Since(c.timedAt) c.timedAt = time.Time{} if c.srtt == 0 { c.srtt, c.rttvar = d, d/2 } else { 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) } // sackAdd puts what the system says it has into sacked. c.mu is held func (c *tcpConn) sackAdd(start, end uint32) { i := 0 for i < len(c.sacked) && int32(c.sacked[i][1]-start) < 0 { i++ } j := i for ; j < len(c.sacked) && int32(c.sacked[j][0]-end) <= 0; j++ { if int32(c.sacked[j][0]-start) < 0 { start = c.sacked[j][0] } if int32(c.sacked[j][1]-end) > 0 { end = c.sacked[j][1] } } if i == j && len(c.sacked) == tcpHeld { return } c.sacked = slices.Replace(c.sacked, i, j, [2]uint32{start, end}) } // congested is called when something is lost: the way to the system takes half of what it was given, // and what is lost is sent again from sndUna. c.mu is held 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{} } // hole returns the first part at or after seq of what is lost: what the system does not have // before something that it says it has, or before lost. There is none if it ends where it starts. c.mu is held func (c *tcpConn) hole(seq uint32) (uint32, uint32) { end := c.lost for _, b := range c.sacked { if int32(b[0]-seq) > 0 { end = b[0] break } if int32(b[1]-seq) > 0 { seq = b[1] } } if int32(end-seq) < 0 { end = seq } return seq, end } // flight is how much of what was sent is on its way: not what the system says it has, // and not what is lost and was not sent again. c.mu is held func (c *tcpConn) flight() uint32 { n := c.sndNxt - c.sndUna if !c.recovering { return n } seq := c.resent for _, b := range c.sacked { n -= b[1] - b[0] if int32(b[0]-seq) > 0 { n -= b[0] - seq } if int32(b[1]-seq) > 0 { seq = b[1] } } if int32(c.lost-seq) > 0 { n -= c.lost - seq } return n } // resend sends again what is lost and was not yet: the segment that the system waits for at once, // the rest as the way to it takes it. c.mu is held func (c *tcpConn) resend() { for { start, end := c.hole(c.resent) if start == end { return } room := uint32(c.mss) if start != c.sndUna { flight := c.flight() if flight+room > c.cwnd { return } room = c.cwnd - flight } end = start + min(end-start, room) c.retransmit(start, end) c.resent = end } } // space is how much can be sent now: what the system takes, and what the way to it takes. c.mu is held func (c *tcpConn) space() uint32 { window, sent, flight := min(c.sndWnd, uint32(c.window)), c.sndNxt-c.sndUna, c.flight() if sent >= window || flight >= c.cwnd { return 0 } return min(window-sent, c.cwnd-flight) } // receive takes the payload of a segment. c.mu is held func (c *tcpConn) receive(s segment) { seq, payload, fin := s.seq, s.payload, s.flags&tcpFin != 0 if len(payload) == 0 && !fin { if int32(c.rcvNxt-seq) > 0 { // a keepalive, or a probe of a window that the system has as closed: answered with what is expected c.sendAck() } return } if before := int32(c.rcvNxt - seq); before > 0 { if int(before) > len(payload) || int(before) == len(payload) && !fin { // sent again, or a keepalive: answered with what is expected c.sendAck() return } seq, payload = c.rcvNxt, payload[before:] } if c.rcvFin { c.sendAck() return } if c.closed && len(payload) > 0 { // nobody reads it anymore c.reset() return } // the system is told at once when something is wrong, or it sends more of the same now := len(c.rcvHeld) > 0 if room := int(c.window - c.rcvLen - int32(seq-c.rcvNxt)); len(payload) > room { payload, fin, now = payload[:max(room, 0)], false, true } if seq != c.rcvNxt { // something before it was lost, it waits for that, and so does the end if this is it c.hold(seq, payload) if fin { c.rcvEnd, c.rcvEnds = seq+uint32(len(payload)), true } c.sendAck() return } c.rcvQueue = buf.MergeBytes(c.rcvQueue, payload) c.rcvLen += int32(len(payload)) c.rcvNxt += uint32(len(payload)) for !fin && len(c.rcvHeld) > 0 && int32(c.rcvNxt-c.rcvHeld[0].seq) >= 0 { h := c.rcvHeld[0] if before := int32(c.rcvNxt - h.seq); before < h.data.Len() { c.rcvQueue = buf.MergeBytes(c.rcvQueue, h.data.BytesFrom(before)) c.rcvLen += h.data.Len() - before c.rcvNxt += uint32(h.data.Len() - before) } h.data.Release() c.rcvHeld = c.rcvHeld[1:] } if fin = fin || c.rcvEnds && c.rcvNxt == c.rcvEnd; fin { c.rcvFin = true c.rcvNxt++ } c.readable.Signal() // the reader acknowledges what it takes, with one segment for all that came by then. // It is not waited for when it does not read for long if fin || now || c.rcvNxt-c.rcvAcked >= tcpWindow/4 { c.sendAck() } else { c.arm(tcpTick) } } // release releases what of sndQueue is acknowledged. c.mu is held func (c *tcpConn) release() { for len(c.sndQueue) > c.writing { end := c.sndBase + uint32(c.sndQueue[0].Len()) if int32(c.sndUna-end) < 0 { break } c.sndQueue[0].Release() c.sndQueue[0] = nil c.sndQueue = c.sndQueue[1:] c.sndBase = end } } // 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++ } // without what the one before and the one after have if i > 0 { if over := int32(c.rcvHeld[i-1].seq + uint32(c.rcvHeld[i-1].data.Len()) - seq); over > 0 { if int(over) >= len(payload) { return } seq, payload = seq+uint32(over), payload[over:] } } if i < len(c.rcvHeld) { if over := int32(seq + uint32(len(payload)) - c.rcvHeld[i].seq); over > 0 { if int(over) >= len(payload) { return } payload = payload[:len(payload)-int(over)] } } if len(payload) == 0 || len(c.rcvHeld) == tcpHeld { return } b := buf.NewWithSize(int32(len(payload))) b.Write(payload) c.rcvHeld = slices.Insert(c.rcvHeld, i, held{seq, b}) c.rcvLast = seq } // retransmit sends again what was sent from seq to end. c.mu is held func (c *tcpConn) retransmit(seq, end uint32) { fin := false if int32(end-c.sndNxt) >= 0 { end, fin = c.sndNxt, c.sndFin if fin { end-- } } for int32(end-seq) > 0 { var part [1][]byte _, n := c.queued(part[:0], seq, min(c.mss, int(end-seq))) if n == 0 { break } c.send(seq, tcpAck, nil, part[0]) seq += uint32(n) } if fin && seq == end { c.send(end, tcpFin|tcpAck, nil, nil) } } // queued adds to parts what of sndQueue starts at seq, size bytes at most and as many parts as it has room for, // and returns them with their size. c.mu is held func (c *tcpConn) queued(parts [][]byte, seq uint32, size int) ([][]byte, int) { skip, n := int32(seq-c.sndBase), 0 for _, b := range c.sndQueue { if skip >= b.Len() { skip -= b.Len() continue } if n == size || len(parts) == cap(parts) { break } p := b.BytesFrom(skip) skip = 0 p = p[:min(len(p), size-n)] parts = append(parts, p) n += len(p) } return parts, n } // 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 { return } c.armed = true if c.timer == nil { c.timer = time.AfterFunc(d, c.tick) } else { c.timer.Reset(d) } } // tick does what was put off and what has to be done again func (c *tcpConn) tick() { c.mu.Lock() defer c.mu.Unlock() c.armed = false if c.err != nil { return } if c.rcvAcked != c.rcvNxt { c.sendAck() } 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.abort(errTCPTimeout) return } c.stalled = 0 c.retransmitTicks = min(c.retransmitTicks*2, tcpRetransmitTicksMost) if !c.established { c.sendSynAck() 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 c.congested() c.cwnd, c.lost = uint32(c.mss), c.sndNxt c.resend() } } c.arm(tcpTick) case c.blocked: // the segment that opens the window may have been lost, this one is answered with the window if c.stalled++; c.stalled%tcpProbeTicks == 0 { c.send(c.sndUna-1, tcpAck, nil, nil) } c.arm(tcpTick) case c.closed: if c.lingered { c.reset() return } c.lingered = true c.arm(tcpLinger) } } // reset tells the system that the connection is gone. c.mu is held func (c *tcpConn) reset() { c.send(c.sndNxt, tcpRst|tcpAck, nil, nil) c.abort(io.ErrClosedPipe) } // abort ends the connection without another word to the system. What it has sent is still there to read. // c.mu is held func (c *tcpConn) abort(err error) { if c.err != nil { return } c.err = err if c.timer != nil { c.timer.Stop() } for _, h := range c.rcvHeld { h.data.Release() } c.rcvHeld = nil for _, b := range c.sndQueue[:len(c.sndQueue)-c.writing] { b.Release() } c.sndQueue = nil c.readable.Broadcast() c.writable.Broadcast() c.stack.remove(c) } // wait waits for something to read. c.mu is held func (c *tcpConn) wait() error { for c.rcvLen == 0 && !c.rcvFin && !c.closed && c.err == nil { c.readable.Wait() } switch { case c.rcvLen > 0: return nil case c.err != nil && !c.rcvFin: return c.err } return io.EOF } // taken is called when size bytes have been read. c.mu is held func (c *tcpConn) taken(size int32) { c.rcvLen -= size // the system may also have stopped at the window that it was told if c.err == nil && (c.rcvAcked != c.rcvNxt || c.rcvAdvertised < c.window/2) { c.sendAck() } } // ReadMultiBuffer implements buf.Reader func (c *tcpConn) ReadMultiBuffer() (buf.MultiBuffer, error) { c.mu.Lock() defer c.mu.Unlock() if err := c.wait(); err != nil { return nil, err } mb := c.rcvQueue c.rcvQueue = nil c.taken(c.rcvLen) return mb, nil } // Read implements net.Conn func (c *tcpConn) Read(p []byte) (int, error) { c.mu.Lock() defer c.mu.Unlock() if err := c.wait(); err != nil { return 0, err } var n int c.rcvQueue, n = buf.SplitBytes(c.rcvQueue, p) c.taken(int32(n)) return n, nil } // WriteMultiBuffer implements buf.Writer. It is not called again before it returns func (c *tcpConn) WriteMultiBuffer(mb buf.MultiBuffer) error { c.mu.Lock() defer c.mu.Unlock() // without the empty ones n, size := 0, uint32(0) for _, b := range mb { if b.IsEmpty() { b.Release() continue } mb[n] = b n++ size += uint32(b.Len()) } mb = mb[:n] if c.err != nil || c.closed { buf.ReleaseMulti(mb) return io.ErrClosedPipe } // what is written is in sndQueue until it is acknowledged, and is not released before all of it is sent c.sndQueue = append(c.sndQueue, mb...) 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] if c.stack.segmenter != nil { limit, parts = tcpSegment, 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 for c.err == nil { if space := c.space(); space >= min(end-c.sndNxt, uint32(c.mss)) || space != 0 && c.sndUna == c.sndNxt { break } c.blocked = true c.arm(tcpTick) c.writable.Wait() c.blocked = false } if c.err != nil { break } most := min(limit, int(c.space())) if most > c.mss { // the same for what the tun splits most -= most % c.mss } p, n := c.queued(parts, c.sndNxt, most) flags := byte(tcpAck) if c.sndNxt+uint32(n) == end { flags |= tcpPsh } var b [ipv6HeaderSize + tcpHeaderSize]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 c.stalled = 0 } c.sndNxt += uint32(n) if c.timedAt.IsZero() { c.timed, c.timedAt = c.sndNxt, time.Now() } c.arm(tcpTick) // segments that come in the meantime are taken while this one is written, and acknowledged after it c.sending = true c.mu.Unlock() if s := c.stack.segmenter; s != nil { t := h[c.ipSize:] binary.BigEndian.PutUint16(t[16:], ^checksum(pseudoSum(h[:c.ipSize], protoTCP, len(t)+n))) _ = s.WriteSegment(h, p, c.mss) } else { c.write(h, p[0]) } c.mu.Lock() c.sending = false if c.ackWanted && c.err == nil { c.sendAck() } c.ackWanted = false } c.writing = 0 if c.err != nil { buf.ReleaseMulti(mb) return io.ErrClosedPipe } c.release() return nil } // Write implements net.Conn func (c *tcpConn) Write(p []byte) (int, error) { if err := c.WriteMultiBuffer(buf.MergeBytes(nil, p)); err != nil { return 0, err } return len(p), nil } // Close implements net.Conn func (c *tcpConn) Close() error { c.mu.Lock() defer c.mu.Unlock() if c.closed { return nil } c.closed = true unread := c.rcvLen > 0 c.rcvQueue, c.rcvLen = buf.ReleaseMulti(c.rcvQueue), 0 if c.err != nil { return nil } if unread || c.writing != 0 { // something was not read or not written, the system has to know c.reset() return nil } c.readable.Broadcast() c.send(c.sndNxt, tcpFin|tcpAck, nil, nil) c.sndFin = true c.sndNxt++ c.arm(tcpTick) return nil } func (c *tcpConn) LocalAddr() net.Addr { return c.dst.RawNetAddr() } func (c *tcpConn) RemoteAddr() net.Addr { return c.src.RawNetAddr() } func (c *tcpConn) SetDeadline(time.Time) error { return nil } func (c *tcpConn) SetReadDeadline(time.Time) error { return nil } func (c *tcpConn) SetWriteDeadline(time.Time) error { return nil }