merge kcp-go kcptun. update p2p protocol

README.md

@@ -1,8 +1,6 @@
 ### Introduction
->A p2p tunnel which establish port forwarding channel for both clients for some special uses, and whitch
- based on kcptun
-
->Need one server with public ip and only one udp port for hole digging. 
+>A p2p tunnel(based on kcptun) which establishs port forwarding channel for both clients.
+>Need one server with public ip and one udp port for hole digging. 
 
 ### QuickStart
 Download a corresponding one from precompiled [Releases](https://github.com/hikaricai/p2p_tun/releases).
@@ -14,13 +12,13 @@ Client_B: ./p2pclient -b ":UDP_PORT" -r "SERVER_IP:4000" -l ":10022" -t "TARGET_
 ```
 The above commands will establish port forwarding channel for both Client_A and Client_B as:
 
-> Application -> **Client_A(10022/tcp) -> Client_B** -> TARGET_IP_B:(22/tcp)
-> Application -> **Client_B(10022/tcp) -> Client_A** -> TARGET_IP_A:(22/tcp)
+> Application1 -> **Client_A(10022/tcp) -> Client_B** -> TARGET_IP_B:(22/tcp)  
+> Application2 -> **Client_B(10022/tcp) -> Client_A** -> TARGET_IP_A:(22/tcp)
 
 which tunnels the original connection:
 
-> Application -> TARGET_IP_B:(22/tcp)
-> Application -> TARGET_IP_B:(22/tcp)
+> Application1 -> TARGET_IP_B:(22/tcp)  
+> Application2 -> TARGET_IP_A:(22/tcp)
 
 ### Install from source
 

buildp2p.sh

@@ -0,0 +1,28 @@
+#!/bin/bash
+
+VERSION=`date -u +%Y%m%d`
+LDFLAGS="-X main.VERSION=$VERSION -s -w"
+GCFLAGS=""
+
+OSES=(linux windows)
+ARCHS=(amd64 386)
+for os in ${OSES[@]}; do
+        for arch in ${ARCHS[@]}; do
+                suffix=""
+                if [ "$os" == "windows" ]
+                then
+                        suffix=".exe"
+                fi
+                env CGO_ENABLED=0 GOOS=$os GOARCH=$arch go build -ldflags "$LDFLAGS" -gcflags "$GCFLAGS" -o p2pclient_${os}_${arch}${suffix} github.com/hikaricai/p2p_tun/p2pclient
+                env CGO_ENABLED=0 GOOS=$os GOARCH=$arch go build -ldflags "$LDFLAGS" -gcflags "$GCFLAGS" -o p2pserver_${os}_${arch}${suffix} github.com/hikaricai/p2p_tun/p2pserver
+                tar -zcf p2ptun-${os}-${arch}-$VERSION.tar.gz p2pclient_${os}_${arch}${suffix} p2pserver_${os}_${arch}${suffix}
+        done
+done
+
+# ARM
+ARMS=(7)
+for v in ${ARMS[@]}; do
+        env CGO_ENABLED=0 GOOS=linux GOARCH=arm GOARM=$v go build -ldflags "$LDFLAGS" -gcflags "$GCFLAGS" -o p2pclient_linux_arm$v  github.com/hikaricai/p2p_tun/p2pclient
+        env CGO_ENABLED=0 GOOS=linux GOARCH=arm GOARM=$v go build -ldflags "$LDFLAGS" -gcflags "$GCFLAGS" -o p2pserver_linux_arm$v  github.com/hikaricai/p2p_tun/p2pserver
+done
+tar -zcf p2ptun-linux-arm-$VERSION.tar.gz p2pclient_linux_arm* p2pserver_linux_arm*

kcp-go/batchconn.go

@@ -0,0 +1,12 @@
+package kcp
+
+import "golang.org/x/net/ipv4"
+
+const (
+	batchSize = 16
+)
+
+type batchConn interface {
+	WriteBatch(ms []ipv4.Message, flags int) (int, error)
+	ReadBatch(ms []ipv4.Message, flags int) (int, error)
+}

kcp-go/entropy.go

@@ -8,6 +8,7 @@ import (
 	"io"
 )
 
+// Entropy defines a entropy source
 type Entropy interface {
 	Init()
 	Fill(nonce []byte)

kcp-go/fec.go

@@ -11,36 +11,34 @@ const (
 	fecHeaderSize      = 6
 	fecHeaderSizePlus2 = fecHeaderSize + 2 // plus 2B data size
 	typeData           = 0xf1
-	typeFEC            = 0xf2
+	typeParity         = 0xf2
 )
 
-type (
-	// fecPacket is a decoded FEC packet
-	fecPacket struct {
-		seqid uint32
-		flag  uint16
-		data  []byte
-	}
+// fecPacket is a decoded FEC packet
+type fecPacket []byte
 
-	// fecDecoder for decoding incoming packets
-	fecDecoder struct {
-		rxlimit      int // queue size limit
-		dataShards   int
-		parityShards int
-		shardSize    int
-		rx           []fecPacket // ordered receive queue
+func (bts fecPacket) seqid() uint32 { return binary.LittleEndian.Uint32(bts) }
+func (bts fecPacket) flag() uint16  { return binary.LittleEndian.Uint16(bts[4:]) }
+func (bts fecPacket) data() []byte  { return bts[6:] }
 
-		// caches
-		decodeCache [][]byte
-		flagCache   []bool
+// fecDecoder for decoding incoming packets
+type fecDecoder struct {
+	rxlimit      int // queue size limit
+	dataShards   int
+	parityShards int
+	shardSize    int
+	rx           []fecPacket // ordered receive queue
 
-		// zeros
-		zeros []byte
+	// caches
+	decodeCache [][]byte
+	flagCache   []bool
 
-		// RS decoder
-		codec reedsolomon.Encoder
-	}
-)
+	// zeros
+	zeros []byte
+
+	// RS decoder
+	codec reedsolomon.Encoder
+}
 
 func newFECDecoder(rxlimit, dataShards, parityShards int) *fecDecoder {
 	if dataShards <= 0 || parityShards <= 0 {
@@ -66,33 +64,24 @@ func newFECDecoder(rxlimit, dataShards, parityShards int) *fecDecoder {
 	return dec
 }
 
-// decodeBytes a fec packet
-func (dec *fecDecoder) decodeBytes(data []byte) fecPacket {
-	var pkt fecPacket
-	pkt.seqid = binary.LittleEndian.Uint32(data)
-	pkt.flag = binary.LittleEndian.Uint16(data[4:])
-	// allocate memory & copy
-	buf := xmitBuf.Get().([]byte)[:len(data)-6]
-	copy(buf, data[6:])
-	pkt.data = buf
-	return pkt
-}
-
 // decode a fec packet
-func (dec *fecDecoder) decode(pkt fecPacket) (recovered [][]byte) {
+func (dec *fecDecoder) decode(in fecPacket) (recovered [][]byte) {
 	// insertion
 	n := len(dec.rx) - 1
 	insertIdx := 0
 	for i := n; i >= 0; i-- {
-		if pkt.seqid == dec.rx[i].seqid { // de-duplicate
-			xmitBuf.Put(pkt.data)
+		if in.seqid() == dec.rx[i].seqid() { // de-duplicate
 			return nil
-		} else if _itimediff(pkt.seqid, dec.rx[i].seqid) > 0 { // insertion
+		} else if _itimediff(in.seqid(), dec.rx[i].seqid()) > 0 { // insertion
 			insertIdx = i + 1
 			break
 		}
 	}
 
+	// make a copy
+	pkt := fecPacket(xmitBuf.Get().([]byte)[:len(in)])
+	copy(pkt, in)
+
 	// insert into ordered rx queue
 	if insertIdx == n+1 {
 		dec.rx = append(dec.rx, pkt)
@@ -103,11 +92,11 @@ func (dec *fecDecoder) decode(pkt fecPacket) (recovered [][]byte) {
 	}
 
 	// shard range for current packet
-	shardBegin := pkt.seqid - pkt.seqid%uint32(dec.shardSize)
+	shardBegin := pkt.seqid() - pkt.seqid()%uint32(dec.shardSize)
 	shardEnd := shardBegin + uint32(dec.shardSize) - 1
 
 	// max search range in ordered queue for current shard
-	searchBegin := insertIdx - int(pkt.seqid%uint32(dec.shardSize))
+	searchBegin := insertIdx - int(pkt.seqid()%uint32(dec.shardSize))
 	if searchBegin < 0 {
 		searchBegin = 0
 	}
@@ -130,21 +119,21 @@ func (dec *fecDecoder) decode(pkt fecPacket) (recovered [][]byte) {
 
 		// shard assembly
 		for i := searchBegin; i <= searchEnd; i++ {
-			seqid := dec.rx[i].seqid
+			seqid := dec.rx[i].seqid()
 			if _itimediff(seqid, shardEnd) > 0 {
 				break
 			} else if _itimediff(seqid, shardBegin) >= 0 {
-				shards[seqid%uint32(dec.shardSize)] = dec.rx[i].data
+				shards[seqid%uint32(dec.shardSize)] = dec.rx[i].data()
 				shardsflag[seqid%uint32(dec.shardSize)] = true
 				numshard++
-				if dec.rx[i].flag == typeData {
+				if dec.rx[i].flag() == typeData {
 					numDataShard++
 				}
 				if numshard == 1 {
 					first = i
 				}
-				if len(dec.rx[i].data) > maxlen {
-					maxlen = len(dec.rx[i].data)
+				if len(dec.rx[i].data()) > maxlen {
+					maxlen = len(dec.rx[i].data())
 				}
 			}
 		}
@@ -159,11 +148,14 @@ func (dec *fecDecoder) decode(pkt fecPacket) (recovered [][]byte) {
 					dlen := len(shards[k])
 					shards[k] = shards[k][:maxlen]
 					copy(shards[k][dlen:], dec.zeros)
+				} else {
+					shards[k] = xmitBuf.Get().([]byte)[:0]
 				}
 			}
 			if err := dec.codec.ReconstructData(shards); err == nil {
 				for k := range shards[:dec.dataShards] {
 					if !shardsflag[k] {
+						// recovered data should be recycled
 						recovered = append(recovered, shards[k])
 					}
 				}
@@ -174,7 +166,7 @@ func (dec *fecDecoder) decode(pkt fecPacket) (recovered [][]byte) {
 
 	// keep rxlimit
 	if len(dec.rx) > dec.rxlimit {
-		if dec.rx[0].flag == typeData { // track the unrecoverable data
+		if dec.rx[0].flag() == typeData { // track the unrecoverable data
 			atomic.AddUint64(&DefaultSnmp.FECShortShards, 1)
 		}
 		dec.rx = dec.freeRange(0, 1, dec.rx)
@@ -182,15 +174,16 @@ func (dec *fecDecoder) decode(pkt fecPacket) (recovered [][]byte) {
 	return
 }
 
-// free a range of fecPacket, and zero for GC recycling
+// free a range of fecPacket
 func (dec *fecDecoder) freeRange(first, n int, q []fecPacket) []fecPacket {
 	for i := first; i < first+n; i++ { // recycle buffer
-		xmitBuf.Put(q[i].data)
+		xmitBuf.Put([]byte(q[i]))
+	}
+
+	if first == 0 && n < cap(q)/2 {
+		return q[n:]
 	}
 	copy(q[first:], q[first+n:])
-	for i := 0; i < n; i++ { // dereference data
-		q[len(q)-1-i].data = nil
-	}
 	return q[:len(q)-n]
 }
 
@@ -252,13 +245,16 @@ func newFECEncoder(dataShards, parityShards, offset int) *fecEncoder {
 // encodes the packet, outputs parity shards if we have collected quorum datashards
 // notice: the contents of 'ps' will be re-written in successive calling
 func (enc *fecEncoder) encode(b []byte) (ps [][]byte) {
+	// The header format:
+	// | FEC SEQID(4B) | FEC TYPE(2B) | SIZE (2B) | PAYLOAD(SIZE-2) |
+	// |<-headerOffset                |<-payloadOffset
 	enc.markData(b[enc.headerOffset:])
 	binary.LittleEndian.PutUint16(b[enc.payloadOffset:], uint16(len(b[enc.payloadOffset:])))
 
-	// copy data to fec datashards
+	// copy data from payloadOffset to fec shard cache
 	sz := len(b)
 	enc.shardCache[enc.shardCount] = enc.shardCache[enc.shardCount][:sz]
-	copy(enc.shardCache[enc.shardCount], b)
+	copy(enc.shardCache[enc.shardCount][enc.payloadOffset:], b[enc.payloadOffset:])
 	enc.shardCount++
 
 	// track max datashard length
@@ -285,7 +281,7 @@ func (enc *fecEncoder) encode(b []byte) (ps [][]byte) {
 		if err := enc.codec.Encode(cache); err == nil {
 			ps = enc.shardCache[enc.dataShards:]
 			for k := range ps {
-				enc.markFEC(ps[k][enc.headerOffset:])
+				enc.markParity(ps[k][enc.headerOffset:])
 				ps[k] = ps[k][:enc.maxSize]
 			}
 		}
@@ -304,8 +300,9 @@ func (enc *fecEncoder) markData(data []byte) {
 	enc.next++
 }
 
-func (enc *fecEncoder) markFEC(data []byte) {
+func (enc *fecEncoder) markParity(data []byte) {
 	binary.LittleEndian.PutUint32(data, enc.next)
-	binary.LittleEndian.PutUint16(data[4:], typeFEC)
+	binary.LittleEndian.PutUint16(data[4:], typeParity)
+	// sequence wrap will only happen at parity shard
 	enc.next = (enc.next + 1) % enc.paws
 }

kcp-go/kcp.go

@@ -1,4 +1,3 @@
-// Package kcp - A Fast and Reliable ARQ Protocol
 package kcp
 
 import (
@@ -104,6 +103,7 @@ type segment struct {
 	xmit     uint32
 	resendts uint32
 	fastack  uint32
+	acked    uint32 // mark if the seg has acked
 	data     []byte
 }
 
@@ -144,8 +144,9 @@ type KCP struct {
 
 	acklist []ackItem
 
-	buffer []byte
-	output output_callback
+	buffer   []byte
+	reserved int
+	output   output_callback
 }
 
 type ackItem struct {
@@ -153,8 +154,11 @@ type ackItem struct {
 	ts uint32
 }
 
-// NewKCP create a new kcp control object, 'conv' must equal in two endpoint
-// from the same connection.
+// NewKCP create a new kcp state machine
+//
+// 'conv' must be equal in the connection peers, or else data will be silently rejected.
+//
+// 'output' function will be called whenever these is data to be sent on wire.
 func NewKCP(conv uint32, output output_callback) *KCP {
 	kcp := new(KCP)
 	kcp.conv = conv
@@ -163,7 +167,7 @@ func NewKCP(conv uint32, output output_callback) *KCP {
 	kcp.rmt_wnd = IKCP_WND_RCV
 	kcp.mtu = IKCP_MTU_DEF
 	kcp.mss = kcp.mtu - IKCP_OVERHEAD
-	kcp.buffer = make([]byte, (kcp.mtu+IKCP_OVERHEAD)*3)
+	kcp.buffer = make([]byte, kcp.mtu)
 	kcp.rx_rto = IKCP_RTO_DEF
 	kcp.rx_minrto = IKCP_RTO_MIN
 	kcp.interval = IKCP_INTERVAL
@@ -181,8 +185,24 @@ func (kcp *KCP) newSegment(size int) (seg segment) {
 }
 
 // delSegment recycles a KCP segment
-func (kcp *KCP) delSegment(seg segment) {
-	xmitBuf.Put(seg.data)
+func (kcp *KCP) delSegment(seg *segment) {
+	if seg.data != nil {
+		xmitBuf.Put(seg.data)
+		seg.data = nil
+	}
+}
+
+// ReserveBytes keeps n bytes untouched from the beginning of the buffer,
+// the output_callback function should be aware of this.
+//
+// Return false if n >= mss
+func (kcp *KCP) ReserveBytes(n int) bool {
+	if n >= int(kcp.mtu-IKCP_OVERHEAD) || n < 0 {
+		return false
+	}
+	kcp.reserved = n
+	kcp.mss = kcp.mtu - IKCP_OVERHEAD - uint32(n)
+	return true
 }
 
 // PeekSize checks the size of next message in the recv queue
@@ -210,19 +230,21 @@ func (kcp *KCP) PeekSize() (length int) {
 	return
 }
 
-// Recv is user/upper level recv: returns size, returns below zero for EAGAIN
+// Receive data from kcp state machine
+//
+// Return number of bytes read.
+//
+// Return -1 when there is no readable data.
+//
+// Return -2 if len(buffer) is smaller than kcp.PeekSize().
 func (kcp *KCP) Recv(buffer []byte) (n int) {
-	if len(kcp.rcv_queue) == 0 {
+	peeksize := kcp.PeekSize()
+	if peeksize < 0 {
 		return -1
 	}
 
-	peeksize := kcp.PeekSize()
-	if peeksize < 0 {
-		return -2
-	}
-
 	if peeksize > len(buffer) {
-		return -3
+		return -2
 	}
 
 	var fast_recover bool
@@ -238,7 +260,7 @@ func (kcp *KCP) Recv(buffer []byte) (n int) {
 		buffer = buffer[len(seg.data):]
 		n += len(seg.data)
 		count++
-		kcp.delSegment(*seg)
+		kcp.delSegment(seg)
 		if seg.frg == 0 {
 			break
 		}
@@ -382,10 +404,12 @@ func (kcp *KCP) parse_ack(sn uint32) {
 	for k := range kcp.snd_buf {
 		seg := &kcp.snd_buf[k]
 		if sn == seg.sn {
-			kcp.delSegment(*seg)
-			copy(kcp.snd_buf[k:], kcp.snd_buf[k+1:])
-			kcp.snd_buf[len(kcp.snd_buf)-1] = segment{}
-			kcp.snd_buf = kcp.snd_buf[:len(kcp.snd_buf)-1]
+			// mark and free space, but leave the segment here,
+			// and wait until `una` to delete this, then we don't
+			// have to shift the segments behind forward,
+			// which is an expensive operation for large window
+			seg.acked = 1
+			kcp.delSegment(seg)
 			break
 		}
 		if _itimediff(sn, seg.sn) < 0 {
@@ -394,7 +418,7 @@ func (kcp *KCP) parse_ack(sn uint32) {
 	}
 }
 
-func (kcp *KCP) parse_fastack(sn uint32) {
+func (kcp *KCP) parse_fastack(sn, ts uint32) {
 	if _itimediff(sn, kcp.snd_una) < 0 || _itimediff(sn, kcp.snd_nxt) >= 0 {
 		return
 	}
@@ -403,7 +427,7 @@ func (kcp *KCP) parse_fastack(sn uint32) {
 		seg := &kcp.snd_buf[k]
 		if _itimediff(sn, seg.sn) < 0 {
 			break
-		} else if sn != seg.sn {
+		} else if sn != seg.sn && _itimediff(seg.ts, ts) <= 0 {
 			seg.fastack++
 		}
 	}
@@ -414,7 +438,7 @@ func (kcp *KCP) parse_una(una uint32) {
 	for k := range kcp.snd_buf {
 		seg := &kcp.snd_buf[k]
 		if _itimediff(una, seg.sn) > 0 {
-			kcp.delSegment(*seg)
+			kcp.delSegment(seg)
 			count++
 		} else {
 			break
@@ -430,12 +454,12 @@ func (kcp *KCP) ack_push(sn, ts uint32) {
 	kcp.acklist = append(kcp.acklist, ackItem{sn, ts})
 }
 
-func (kcp *KCP) parse_data(newseg segment) {
+// returns true if data has repeated
+func (kcp *KCP) parse_data(newseg segment) bool {
 	sn := newseg.sn
 	if _itimediff(sn, kcp.rcv_nxt+kcp.rcv_wnd) >= 0 ||
 		_itimediff(sn, kcp.rcv_nxt) < 0 {
-		kcp.delSegment(newseg)
-		return
+		return true
 	}
 
 	n := len(kcp.rcv_buf) - 1
@@ -445,7 +469,6 @@ func (kcp *KCP) parse_data(newseg segment) {
 		seg := &kcp.rcv_buf[i]
 		if seg.sn == sn {
 			repeat = true
-			atomic.AddUint64(&DefaultSnmp.RepeatSegs, 1)
 			break
 		}
 		if _itimediff(sn, seg.sn) > 0 {
@@ -455,6 +478,11 @@ func (kcp *KCP) parse_data(newseg segment) {
 	}
 
 	if !repeat {
+		// replicate the content if it's new
+		dataCopy := xmitBuf.Get().([]byte)[:len(newseg.data)]
+		copy(dataCopy, newseg.data)
+		newseg.data = dataCopy
+
 		if insert_idx == n+1 {
 			kcp.rcv_buf = append(kcp.rcv_buf, newseg)
 		} else {
@@ -462,8 +490,6 @@ func (kcp *KCP) parse_data(newseg segment) {
 			copy(kcp.rcv_buf[insert_idx+1:], kcp.rcv_buf[insert_idx:])
 			kcp.rcv_buf[insert_idx] = newseg
 		}
-	} else {
-		kcp.delSegment(newseg)
 	}
 
 	// move available data from rcv_buf -> rcv_queue
@@ -481,18 +507,23 @@ func (kcp *KCP) parse_data(newseg segment) {
 		kcp.rcv_queue = append(kcp.rcv_queue, kcp.rcv_buf[:count]...)
 		kcp.rcv_buf = kcp.remove_front(kcp.rcv_buf, count)
 	}
+
+	return repeat
 }
 
-// Input when you received a low level packet (eg. UDP packet), call it
-// regular indicates a regular packet has received(not from FEC)
+// Input a packet into kcp state machine.
+//
+// 'regular' indicates it's a real data packet from remote, and it means it's not generated from ReedSolomon
+// codecs.
+//
+// 'ackNoDelay' will trigger immediate ACK, but surely it will not be efficient in bandwidth
 func (kcp *KCP) Input(data []byte, regular, ackNoDelay bool) int {
 	snd_una := kcp.snd_una
 	if len(data) < IKCP_OVERHEAD {
 		return -1
 	}
 
-	var maxack uint32
-	var lastackts uint32
+	var latest uint32 // the latest ack packet
 	var flag int
 	var inSegs uint64
 
@@ -535,20 +566,15 @@ func (kcp *KCP) Input(data []byte, regular, ackNoDelay bool) int {
 
 		if cmd == IKCP_CMD_ACK {
 			kcp.parse_ack(sn)
-			kcp.shrink_buf()
-			if flag == 0 {
-				flag = 1
-				maxack = sn
-				lastackts = ts
-			} else if _itimediff(sn, maxack) > 0 {
-				maxack = sn
-				lastackts = ts
-			}
+			kcp.parse_fastack(sn, ts)
+			flag |= 1
+			latest = ts
 		} else if cmd == IKCP_CMD_PUSH {
+			repeat := true
 			if _itimediff(sn, kcp.rcv_nxt+kcp.rcv_wnd) < 0 {
 				kcp.ack_push(sn, ts)
 				if _itimediff(sn, kcp.rcv_nxt) >= 0 {
-					seg := kcp.newSegment(int(length))
+					var seg segment
 					seg.conv = conv
 					seg.cmd = cmd
 					seg.frg = frg
@@ -556,12 +582,11 @@ func (kcp *KCP) Input(data []byte, regular, ackNoDelay bool) int {
 					seg.ts = ts
 					seg.sn = sn
 					seg.una = una
-					copy(seg.data, data[:length])
-					kcp.parse_data(seg)
-				} else {
-					atomic.AddUint64(&DefaultSnmp.RepeatSegs, 1)
+					seg.data = data[:length] // delayed data copying
+					repeat = kcp.parse_data(seg)
 				}
-			} else {
+			}
+			if regular && repeat {
 				atomic.AddUint64(&DefaultSnmp.RepeatSegs, 1)
 			}
 		} else if cmd == IKCP_CMD_WASK {
@@ -579,32 +604,36 @@ func (kcp *KCP) Input(data []byte, regular, ackNoDelay bool) int {
 	}
 	atomic.AddUint64(&DefaultSnmp.InSegs, inSegs)
 
+	// update rtt with the latest ts
+	// ignore the FEC packet
 	if flag != 0 && regular {
-		kcp.parse_fastack(maxack)
 		current := currentMs()
-		if _itimediff(current, lastackts) >= 0 {
-			kcp.update_ack(_itimediff(current, lastackts))
+		if _itimediff(current, latest) >= 0 {
+			kcp.update_ack(_itimediff(current, latest))
 		}
 	}
 
-	if _itimediff(kcp.snd_una, snd_una) > 0 {
-		if kcp.cwnd < kcp.rmt_wnd {
-			mss := kcp.mss
-			if kcp.cwnd < kcp.ssthresh {
-				kcp.cwnd++
-				kcp.incr += mss
-			} else {
-				if kcp.incr < mss {
-					kcp.incr = mss
-				}
-				kcp.incr += (mss*mss)/kcp.incr + (mss / 16)
-				if (kcp.cwnd+1)*mss <= kcp.incr {
+	// cwnd update when packet arrived
+	if kcp.nocwnd == 0 {
+		if _itimediff(kcp.snd_una, snd_una) > 0 {
+			if kcp.cwnd < kcp.rmt_wnd {
+				mss := kcp.mss
+				if kcp.cwnd < kcp.ssthresh {
 					kcp.cwnd++
+					kcp.incr += mss
+				} else {
+					if kcp.incr < mss {
+						kcp.incr = mss
+					}
+					kcp.incr += (mss*mss)/kcp.incr + (mss / 16)
+					if (kcp.cwnd+1)*mss <= kcp.incr {
+						kcp.cwnd++
+					}
+				}
+				if kcp.cwnd > kcp.rmt_wnd {
+					kcp.cwnd = kcp.rmt_wnd
+					kcp.incr = kcp.rmt_wnd * mss
 				}
-			}
-			if kcp.cwnd > kcp.rmt_wnd {
-				kcp.cwnd = kcp.rmt_wnd
-				kcp.incr = kcp.rmt_wnd * mss
 			}
 		}
 	}
@@ -631,14 +660,28 @@ func (kcp *KCP) flush(ackOnly bool) uint32 {
 	seg.una = kcp.rcv_nxt
 
 	buffer := kcp.buffer
-	// flush acknowledges
-	ptr := buffer
-	for i, ack := range kcp.acklist {
+	ptr := buffer[kcp.reserved:] // keep n bytes untouched
+
+	// makeSpace makes room for writing
+	makeSpace := func(space int) {
 		size := len(buffer) - len(ptr)
-		if size+IKCP_OVERHEAD > int(kcp.mtu) {
+		if size+space > int(kcp.mtu) {
 			kcp.output(buffer, size)
-			ptr = buffer
+			ptr = buffer[kcp.reserved:]
 		}
+	}
+
+	// flush bytes in buffer if there is any
+	flushBuffer := func() {
+		size := len(buffer) - len(ptr)
+		if size > kcp.reserved {
+			kcp.output(buffer, size)
+		}
+	}
+
+	// flush acknowledges
+	for i, ack := range kcp.acklist {
+		makeSpace(IKCP_OVERHEAD)
 		// filter jitters caused by bufferbloat
 		if ack.sn >= kcp.rcv_nxt || len(kcp.acklist)-1 == i {
 			seg.sn, seg.ts = ack.sn, ack.ts
@@ -648,10 +691,7 @@ func (kcp *KCP) flush(ackOnly bool) uint32 {
 	kcp.acklist = kcp.acklist[0:0]
 
 	if ackOnly { // flash remain ack segments
-		size := len(buffer) - len(ptr)
-		if size > 0 {
-			kcp.output(buffer, size)
-		}
+		flushBuffer()
 		return kcp.interval
 	}
 
@@ -682,22 +722,14 @@ func (kcp *KCP) flush(ackOnly bool) uint32 {
 	// flush window probing commands
 	if (kcp.probe & IKCP_ASK_SEND) != 0 {
 		seg.cmd = IKCP_CMD_WASK
-		size := len(buffer) - len(ptr)
-		if size+IKCP_OVERHEAD > int(kcp.mtu) {
-			kcp.output(buffer, size)
-			ptr = buffer
-		}
+		makeSpace(IKCP_OVERHEAD)
 		ptr = seg.encode(ptr)
 	}
 
 	// flush window probing commands
 	if (kcp.probe & IKCP_ASK_TELL) != 0 {
 		seg.cmd = IKCP_CMD_WINS
-		size := len(buffer) - len(ptr)
-		if size+IKCP_OVERHEAD > int(kcp.mtu) {
-			kcp.output(buffer, size)
-			ptr = buffer
-		}
+		makeSpace(IKCP_OVERHEAD)
 		ptr = seg.encode(ptr)
 	}
 
@@ -722,7 +754,6 @@ func (kcp *KCP) flush(ackOnly bool) uint32 {
 		kcp.snd_buf = append(kcp.snd_buf, newseg)
 		kcp.snd_nxt++
 		newSegsCount++
-		kcp.snd_queue[k].data = nil
 	}
 	if newSegsCount > 0 {
 		kcp.snd_queue = kcp.remove_front(kcp.snd_queue, newSegsCount)
@@ -743,6 +774,9 @@ func (kcp *KCP) flush(ackOnly bool) uint32 {
 	for k := range ref {
 		segment := &ref[k]
 		needsend := false
+		if segment.acked == 1 {
+			continue
+		}
 		if segment.xmit == 0 { // initial transmit
 			needsend = true
 			segment.rto = kcp.rx_rto
@@ -774,20 +808,14 @@ func (kcp *KCP) flush(ackOnly bool) uint32 {
 		}
 
 		if needsend {
+			current = currentMs()
 			segment.xmit++
 			segment.ts = current
 			segment.wnd = seg.wnd
 			segment.una = seg.una
 
-			size := len(buffer) - len(ptr)
 			need := IKCP_OVERHEAD + len(segment.data)
-
-			if size+need > int(kcp.mtu) {
-				kcp.output(buffer, size)
-				current = currentMs() // time update for a blocking call
-				ptr = buffer
-			}
-
+			makeSpace(need)
 			ptr = segment.encode(ptr)
 			copy(ptr, segment.data)
 			ptr = ptr[len(segment.data):]
@@ -798,16 +826,13 @@ func (kcp *KCP) flush(ackOnly bool) uint32 {
 		}
 
 		// get the nearest rto
-		if rto := _itimediff(segment.rto, current); rto > 0 && rto < minrto {
+		if rto := _itimediff(segment.resendts, current); rto > 0 && rto < minrto {
 			minrto = rto
 		}
 	}
 
 	// flash remain segments
-	size := len(buffer) - len(ptr)
-	if size > 0 {
-		kcp.output(buffer, size)
-	}
+	flushBuffer()
 
 	// counter updates
 	sum := lostSegs
@@ -826,36 +851,41 @@ func (kcp *KCP) flush(ackOnly bool) uint32 {
 		atomic.AddUint64(&DefaultSnmp.RetransSegs, sum)
 	}
 
-	// update ssthresh
-	// rate halving, https://tools.ietf.org/html/rfc6937
-	if change > 0 {
-		inflight := kcp.snd_nxt - kcp.snd_una
-		kcp.ssthresh = inflight / 2
-		if kcp.ssthresh < IKCP_THRESH_MIN {
-			kcp.ssthresh = IKCP_THRESH_MIN
+	// cwnd update
+	if kcp.nocwnd == 0 {
+		// update ssthresh
+		// rate halving, https://tools.ietf.org/html/rfc6937
+		if change > 0 {
+			inflight := kcp.snd_nxt - kcp.snd_una
+			kcp.ssthresh = inflight / 2
+			if kcp.ssthresh < IKCP_THRESH_MIN {
+				kcp.ssthresh = IKCP_THRESH_MIN
+			}
+			kcp.cwnd = kcp.ssthresh + resent
+			kcp.incr = kcp.cwnd * kcp.mss
 		}
-		kcp.cwnd = kcp.ssthresh + resent
-		kcp.incr = kcp.cwnd * kcp.mss
-	}
 
-	// congestion control, https://tools.ietf.org/html/rfc5681
-	if lost > 0 {
-		kcp.ssthresh = cwnd / 2
-		if kcp.ssthresh < IKCP_THRESH_MIN {
-			kcp.ssthresh = IKCP_THRESH_MIN
+		// congestion control, https://tools.ietf.org/html/rfc5681
+		if lost > 0 {
+			kcp.ssthresh = cwnd / 2
+			if kcp.ssthresh < IKCP_THRESH_MIN {
+				kcp.ssthresh = IKCP_THRESH_MIN
+			}
+			kcp.cwnd = 1
+			kcp.incr = kcp.mss
 		}
-		kcp.cwnd = 1
-		kcp.incr = kcp.mss
-	}
 
-	if kcp.cwnd < 1 {
-		kcp.cwnd = 1
-		kcp.incr = kcp.mss
+		if kcp.cwnd < 1 {
+			kcp.cwnd = 1
+			kcp.incr = kcp.mss
+		}
 	}
 
 	return uint32(minrto)
 }
 
+// (deprecated)
+//
 // Update updates state (call it repeatedly, every 10ms-100ms), or you can ask
 // ikcp_check when to call it again (without ikcp_input/_send calling).
 // 'current' - current timestamp in millisec.
@@ -884,6 +914,8 @@ func (kcp *KCP) Update() {
 	}
 }
 
+// (deprecated)
+//
 // Check determines when should you invoke ikcp_update:
 // returns when you should invoke ikcp_update in millisec, if there
 // is no ikcp_input/_send calling. you can call ikcp_update in that
@@ -939,12 +971,16 @@ func (kcp *KCP) SetMtu(mtu int) int {
 	if mtu < 50 || mtu < IKCP_OVERHEAD {
 		return -1
 	}
-	buffer := make([]byte, (mtu+IKCP_OVERHEAD)*3)
+	if kcp.reserved >= int(kcp.mtu-IKCP_OVERHEAD) || kcp.reserved < 0 {
+		return -1
+	}
+
+	buffer := make([]byte, mtu)
 	if buffer == nil {
 		return -2
 	}
 	kcp.mtu = uint32(mtu)
-	kcp.mss = kcp.mtu - IKCP_OVERHEAD
+	kcp.mss = kcp.mtu - IKCP_OVERHEAD - uint32(kcp.reserved)
 	kcp.buffer = buffer
 	return 0
 }
@@ -998,11 +1034,13 @@ func (kcp *KCP) WaitSnd() int {
 }
 
 // remove front n elements from queue
+// if the number of elements to remove is more than half of the size.
+// just shift the rear elements to front, otherwise just reslice q to q[n:]
+// then the cost of runtime.growslice can always be less than n/2
 func (kcp *KCP) remove_front(q []segment, n int) []segment {
-	newn := copy(q, q[n:])
-	gc := q[newn:]
-	for k := range gc {
-		gc[k].data = nil // de-ref data
+	if n > cap(q)/2 {
+		newn := copy(q, q[n:])
+		return q[:newn]
 	}
-	return q[:newn]
+	return q[n:]
 }

kcp-go/readloop_generic.go

@@ -0,0 +1,50 @@
+// +build !linux
+
+package kcp
+
+import (
+	"sync/atomic"
+
+	"github.com/pkg/errors"
+)
+
+func (s *UDPSession) readLoop() {
+	buf := make([]byte, mtuLimit)
+	var src string
+	for {
+		if n, addr, err := s.conn.ReadFrom(buf); err == nil {
+			// make sure the packet is from the same source
+			if src == "" { // set source address
+				src = addr.String()
+			} else if addr.String() != src {
+				atomic.AddUint64(&DefaultSnmp.InErrs, 1)
+				continue
+			}
+
+			if n >= s.headerSize+IKCP_OVERHEAD {
+				s.packetInput(buf[:n])
+			} else {
+				atomic.AddUint64(&DefaultSnmp.InErrs, 1)
+			}
+		} else {
+			s.notifyReadError(errors.WithStack(err))
+			return
+		}
+	}
+}
+
+func (l *Listener) monitor() {
+	buf := make([]byte, mtuLimit)
+	for {
+		if n, from, err := l.conn.ReadFrom(buf); err == nil {
+			if n >= l.headerSize+IKCP_OVERHEAD {
+				l.packetInput(buf[:n], from)
+			} else {
+				atomic.AddUint64(&DefaultSnmp.InErrs, 1)
+			}
+		} else {
+			l.notifyReadError(errors.WithStack(err))
+			return
+		}
+	}
+}

kcp-go/readloop_linux.go

@@ -0,0 +1,80 @@
+// +build linux
+
+package kcp
+
+import (
+	"net"
+	"sync/atomic"
+
+	"github.com/pkg/errors"
+	"golang.org/x/net/ipv4"
+	"golang.org/x/net/ipv6"
+)
+
+// the read loop for a client session
+func (s *UDPSession) readLoop() {
+	var src string
+	msgs := make([]ipv4.Message, batchSize)
+	for k := range msgs {
+		msgs[k].Buffers = [][]byte{make([]byte, mtuLimit)}
+	}
+
+	for {
+		if count, err := s.xconn.ReadBatch(msgs, 0); err == nil {
+			for i := 0; i < count; i++ {
+				msg := &msgs[i]
+				// make sure the packet is from the same source
+				if src == "" { // set source address if nil
+					src = msg.Addr.String()
+					s.remote = msg.Addr// added by hikari
+				} else if msg.Addr.String() != src {
+					atomic.AddUint64(&DefaultSnmp.InErrs, 1)
+					continue
+				}
+
+				if msg.N < s.headerSize+IKCP_OVERHEAD {
+					atomic.AddUint64(&DefaultSnmp.InErrs, 1)
+					continue
+				}
+
+				// source and size has validated
+				s.packetInput(msg.Buffers[0][:msg.N])
+			}
+		} else {
+			s.notifyReadError(errors.WithStack(err))
+			return
+		}
+	}
+}
+
+// monitor incoming data for all connections of server
+func (l *Listener) monitor() {
+	addr, _ := net.ResolveUDPAddr("udp", l.conn.LocalAddr().String())
+	var xconn batchConn
+	if addr.IP.To4() != nil {
+		xconn = ipv4.NewPacketConn(l.conn)
+	} else {
+		xconn = ipv6.NewPacketConn(l.conn)
+	}
+
+	msgs := make([]ipv4.Message, batchSize)
+	for k := range msgs {
+		msgs[k].Buffers = [][]byte{make([]byte, mtuLimit)}
+	}
+
+	for {
+		if count, err := xconn.ReadBatch(msgs, 0); err == nil {
+			for i := 0; i < count; i++ {
+				msg := &msgs[i]
+				if msg.N >= l.headerSize+IKCP_OVERHEAD {
+					l.packetInput(msg.Buffers[0][:msg.N], msg.Addr)
+				} else {
+					atomic.AddUint64(&DefaultSnmp.InErrs, 1)
+				}
+			}
+		} else {
+			l.notifyReadError(errors.WithStack(err))
+			return
+		}
+	}
+}

kcp-go/tx_generic.go

@@ -0,0 +1,27 @@
+// +build !linux
+
+package kcp
+
+import (
+	"sync/atomic"
+
+	"github.com/pkg/errors"
+	"golang.org/x/net/ipv4"
+)
+
+func (s *UDPSession) tx(txqueue []ipv4.Message) {
+	nbytes := 0
+	npkts := 0
+	for k := range txqueue {
+		if n, err := s.conn.WriteTo(txqueue[k].Buffers[0], txqueue[k].Addr); err == nil {
+			nbytes += n
+			npkts++
+			xmitBuf.Put(txqueue[k].Buffers[0])
+		} else {
+			s.notifyWriteError(errors.WithStack(err))
+			break
+		}
+	}
+	atomic.AddUint64(&DefaultSnmp.OutPkts, uint64(npkts))
+	atomic.AddUint64(&DefaultSnmp.OutBytes, uint64(nbytes))
+}

kcp-go/tx_linux.go

@@ -0,0 +1,31 @@
+// +build linux
+
+package kcp
+
+import (
+	"sync/atomic"
+
+	"github.com/pkg/errors"
+	"golang.org/x/net/ipv4"
+)
+
+func (s *UDPSession) tx(txqueue []ipv4.Message) {
+	nbytes := 0
+	npkts := 0
+	for len(txqueue) > 0 {
+		if n, err := s.xconn.WriteBatch(txqueue, 0); err == nil {
+			for k := range txqueue[:n] {
+				nbytes += len(txqueue[k].Buffers[0])
+				xmitBuf.Put(txqueue[k].Buffers[0])
+			}
+			npkts += n
+			txqueue = txqueue[n:]
+		} else {
+			s.notifyWriteError(errors.WithStack(err))
+			break
+		}
+	}
+
+	atomic.AddUint64(&DefaultSnmp.OutPkts, uint64(npkts))
+	atomic.AddUint64(&DefaultSnmp.OutBytes, uint64(nbytes))
+}

kcp-go/updater.go

@@ -76,10 +76,10 @@ func (h *updateHeap) wakeup() {
 }
 
 func (h *updateHeap) updateTask() {
-	var timer <-chan time.Time
+	timer := time.NewTimer(0)
 	for {
 		select {
-		case <-timer:
+		case <-timer.C:
 		case <-h.chWakeUp:
 		}
 
@@ -87,7 +87,7 @@ func (h *updateHeap) updateTask() {
 		hlen := h.Len()
 		for i := 0; i < hlen; i++ {
 			entry := &h.entries[0]
-			if time.Now().After(entry.ts) {
+			if !time.Now().Before(entry.ts) {
 				interval := entry.s.update()
 				entry.ts = time.Now().Add(interval)
 				heap.Fix(h, 0)
@@ -97,7 +97,7 @@ func (h *updateHeap) updateTask() {
 		}
 
 		if hlen > 0 {
-			timer = time.After(h.entries[0].ts.Sub(time.Now()))
+			timer.Reset(h.entries[0].ts.Sub(time.Now()))
 		}
 		h.mu.Unlock()
 	}

p2pclient/config.go

@@ -45,3 +45,4 @@ func parseJSONConfig(config *Config, path string) error {
 
 	return json.NewDecoder(file).Decode(config)
 }
+

p2pclient/main.go

@@ -3,24 +3,20 @@ package main
 import (
 	"bufio"
 	"crypto/sha1"
-	"encoding/csv"
 	"encoding/json"
-	"fmt"
+	"github.com/hikaricai/p2p_tun/kcp-go"
+	"github.com/xtaci/kcptun/generic"
 	"io"
 	"log"
 	"math/rand"
 	"net"
 	"os"
+	"sync"
 	"sync/atomic"
 	"time"
-
 	"golang.org/x/crypto/pbkdf2"
-
-	"github.com/hikaricai/p2p_tun/kcp-go"
 	"github.com/urfave/cli"
 	"github.com/xtaci/smux"
-
-	"path/filepath"
 )
 
 var (
@@ -29,6 +25,8 @@ var (
 	// SALT is use for pbkdf2 key expansion
 	SALT = "kcp-go"
 	isServer = false
+	pingInterval = 10
+	xmitBuf sync.Pool
 )
 
 func handleLocalTcp(sess *smux.Session, p1 io.ReadWriteCloser, quiet bool) {
@@ -36,7 +34,6 @@ func handleLocalTcp(sess *smux.Session, p1 io.ReadWriteCloser, quiet bool) {
 		log.Println("stream opened")
 		defer log.Println("stream closed")
 	}
-
 	defer p1.Close()
 	p2, err := sess.OpenStream()
 	if err != nil {
@@ -44,19 +41,20 @@ func handleLocalTcp(sess *smux.Session, p1 io.ReadWriteCloser, quiet bool) {
 	}
 	defer p2.Close()
 
-	// start tunnel
-	p1die := make(chan struct{})
-	buf1 := make([]byte, 65535)
-	go func() { io.CopyBuffer(p1, p2, buf1); close(p1die) }()
+	streamCopy := func(dst io.Writer, src io.ReadCloser) chan struct{} {
+		die := make(chan struct{})
+		go func() {
+			buf := xmitBuf.Get().([]byte)
+			generic.CopyBuffer(dst, src, buf)
+			xmitBuf.Put(buf)
+			close(die)
+		}()
+		return die
+	}
 
-	p2die := make(chan struct{})
-	buf2 := make([]byte, 65535)
-	go func() { io.CopyBuffer(p2, p1, buf2); close(p2die) }()
-
-	// wait for tunnel termination
 	select {
-	case <-p1die:
-	case <-p2die:
+	case <-streamCopy(p1, p2):
+	case <-streamCopy(p2, p1):
 	}
 }
 
@@ -73,6 +71,9 @@ func main() {
 		// add more log flags for debugging
 		log.SetFlags(log.LstdFlags | log.Lshortfile)
 	}
+	xmitBuf.New = func() interface{} {
+		return make([]byte, 32768)
+	}
 	myApp := cli.NewApp()
 	myApp.Name = "p2pclient"
 	myApp.Usage = "client(with kcptun)"
@@ -85,19 +86,14 @@ func main() {
 		},
 		cli.StringFlag{
 			Name:  "listentcp,l",
-			Value: ":12948",
+			Value: ":2022",
 			Usage: "local listen address",
 		},
 		cli.StringFlag{
 			Name:  "remoteudp, r",
-			Value: "vps:29900",
+			Value: "127.0.0.1:4000",
 			Usage: "kcp server address",
 		},
-		cli.StringFlag{
-			Name:  "bindudp, b",
-			Value: ":29900",
-			Usage: "bind local udp",
-		},
 		cli.StringFlag{
 			Name:  "key, k",
 			Value: "1234",
@@ -111,12 +107,12 @@ func main() {
 		},
 		cli.StringFlag{
 			Name:  "crypt",
-			Value: "aes",
+			Value: "none",
 			Usage: "aes, aes-128, aes-192, salsa20, blowfish, twofish, cast5, 3des, tea, xtea, xor, sm4, none",
 		},
 		cli.StringFlag{
 			Name:  "mode",
-			Value: "fast2",
+			Value: "fast",
 			Usage: "profiles: fast3, fast2, fast, normal, manual",
 		},
 		cli.IntFlag{
@@ -146,12 +142,12 @@ func main() {
 		},
 		cli.IntFlag{
 			Name:  "datashard,ds",
-			Value: 10,
+			Value: 0,
 			Usage: "set reed-solomon erasure coding - datashard",
 		},
 		cli.IntFlag{
 			Name:  "parityshard,ps",
-			Value: 3,
+			Value: 0,
 			Usage: "set reed-solomon erasure coding - parityshard",
 		},
 		cli.IntFlag{
@@ -228,12 +224,10 @@ func main() {
 		config.ListenTcp = c.String("listentcp")
 		config.RemoteUdp = c.String("remoteudp")
 		config.TargetTcp = c.String("targettcp")
-		config.BindUdp = c.String("bindudp")
 		config.Key		= c.String("key")
 		config.Passwd = c.String("passwd")
 		config.Crypt = c.String("crypt")
 		config.Mode = c.String("mode")
-		config.AutoExpire = c.Int("autoexpire")
 		config.MTU = c.Int("mtu")
 		config.SndWnd = c.Int("sndwnd")
 		config.RcvWnd = c.Int("rcvwnd")
@@ -248,9 +242,6 @@ func main() {
 		config.NoCongestion = c.Int("nc")
 		config.SockBuf = c.Int("sockbuf")
 		config.KeepAlive = c.Int("keepalive")
-		config.Log = c.String("log")
-		config.SnmpLog = c.String("snmplog")
-		config.SnmpPeriod = c.Int("snmpperiod")
 		config.Quiet = c.Bool("quiet")
 
 		if c.String("c") != "" {
@@ -258,14 +249,6 @@ func main() {
 			checkError(err)
 		}
 
-		// log redirect
-		if config.Log != "" {
-			f, err := os.OpenFile(config.Log, os.O_RDWR|os.O_CREATE|os.O_APPEND, 0666)
-			checkError(err)
-			defer f.Close()
-			log.SetOutput(f)
-		}
-
 		switch config.Mode {
 		case "normal":
 			config.NoDelay, config.Interval, config.Resend, config.NoCongestion = 0, 40, 2, 1
@@ -279,30 +262,16 @@ func main() {
 
 		log.Println("version:", VERSION)
 
-		log.Println("encryption:", config.Crypt)
-		log.Println("nodelay parameters:", config.NoDelay, config.Interval, config.Resend, config.NoCongestion)
-		log.Println("remote udp  address:", config.RemoteUdp)
-		log.Println("sndwnd:", config.SndWnd, "rcvwnd:", config.RcvWnd)
-		log.Println("compression:", !config.NoComp)
-		log.Println("mtu:", config.MTU)
-		log.Println("datashard:", config.DataShard, "parityshard:", config.ParityShard)
-		log.Println("acknodelay:", config.AckNodelay)
-		log.Println("dscp:", config.DSCP)
-		log.Println("sockbuf:", config.SockBuf)
-		log.Println("keepalive:", config.KeepAlive)
-		log.Println("autoexpire:", config.AutoExpire)
-		log.Println("snmplog:", config.SnmpLog)
-		log.Println("snmpperiod:", config.SnmpPeriod)
-		log.Println("quiet:", config.Quiet)
-		go snmpLogger(config.SnmpLog, config.SnmpPeriod)
-
 		chTCPConn := make(chan *net.TCPConn, 16)
 
 		go tcpListener(chTCPConn, &config)
 		for{
 			peerAddr, err := getPeerAddr(&config)
+			time.Sleep(2*time.Second)
 			if err == nil{
 				p2pHandle(&config, peerAddr, chTCPConn)
+			}else{
+				time.Sleep(time.Duration(pingInterval)*time.Second)
 			}
 		}
 	}
@@ -310,9 +279,9 @@ func main() {
 }
 
 func p2pHandle(config *Config, peerAddr string, chTCPConn chan *net.TCPConn){
-	udpAddr, err := net.ResolveUDPAddr("udp", config.BindUdp)
+	udpAddr, err := net.ResolveUDPAddr("udp4", config.BindUdp)
 	checkError(err)
-	udpconn, err := net.ListenUDP("udp", udpAddr)
+	udpconn, err := net.ListenUDP("udp4", udpAddr)
 	checkError(err)
 	defer udpconn.Close()
 
@@ -336,9 +305,9 @@ func p2pHandle(config *Config, peerAddr string, chTCPConn chan *net.TCPConn){
 }
 
 func tcpListener(chTCPConn chan *net.TCPConn, config *Config){
-	listenTcpAddr, err := net.ResolveTCPAddr("tcp", config.ListenTcp)
+	listenTcpAddr, err := net.ResolveTCPAddr("tcp4", config.ListenTcp)
 	checkError(err)
-	listener, err := net.ListenTCP("tcp", listenTcpAddr)
+	listener, err := net.ListenTCP("tcp4", listenTcpAddr)
 	checkError(err)
 	log.Println("listening on:", listener.Addr())
 	for{
@@ -352,11 +321,12 @@ func tcpListener(chTCPConn chan *net.TCPConn, config *Config){
 }
 
 func getPeerAddr(config *Config)(string, error){
-
-	udpAddr, err := net.ResolveUDPAddr("udp", config.BindUdp)
+	udpAddr, err := net.ResolveUDPAddr("udp4", "0.0.0.0:0")
 	checkError(err)
-	udpconn, err := net.ListenUDP("udp", udpAddr)
+	udpconn, err := net.ListenUDP("udp4", udpAddr)
 	checkError(err)
+	config.BindUdp = udpconn.LocalAddr().String()
+	log.Println("config.BindUdp is ", config.BindUdp)
 	defer udpconn.Close()
 
 	kcpConn, err := newKcpConn(udpconn, config, config.RemoteUdp)
@@ -372,14 +342,12 @@ func getPeerAddr(config *Config)(string, error){
 	finMess := phaseJsonMess("fin", "good bye")
 	var peerAddr string
 	log.Println("writing mess")
-	n, err := kcpConn.Write(pairMess)
+	_, err = kcpConn.Write(pairMess)
 	if err != nil {
 		log.Println("kcpConn.Write", err)
 		return "", err
 	}
-	log.Println("writen ", n)
 	for {
-		pair_s := false
 		log.Println("waiting for server")
 		line, err := reader.ReadString('\n')
 		if err != nil {
@@ -394,40 +362,34 @@ func getPeerAddr(config *Config)(string, error){
 			atomic.StoreInt32(&dataReady, 1)
 			log.Println("rcv ping")
 		case "pair_s":
-			pair_s = true
-			fallthrough
-		case "pair_c":
-			isServer = pair_s
+			isServer = true
 			peerAddr = mess.Data
 			log.Println("peer addr is ", peerAddr)
-			n, err := kcpConn.Write(finMess)
-			if err != nil {
-				log.Println("kcpConn.Write", err)
-				return "", err
-			}
-			log.Println("writen ", n)
-			time.Sleep(1*time.Second)
+			kcpConn.Write(finMess)
+		case "pair_c":
+			isServer = false
+			peerAddr = mess.Data
+			log.Println("peer addr is ", peerAddr)
+			kcpConn.Write(finMess)
+		case "fin":
 			return peerAddr, nil
 		}
 	}
 }
 
 func pingCheck(conn *kcp.UDPSession, dataReady *int32, chPing chan struct {}){
-	tickerDie := time.NewTicker(30*time.Second)
-	defer tickerDie.Stop()
 	jsonPingMess := phaseJsonMess("ping", "hello")
-	tickerPing := time.NewTicker(10 * time.Second)
+	tickerPing := time.NewTicker(time.Duration(pingInterval)*time.Second)
 	defer tickerPing.Stop()
 	defer 	log.Println("pingCheck return")
 	for {
 		select {
-		case <-tickerDie.C:
+		case <-tickerPing.C:
 			if !atomic.CompareAndSwapInt32(dataReady, 1, 0) {
 				log.Println("ping timeout")
 				conn.Close()
 				return
 			}
-		case <-tickerPing.C:
 			conn.Write(jsonPingMess)
 		case <- chPing:
 			return
@@ -450,7 +412,6 @@ func phaseJsonMess(cmd string, data string) []byte {
 }
 
 func newSmuxSession(udpconn net.PacketConn, config *Config, remoteAddr string) (*smux.Session, error) {
-
 	kcpconn, err := newKcpConn(udpconn, config, remoteAddr)
 	if err != nil {
 		return nil, err
@@ -458,9 +419,9 @@ func newSmuxSession(udpconn net.PacketConn, config *Config, remoteAddr string) (
 	smuxConfig := smux.DefaultConfig()
 	smuxConfig.MaxReceiveBuffer = config.SockBuf
 	smuxConfig.KeepAliveInterval = time.Duration(config.KeepAlive) * time.Second
+	smuxConfig.KeepAliveTimeout = time.Duration(config.KeepAlive) * time.Second*3
 	// stream multiplex
 	var smuxSession *smux.Session
-
 	if isServer {
 		smuxSession, err = smux.Server(kcpconn, smuxConfig)
 	} else {
@@ -551,54 +512,22 @@ func handleTargetTcp(addr string, session *smux.Session, quiet bool) {
 			defer p1.Close()
 			defer p2.Close()
 
-			// start tunnel
-			p1die := make(chan struct{})
-			buf1 := make([]byte, 65535)
-			go func() { io.CopyBuffer(p1, p2, buf1); close(p1die) }()
+			streamCopy := func(dst io.Writer, src io.ReadCloser) chan struct{} {
+				die := make(chan struct{})
+				go func() {
+					buf := xmitBuf.Get().([]byte)
+					generic.CopyBuffer(dst, src, buf)
+					xmitBuf.Put(buf)
+					close(die)
+				}()
+				return die
+			}
 
-			p2die := make(chan struct{})
-			buf2 := make([]byte, 65535)
-			go func() { io.CopyBuffer(p2, p1, buf2); close(p2die) }()
-
-			// wait for tunnel termination
 			select {
-			case <-p1die:
-			case <-p2die:
+			case <-streamCopy(p1, p2):
+			case <-streamCopy(p2, p1):
 			}
 		}()
 	}
 }
 
-func snmpLogger(path string, interval int) {
-	if path == "" || interval == 0 {
-		return
-	}
-	ticker := time.NewTicker(time.Duration(interval) * time.Second)
-	defer ticker.Stop()
-	for {
-		select {
-		case <-ticker.C:
-			// split path into dirname and filename
-			logdir, logfile := filepath.Split(path)
-			// only format logfile
-			f, err := os.OpenFile(logdir+time.Now().Format(logfile), os.O_RDWR|os.O_CREATE|os.O_APPEND, 0666)
-			if err != nil {
-				log.Println(err)
-				return
-			}
-			w := csv.NewWriter(f)
-			// write header in empty file
-			if stat, err := f.Stat(); err == nil && stat.Size() == 0 {
-				if err := w.Write(append([]string{"Unix"}, kcp.DefaultSnmp.Header()...)); err != nil {
-					log.Println(err)
-				}
-			}
-			if err := w.Write(append([]string{fmt.Sprint(time.Now().Unix())}, kcp.DefaultSnmp.ToSlice()...)); err != nil {
-				log.Println(err)
-			}
-			kcp.DefaultSnmp.Reset()
-			w.Flush()
-			f.Close()
-		}
-	}
-}

p2pserver/config.go

@@ -8,7 +8,7 @@ import (
 // Config for server
 type Config struct {
 	Listen       string `json:"listen"`
-	Key          string `json:"key"`
+	Passwd       string `json:"passwd"`
 	Crypt        string `json:"crypt"`
 	Mode         string `json:"mode"`
 	MTU          int    `json:"mtu"`

p2pserver/main.go

@@ -3,13 +3,10 @@ package main
 import (
 	"bufio"
 	"crypto/sha1"
-	"encoding/csv"
 	"encoding/json"
-	"fmt"
+	"github.com/hikaricai/p2p_tun/kcp-go"
 	"log"
 	"math/rand"
-	"net/http"
-	_ "net/http/pprof"
 	"os"
 	"sync"
 	"sync/atomic"
@@ -17,9 +14,6 @@ import (
 
 	"golang.org/x/crypto/pbkdf2"
 
-	"path/filepath"
-
-	"github.com/hikaricai/p2p_tun/kcp-go"
 	"github.com/urfave/cli"
 )
 
@@ -28,6 +22,7 @@ var (
 	VERSION = "SELFBUILD"
 	// SALT is use for pbkdf2 key expansion
 	SALT = "kcp-go"
+	pingInterval = 10
 )
 
 func checkError(err error) {
@@ -50,23 +45,23 @@ func main() {
 	myApp.Flags = []cli.Flag{
 		cli.StringFlag{
 			Name:  "listen,l",
-			Value: ":4000",
+			Value: "0.0.0.0:4000",
 			Usage: "kcp server listen address",
 		},
 		cli.StringFlag{
-			Name:   "key",
+			Name:   "passwd",
 			Value:  "1234",
 			Usage:  "pre-shared secret between client and server",
 			EnvVar: "KCPTUN_KEY",
 		},
 		cli.StringFlag{
 			Name:  "crypt",
-			Value: "aes",
+			Value: "none",
 			Usage: "aes, aes-128, aes-192, salsa20, blowfish, twofish, cast5, 3des, tea, xtea, xor, sm4, none",
 		},
 		cli.StringFlag{
 			Name:  "mode",
-			Value: "fast2",
+			Value: "fast",
 			Usage: "profiles: fast3, fast2, fast, normal, manual",
 		},
 		cli.IntFlag{
@@ -86,12 +81,12 @@ func main() {
 		},
 		cli.IntFlag{
 			Name:  "datashard,ds",
-			Value: 10,
+			Value: 0,
 			Usage: "set reed-solomon erasure coding - datashard",
 		},
 		cli.IntFlag{
 			Name:  "parityshard,ps",
-			Value: 3,
+			Value: 0,
 			Usage: "set reed-solomon erasure coding - parityshard",
 		},
 		cli.IntFlag{
@@ -170,7 +165,7 @@ func main() {
 	myApp.Action = func(c *cli.Context) error {
 		config := Config{}
 		config.Listen = c.String("listen")
-		config.Key = c.String("key")
+		config.Passwd = c.String("passwd")
 		config.Crypt = c.String("crypt")
 		config.Mode = c.String("mode")
 		config.MTU = c.Int("mtu")
@@ -187,7 +182,6 @@ func main() {
 		config.NoCongestion = c.Int("nc")
 		config.SockBuf = c.Int("sockbuf")
 		config.KeepAlive = c.Int("keepalive")
-		config.Log = c.String("log")
 		config.SnmpLog = c.String("snmplog")
 		config.SnmpPeriod = c.Int("snmpperiod")
 		config.Pprof = c.Bool("pprof")
@@ -198,15 +192,6 @@ func main() {
 			err := parseJSONConfig(&config, c.String("c"))
 			checkError(err)
 		}
-
-		// log redirect
-		if config.Log != "" {
-			f, err := os.OpenFile(config.Log, os.O_RDWR|os.O_CREATE|os.O_APPEND, 0666)
-			checkError(err)
-			defer f.Close()
-			log.SetOutput(f)
-		}
-
 		switch config.Mode {
 		case "normal":
 			config.NoDelay, config.Interval, config.Resend, config.NoCongestion = 0, 40, 2, 1
@@ -220,7 +205,7 @@ func main() {
 
 		log.Println("version:", VERSION)
 		log.Println("initiating key derivation")
-		pass := pbkdf2.Key([]byte(config.Key), []byte(SALT), 4096, 32, sha1.New)
+		pass := pbkdf2.Key([]byte(config.Passwd), []byte(SALT), 4096, 32, sha1.New)
 		var block kcp.BlockCrypt
 		switch config.Crypt {
 		case "sm4":
@@ -255,20 +240,6 @@ func main() {
 		lis, err := kcp.ListenWithOptions(config.Listen, block, config.DataShard, config.ParityShard)
 		checkError(err)
 		log.Println("listening on:", lis.Addr())
-		log.Println("encryption:", config.Crypt)
-		log.Println("nodelay parameters:", config.NoDelay, config.Interval, config.Resend, config.NoCongestion)
-		log.Println("sndwnd:", config.SndWnd, "rcvwnd:", config.RcvWnd)
-		log.Println("compression:", !config.NoComp)
-		log.Println("mtu:", config.MTU)
-		log.Println("datashard:", config.DataShard, "parityshard:", config.ParityShard)
-		log.Println("acknodelay:", config.AckNodelay)
-		log.Println("dscp:", config.DSCP)
-		log.Println("sockbuf:", config.SockBuf)
-		log.Println("keepalive:", config.KeepAlive)
-		log.Println("snmplog:", config.SnmpLog)
-		log.Println("snmpperiod:", config.SnmpPeriod)
-		log.Println("pprof:", config.Pprof)
-		log.Println("quiet:", config.Quiet)
 
 		if err := lis.SetDSCP(config.DSCP); err != nil {
 			log.Println("SetDSCP:", err)
@@ -280,11 +251,6 @@ func main() {
 			log.Println("SetWriteBuffer:", err)
 		}
 
-		go snmpLogger(config.SnmpLog, config.SnmpPeriod)
-		if config.Pprof {
-			go http.ListenAndServe(":6060", nil)
-		}
-
 		for {
 			log.Println("listening new kcp")
 			if conn, err := lis.AcceptKCP(); err == nil {
@@ -310,22 +276,54 @@ type DigHoleMess struct {
 	Data string
 }
 
-type Peer struct {
-	addr   string
-	chPair chan string
+type UdpSess struct {
+	conn *kcp.UDPSession
+	pingFlag int32
+	isFin bool
+}
+type P2PSession struct {
+	key string
+	sess1 *UdpSess
+	sess2 *UdpSess
+	chFin chan struct{}
+	mu  sync.Mutex
 }
 
-var keymap = make(map[string]*Peer)
+var keymap = make(map[string]*P2PSession)
 
 var keymu  sync.Mutex
-
+func registSession(session *P2PSession, sess *UdpSess) (ret bool){
+	ret = true
+	session.mu.Lock()
+	if session.sess1 == nil{
+		session.sess1 = sess
+	}else if session.sess2 == nil{
+		session.sess2 = sess
+	}else{
+		ret = false
+	}
+	if  ret == true && session.sess1 != nil && session.sess2 != nil{
+		conn1 := session.sess1.conn
+		conn2 := session.sess2.conn
+		addr1 := conn1.RemoteAddr().String()
+		addr2 := conn2.RemoteAddr().String()
+		jsonPair1Mess := phaseJsonMess("pair_s", addr2)
+		jsonPair2Mess := phaseJsonMess("pair_c", addr1)
+		log.Println("pairing ", addr1,addr2)
+		conn1.Write(jsonPair1Mess)
+		conn2.Write(jsonPair2Mess)
+	}
+	session.mu.Unlock()
+	return ret
+}
 func handleClient(conn *kcp.UDPSession) {
+	var session *P2PSession;
+	key := ""
 	reader := bufio.NewReader(conn)
 	defer conn.Close()
-	var dataReady int32
-	var chThreadDie chan struct{}= make(chan struct{})
-	defer close(chThreadDie)
-	go timeout(conn, &dataReady, chThreadDie)
+	var ok bool
+	var registed  = false
+	sess := &UdpSess{conn,1,false}
 	for {
 		line, err := reader.ReadString('\n')
 		if err != nil {
@@ -333,71 +331,100 @@ func handleClient(conn *kcp.UDPSession) {
 			return
 		}
 		var mess DigHoleMess
-		json.Unmarshal([]byte(line), &mess)
+		err = json.Unmarshal([]byte(line), &mess)
+		if err != nil{
+			continue
+		}
 		switch mess.Cmd {
 		case "login":
 			remoteAddr := conn.RemoteAddr().String()
 			log.Println("login from ", remoteAddr)
-			key := mess.Data
-
+			key = mess.Data
 			log.Println("key is ", key)
 			keymu.Lock()
-			peer, ok := keymap[key]
+			session, ok = keymap[key]
 			if ok {
-				peerAddr := peer.addr
-				log.Println("find peer and addr is", peerAddr)
-				delete(keymap, key)
-				keymu.Unlock()
-				peer.chPair <- remoteAddr
-				jsonPairMess := phaseJsonMess("pair_c", peerAddr)
-				conn.Write(jsonPairMess)
+				registed = registSession(session,sess)
 			} else {
-				log.Println("no peer, registed")
-				peer := Peer{remoteAddr, make(chan string)}
-				keymap[key] = &peer
-				keymu.Unlock()
-				go notifyAddr(conn, &peer, chThreadDie)
+				log.Println("make new P2PSession")
+				session = new(P2PSession)
+				session.key = key
+				session.sess1 = sess
+				session.chFin = make(chan struct{})
+				keymap[key] = session
+				registed = true
+				go p2pSessionHandler(session)
 			}
-		case "ping":
-			atomic.StoreInt32(&dataReady, 1)
+			keymu.Unlock()
 			jsonPingMess := phaseJsonMess("ping", "hello")
 			conn.Write(jsonPingMess)
+		case "ping":
 			log.Println("rcv ping from ", conn.RemoteAddr().String())
+			if registed == false{
+				keymu.Lock()
+				session, ok = keymap[key]
+				if ok {
+					registed = registSession(session,sess)
+				} else {
+					log.Println("make new P2PSession")
+					session = new(P2PSession)
+					session.key = key
+					session.sess1 = sess
+					session.chFin = make(chan struct{})
+					keymap[key] = session
+					registed = true
+					go p2pSessionHandler(session)
+				}
+				keymu.Unlock()
+			}
+			atomic.StoreInt32(&sess.pingFlag, 1)
+			jsonPingMess := phaseJsonMess("ping", "hello")
+			conn.Write(jsonPingMess)
 		case "fin":
 			log.Println("fin from", conn.RemoteAddr().String())
-			time.Sleep(time.Second)
-			return
+			sess.isFin = true
+			session.chFin <- struct{}{}
 		}
 	}
 }
 
-func timeout(conn *kcp.UDPSession, dataReady *int32, chThreadDie chan struct{}){
-	tickerDie := time.NewTicker(30*time.Second)
-	defer tickerDie.Stop()
 
+func p2pSessionHandler(session *P2PSession){
+	tickerPing := time.NewTicker(time.Duration(pingInterval)*time.Second)
+	defer tickerPing.Stop()
 	for {
 		select {
-		case <-tickerDie.C:
-			if !atomic.CompareAndSwapInt32(dataReady, 1, 0) {
-				log.Println("ping timeout")
-				conn.Close()
+		case <-tickerPing.C:
+			if session.sess1 != nil && !atomic.CompareAndSwapInt32(&session.sess1.pingFlag, 1, 0) {
+				session.mu.Lock()
+				session.sess1.conn.Close()
+				session.sess1 = nil
+				log.Println("sess1 ping timeout")
+				session.mu.Unlock()
+			}
+			if session.sess2 != nil && !atomic.CompareAndSwapInt32(&session.sess2.pingFlag, 1, 0) {
+				session.mu.Lock()
+				session.sess2.conn.Close()
+				session.sess2 = nil
+				log.Println("sess2 ping timeout")
+				session.mu.Unlock()
+			}
+		case <-session.chFin:
+			if session.sess1 == nil || session.sess2 == nil{
+				continue
+			}
+			if session.sess1.isFin == true && session.sess2.isFin == true{
+				jsonFinMess := phaseJsonMess("fin", "bye")
+				session.sess1.conn.Write(jsonFinMess)
+				session.sess2.conn.Write(jsonFinMess)
+				time.Sleep(time.Second)
+				session.sess1.conn.Close()
+				session.sess2.conn.Close()
+				keymu.Lock()
+				delete(keymap,session.key)
+				keymu.Unlock()
 				return
 			}
-		case <-chThreadDie:
-			return
-		}
-	}
-}
-
-func notifyAddr(conn *kcp.UDPSession, peer *Peer, chThreadDie chan struct{}){
-	for {
-		select {
-		case <-chThreadDie:
-			return
-		case peerAddr := <-peer.chPair:
-			jsonPairMess := phaseJsonMess("pair_s", peerAddr)
-			conn.Write(jsonPairMess)
-			return
 		}
 	}
 }
@@ -410,37 +437,3 @@ func phaseJsonMess(cmd string, data string) []byte {
 	}
 	return append(jsonMess, '\n')
 }
-
-func snmpLogger(path string, interval int) {
-	if path == "" || interval == 0 {
-		return
-	}
-	ticker := time.NewTicker(time.Duration(interval) * time.Second)
-	defer ticker.Stop()
-	for {
-		select {
-		case <-ticker.C:
-			// split path into dirname and filename
-			logdir, logfile := filepath.Split(path)
-			// only format logfile
-			f, err := os.OpenFile(logdir+time.Now().Format(logfile), os.O_RDWR|os.O_CREATE|os.O_APPEND, 0666)
-			if err != nil {
-				log.Println(err)
-				return
-			}
-			w := csv.NewWriter(f)
-			// write header in empty file
-			if stat, err := f.Stat(); err == nil && stat.Size() == 0 {
-				if err := w.Write(append([]string{"Unix"}, kcp.DefaultSnmp.Header()...)); err != nil {
-					log.Println(err)
-				}
-			}
-			if err := w.Write(append([]string{fmt.Sprint(time.Now().Unix())}, kcp.DefaultSnmp.ToSlice()...)); err != nil {
-				log.Println(err)
-			}
-			kcp.DefaultSnmp.Reset()
-			w.Flush()
-			f.Close()
-		}
-	}
-}

smux/LICENSE

@@ -1,21 +0,0 @@
-MIT License
-
-Copyright (c) 2016-2017 Daniel Fu
-
-Permission is hereby granted, free of charge, to any person obtaining a copy
-of this software and associated documentation files (the "Software"), to deal
-in the Software without restriction, including without limitation the rights
-to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
-copies of the Software, and to permit persons to whom the Software is
-furnished to do so, subject to the following conditions:
-
-The above copyright notice and this permission notice shall be included in all
-copies or substantial portions of the Software.
-
-THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
-IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
-FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
-AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
-LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
-OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
-SOFTWARE.

smux/frame.go

@@ -1,60 +0,0 @@
-package smux
-
-import (
-	"encoding/binary"
-	"fmt"
-)
-
-const (
-	version = 1
-)
-
-const ( // cmds
-	cmdSYN byte = iota // stream open
-	cmdFIN             // stream close, a.k.a EOF mark
-	cmdPSH             // data push
-	cmdNOP             // no operation
-)
-
-const (
-	sizeOfVer    = 1
-	sizeOfCmd    = 1
-	sizeOfLength = 2
-	sizeOfSid    = 4
-	headerSize   = sizeOfVer + sizeOfCmd + sizeOfSid + sizeOfLength
-)
-
-// Frame defines a packet from or to be multiplexed into a single connection
-type Frame struct {
-	ver  byte
-	cmd  byte
-	sid  uint32
-	data []byte
-}
-
-func newFrame(cmd byte, sid uint32) Frame {
-	return Frame{ver: version, cmd: cmd, sid: sid}
-}
-
-type rawHeader [headerSize]byte
-
-func (h rawHeader) Version() byte {
-	return h[0]
-}
-
-func (h rawHeader) Cmd() byte {
-	return h[1]
-}
-
-func (h rawHeader) Length() uint16 {
-	return binary.LittleEndian.Uint16(h[2:])
-}
-
-func (h rawHeader) StreamID() uint32 {
-	return binary.LittleEndian.Uint32(h[4:])
-}
-
-func (h rawHeader) String() string {
-	return fmt.Sprintf("Version:%d Cmd:%d StreamID:%d Length:%d",
-		h.Version(), h.Cmd(), h.StreamID(), h.Length())
-}

smux/mux.go

@@ -1,80 +0,0 @@
-package smux
-
-import (
-	"fmt"
-	"io"
-	"time"
-
-	"github.com/pkg/errors"
-)
-
-// Config is used to tune the Smux session
-type Config struct {
-	// KeepAliveInterval is how often to send a NOP command to the remote
-	KeepAliveInterval time.Duration
-
-	// KeepAliveTimeout is how long the session
-	// will be closed if no data has arrived
-	KeepAliveTimeout time.Duration
-
-	// MaxFrameSize is used to control the maximum
-	// frame size to sent to the remote
-	MaxFrameSize int
-
-	// MaxReceiveBuffer is used to control the maximum
-	// number of data in the buffer pool
-	MaxReceiveBuffer int
-}
-
-// DefaultConfig is used to return a default configuration
-func DefaultConfig() *Config {
-	return &Config{
-		KeepAliveInterval: 10 * time.Second,
-		KeepAliveTimeout:  30 * time.Second,
-		MaxFrameSize:      32768,
-		MaxReceiveBuffer:  4194304,
-	}
-}
-
-// VerifyConfig is used to verify the sanity of configuration
-func VerifyConfig(config *Config) error {
-	if config.KeepAliveInterval == 0 {
-		return errors.New("keep-alive interval must be positive")
-	}
-	if config.KeepAliveTimeout < config.KeepAliveInterval {
-		return fmt.Errorf("keep-alive timeout must be larger than keep-alive interval")
-	}
-	if config.MaxFrameSize <= 0 {
-		return errors.New("max frame size must be positive")
-	}
-	if config.MaxFrameSize > 65535 {
-		return errors.New("max frame size must not be larger than 65535")
-	}
-	if config.MaxReceiveBuffer <= 0 {
-		return errors.New("max receive buffer must be positive")
-	}
-	return nil
-}
-
-// Server is used to initialize a new server-side connection.
-func Server(conn io.ReadWriteCloser, config *Config) (*Session, error) {
-	if config == nil {
-		config = DefaultConfig()
-	}
-	if err := VerifyConfig(config); err != nil {
-		return nil, err
-	}
-	return newSession(config, conn, false), nil
-}
-
-// Client is used to initialize a new client-side connection.
-func Client(conn io.ReadWriteCloser, config *Config) (*Session, error) {
-	if config == nil {
-		config = DefaultConfig()
-	}
-
-	if err := VerifyConfig(config); err != nil {
-		return nil, err
-	}
-	return newSession(config, conn, true), nil
-}

smux/session.go

@@ -1,350 +0,0 @@
-package smux
-
-import (
-	"encoding/binary"
-	"io"
-	"sync"
-	"sync/atomic"
-	"time"
-
-	"github.com/pkg/errors"
-)
-
-const (
-	defaultAcceptBacklog = 1024
-)
-
-const (
-	errBrokenPipe      = "broken pipe"
-	errInvalidProtocol = "invalid protocol version"
-	errGoAway          = "stream id overflows, should start a new connection"
-)
-
-type writeRequest struct {
-	frame  Frame
-	result chan writeResult
-}
-
-type writeResult struct {
-	n   int
-	err error
-}
-
-// Session defines a multiplexed connection for streams
-type Session struct {
-	conn io.ReadWriteCloser
-
-	config           *Config
-	nextStreamID     uint32 // next stream identifier
-	nextStreamIDLock sync.Mutex
-
-	bucket       int32         // token bucket
-	bucketNotify chan struct{} // used for waiting for tokens
-
-	streams    map[uint32]*Stream // all streams in this session
-	streamLock sync.Mutex         // locks streams
-
-	die       chan struct{} // flag session has died
-	dieLock   sync.Mutex
-	chAccepts chan *Stream
-
-	dataReady int32 // flag data has arrived
-
-	goAway int32 // flag id exhausted
-
-	deadline atomic.Value
-
-	writes chan writeRequest
-}
-
-func newSession(config *Config, conn io.ReadWriteCloser, client bool) *Session {
-	s := new(Session)
-	s.die = make(chan struct{})
-	s.conn = conn
-	s.config = config
-	s.streams = make(map[uint32]*Stream)
-	s.chAccepts = make(chan *Stream, defaultAcceptBacklog)
-	s.bucket = int32(config.MaxReceiveBuffer)
-	s.bucketNotify = make(chan struct{}, 1)
-	s.writes = make(chan writeRequest)
-
-	if client {
-		s.nextStreamID = 1
-	} else {
-		s.nextStreamID = 0
-	}
-	go s.recvLoop()
-	go s.sendLoop()
-	go s.keepalive()
-	return s
-}
-
-// OpenStream is used to create a new stream
-func (s *Session) OpenStream() (*Stream, error) {
-	if s.IsClosed() {
-		return nil, errors.New(errBrokenPipe)
-	}
-
-	// generate stream id
-	s.nextStreamIDLock.Lock()
-	if s.goAway > 0 {
-		s.nextStreamIDLock.Unlock()
-		return nil, errors.New(errGoAway)
-	}
-
-	s.nextStreamID += 2
-	sid := s.nextStreamID
-	if sid == sid%2 { // stream-id overflows
-		s.goAway = 1
-		s.nextStreamIDLock.Unlock()
-		return nil, errors.New(errGoAway)
-	}
-	s.nextStreamIDLock.Unlock()
-
-	stream := newStream(sid, s.config.MaxFrameSize, s)
-
-	if _, err := s.writeFrame(newFrame(cmdSYN, sid)); err != nil {
-		return nil, errors.Wrap(err, "writeFrame")
-	}
-
-	s.streamLock.Lock()
-	s.streams[sid] = stream
-	s.streamLock.Unlock()
-	return stream, nil
-}
-
-// AcceptStream is used to block until the next available stream
-// is ready to be accepted.
-func (s *Session) AcceptStream() (*Stream, error) {
-	var deadline <-chan time.Time
-	if d, ok := s.deadline.Load().(time.Time); ok && !d.IsZero() {
-		timer := time.NewTimer(time.Until(d))
-		defer timer.Stop()
-		deadline = timer.C
-	}
-	select {
-	case stream := <-s.chAccepts:
-		return stream, nil
-	case <-deadline:
-		return nil, errTimeout
-	case <-s.die:
-		return nil, errors.New(errBrokenPipe)
-	}
-}
-
-// Close is used to close the session and all streams.
-func (s *Session) Close() (err error) {
-	s.dieLock.Lock()
-
-	select {
-	case <-s.die:
-		s.dieLock.Unlock()
-		return errors.New(errBrokenPipe)
-	default:
-		close(s.die)
-		s.dieLock.Unlock()
-		s.streamLock.Lock()
-		for k := range s.streams {
-			s.streams[k].sessionClose()
-		}
-		s.streamLock.Unlock()
-		s.notifyBucket()
-		return s.conn.Close()
-	}
-}
-
-// notifyBucket notifies recvLoop that bucket is available
-func (s *Session) notifyBucket() {
-	select {
-	case s.bucketNotify <- struct{}{}:
-	default:
-	}
-}
-
-// IsClosed does a safe check to see if we have shutdown
-func (s *Session) IsClosed() bool {
-	select {
-	case <-s.die:
-		return true
-	default:
-		return false
-	}
-}
-
-// NumStreams returns the number of currently open streams
-func (s *Session) NumStreams() int {
-	if s.IsClosed() {
-		return 0
-	}
-	s.streamLock.Lock()
-	defer s.streamLock.Unlock()
-	return len(s.streams)
-}
-
-// SetDeadline sets a deadline used by Accept* calls.
-// A zero time value disables the deadline.
-func (s *Session) SetDeadline(t time.Time) error {
-	s.deadline.Store(t)
-	return nil
-}
-
-// notify the session that a stream has closed
-func (s *Session) streamClosed(sid uint32) {
-	s.streamLock.Lock()
-	if n := s.streams[sid].recycleTokens(); n > 0 { // return remaining tokens to the bucket
-		if atomic.AddInt32(&s.bucket, int32(n)) > 0 {
-			s.notifyBucket()
-		}
-	}
-	delete(s.streams, sid)
-	s.streamLock.Unlock()
-}
-
-// returnTokens is called by stream to return token after read
-func (s *Session) returnTokens(n int) {
-	if atomic.AddInt32(&s.bucket, int32(n)) > 0 {
-		s.notifyBucket()
-	}
-}
-
-// session read a frame from underlying connection
-// it's data is pointed to the input buffer
-func (s *Session) readFrame(buffer []byte) (f Frame, err error) {
-	var hdr rawHeader
-	if _, err := io.ReadFull(s.conn, hdr[:]); err != nil {
-		return f, errors.Wrap(err, "readFrame")
-	}
-
-	if hdr.Version() != version {
-		return f, errors.New(errInvalidProtocol)
-	}
-
-	f.ver = hdr.Version()
-	f.cmd = hdr.Cmd()
-	f.sid = hdr.StreamID()
-	if length := hdr.Length(); length > 0 {
-		f.data = buffer[:length]
-		if _, err := io.ReadFull(s.conn, f.data); err != nil {
-			return f, errors.Wrap(err, "readFrame")
-		}
-	}
-	return f, nil
-}
-
-// recvLoop keeps on reading from underlying connection if tokens are available
-func (s *Session) recvLoop() {
-	buffer := make([]byte, 1<<16)
-	for {
-		for atomic.LoadInt32(&s.bucket) <= 0 && !s.IsClosed() {
-			<-s.bucketNotify
-		}
-
-		if f, err := s.readFrame(buffer); err == nil {
-			atomic.StoreInt32(&s.dataReady, 1)
-
-			switch f.cmd {
-			case cmdNOP:
-			case cmdSYN:
-				s.streamLock.Lock()
-				if _, ok := s.streams[f.sid]; !ok {
-					stream := newStream(f.sid, s.config.MaxFrameSize, s)
-					s.streams[f.sid] = stream
-					select {
-					case s.chAccepts <- stream:
-					case <-s.die:
-					}
-				}
-				s.streamLock.Unlock()
-			case cmdFIN:
-				s.streamLock.Lock()
-				if stream, ok := s.streams[f.sid]; ok {
-					stream.markRST()
-					stream.notifyReadEvent()
-				}
-				s.streamLock.Unlock()
-			case cmdPSH:
-				s.streamLock.Lock()
-				if stream, ok := s.streams[f.sid]; ok {
-					atomic.AddInt32(&s.bucket, -int32(len(f.data)))
-					stream.pushBytes(f.data)
-					stream.notifyReadEvent()
-				}
-				s.streamLock.Unlock()
-			default:
-				s.Close()
-				return
-			}
-		} else {
-			s.Close()
-			return
-		}
-	}
-}
-
-func (s *Session) keepalive() {
-	tickerPing := time.NewTicker(s.config.KeepAliveInterval)
-	tickerTimeout := time.NewTicker(s.config.KeepAliveTimeout)
-	defer tickerPing.Stop()
-	defer tickerTimeout.Stop()
-	for {
-		select {
-		case <-tickerPing.C:
-			s.writeFrame(newFrame(cmdNOP, 0))
-			s.notifyBucket() // force a signal to the recvLoop
-		case <-tickerTimeout.C:
-			if !atomic.CompareAndSwapInt32(&s.dataReady, 1, 0) {
-				s.Close()
-				return
-			}
-		case <-s.die:
-			return
-		}
-	}
-}
-
-func (s *Session) sendLoop() {
-	buf := make([]byte, (1<<16)+headerSize)
-	for {
-		select {
-		case <-s.die:
-			return
-		case request := <-s.writes:
-			buf[0] = request.frame.ver
-			buf[1] = request.frame.cmd
-			binary.LittleEndian.PutUint16(buf[2:], uint16(len(request.frame.data)))
-			binary.LittleEndian.PutUint32(buf[4:], request.frame.sid)
-			copy(buf[headerSize:], request.frame.data)
-			n, err := s.conn.Write(buf[:headerSize+len(request.frame.data)])
-
-			n -= headerSize
-			if n < 0 {
-				n = 0
-			}
-
-			result := writeResult{
-				n:   n,
-				err: err,
-			}
-
-			request.result <- result
-			close(request.result)
-		}
-	}
-}
-
-// writeFrame writes the frame to the underlying connection
-// and returns the number of bytes written if successful
-func (s *Session) writeFrame(f Frame) (n int, err error) {
-	req := writeRequest{
-		frame:  f,
-		result: make(chan writeResult, 1),
-	}
-	select {
-	case <-s.die:
-		return 0, errors.New(errBrokenPipe)
-	case s.writes <- req:
-	}
-
-	result := <-req.result
-	return result.n, result.err
-}

smux/stream.go

@@ -1,262 +0,0 @@
-package smux
-
-import (
-	"bytes"
-	"io"
-	"net"
-	"sync"
-	"sync/atomic"
-	"time"
-
-	"github.com/pkg/errors"
-)
-
-// Stream implements net.Conn
-type Stream struct {
-	id            uint32
-	rstflag       int32
-	sess          *Session
-	buffer        bytes.Buffer
-	bufferLock    sync.Mutex
-	frameSize     int
-	chReadEvent   chan struct{} // notify a read event
-	die           chan struct{} // flag the stream has closed
-	dieLock       sync.Mutex
-	readDeadline  atomic.Value
-	writeDeadline atomic.Value
-}
-
-// newStream initiates a Stream struct
-func newStream(id uint32, frameSize int, sess *Session) *Stream {
-	s := new(Stream)
-	s.id = id
-	s.chReadEvent = make(chan struct{}, 1)
-	s.frameSize = frameSize
-	s.sess = sess
-	s.die = make(chan struct{})
-	return s
-}
-
-// ID returns the unique stream ID.
-func (s *Stream) ID() uint32 {
-	return s.id
-}
-
-// Read implements net.Conn
-func (s *Stream) Read(b []byte) (n int, err error) {
-	if len(b) == 0 {
-		select {
-		case <-s.die:
-			return 0, errors.New(errBrokenPipe)
-		default:
-			return 0, nil
-		}
-	}
-
-	var deadline <-chan time.Time
-	if d, ok := s.readDeadline.Load().(time.Time); ok && !d.IsZero() {
-		timer := time.NewTimer(time.Until(d))
-		defer timer.Stop()
-		deadline = timer.C
-	}
-
-READ:
-	s.bufferLock.Lock()
-	n, _ = s.buffer.Read(b)
-	s.bufferLock.Unlock()
-
-	if n > 0 {
-		s.sess.returnTokens(n)
-		return n, nil
-	} else if atomic.LoadInt32(&s.rstflag) == 1 {
-		_ = s.Close()
-		return 0, io.EOF
-	}
-
-	select {
-	case <-s.chReadEvent:
-		goto READ
-	case <-deadline:
-		return n, errTimeout
-	case <-s.die:
-		return 0, errors.New(errBrokenPipe)
-	}
-}
-
-// Write implements net.Conn
-func (s *Stream) Write(b []byte) (n int, err error) {
-	var deadline <-chan time.Time
-	if d, ok := s.writeDeadline.Load().(time.Time); ok && !d.IsZero() {
-		timer := time.NewTimer(time.Until(d))
-		defer timer.Stop()
-		deadline = timer.C
-	}
-
-	select {
-	case <-s.die:
-		return 0, errors.New(errBrokenPipe)
-	default:
-	}
-
-	frames := s.split(b, cmdPSH, s.id)
-	sent := 0
-	for k := range frames {
-		req := writeRequest{
-			frame:  frames[k],
-			result: make(chan writeResult, 1),
-		}
-
-		select {
-		case s.sess.writes <- req:
-		case <-s.die:
-			return sent, errors.New(errBrokenPipe)
-		case <-deadline:
-			return sent, errTimeout
-		}
-
-		select {
-		case result := <-req.result:
-			sent += result.n
-			if result.err != nil {
-				return sent, result.err
-			}
-		case <-s.die:
-			return sent, errors.New(errBrokenPipe)
-		case <-deadline:
-			return sent, errTimeout
-		}
-	}
-	return sent, nil
-}
-
-// Close implements net.Conn
-func (s *Stream) Close() error {
-	s.dieLock.Lock()
-
-	select {
-	case <-s.die:
-		s.dieLock.Unlock()
-		return errors.New(errBrokenPipe)
-	default:
-		close(s.die)
-		s.dieLock.Unlock()
-		s.sess.streamClosed(s.id)
-		_, err := s.sess.writeFrame(newFrame(cmdFIN, s.id))
-		return err
-	}
-}
-
-// SetReadDeadline sets the read deadline as defined by
-// net.Conn.SetReadDeadline.
-// A zero time value disables the deadline.
-func (s *Stream) SetReadDeadline(t time.Time) error {
-	s.readDeadline.Store(t)
-	return nil
-}
-
-// SetWriteDeadline sets the write deadline as defined by
-// net.Conn.SetWriteDeadline.
-// A zero time value disables the deadline.
-func (s *Stream) SetWriteDeadline(t time.Time) error {
-	s.writeDeadline.Store(t)
-	return nil
-}
-
-// SetDeadline sets both read and write deadlines as defined by
-// net.Conn.SetDeadline.
-// A zero time value disables the deadlines.
-func (s *Stream) SetDeadline(t time.Time) error {
-	if err := s.SetReadDeadline(t); err != nil {
-		return err
-	}
-	if err := s.SetWriteDeadline(t); err != nil {
-		return err
-	}
-	return nil
-}
-
-// session closes the stream
-func (s *Stream) sessionClose() {
-	s.dieLock.Lock()
-	defer s.dieLock.Unlock()
-
-	select {
-	case <-s.die:
-	default:
-		close(s.die)
-	}
-}
-
-// LocalAddr satisfies net.Conn interface
-func (s *Stream) LocalAddr() net.Addr {
-	if ts, ok := s.sess.conn.(interface {
-		LocalAddr() net.Addr
-	}); ok {
-		return ts.LocalAddr()
-	}
-	return nil
-}
-
-// RemoteAddr satisfies net.Conn interface
-func (s *Stream) RemoteAddr() net.Addr {
-	if ts, ok := s.sess.conn.(interface {
-		RemoteAddr() net.Addr
-	}); ok {
-		return ts.RemoteAddr()
-	}
-	return nil
-}
-
-// pushBytes a slice into buffer
-func (s *Stream) pushBytes(p []byte) {
-	s.bufferLock.Lock()
-	s.buffer.Write(p)
-	s.bufferLock.Unlock()
-}
-
-// recycleTokens transform remaining bytes to tokens(will truncate buffer)
-func (s *Stream) recycleTokens() (n int) {
-	s.bufferLock.Lock()
-	n = s.buffer.Len()
-	s.buffer.Reset()
-	s.bufferLock.Unlock()
-	return
-}
-
-// split large byte buffer into smaller frames, reference only
-func (s *Stream) split(bts []byte, cmd byte, sid uint32) []Frame {
-	frames := make([]Frame, 0, len(bts)/s.frameSize+1)
-	for len(bts) > s.frameSize {
-		frame := newFrame(cmd, sid)
-		frame.data = bts[:s.frameSize]
-		bts = bts[s.frameSize:]
-		frames = append(frames, frame)
-	}
-	if len(bts) > 0 {
-		frame := newFrame(cmd, sid)
-		frame.data = bts
-		frames = append(frames, frame)
-	}
-	return frames
-}
-
-// notify read event
-func (s *Stream) notifyReadEvent() {
-	select {
-	case s.chReadEvent <- struct{}{}:
-	default:
-	}
-}
-
-// mark this stream has been reset
-func (s *Stream) markRST() {
-	atomic.StoreInt32(&s.rstflag, 1)
-}
-
-var errTimeout error = &timeoutError{}
-
-type timeoutError struct{}
-
-func (e *timeoutError) Error() string   { return "i/o timeout" }
-func (e *timeoutError) Timeout() bool   { return true }
-func (e *timeoutError) Temporary() bool { return true }