first commit

.gitignore

@@ -0,0 +1 @@
+/.idea

README.md

@@ -0,0 +1 @@
+p2ptun first commit

kcp-go/LICENSE

@@ -0,0 +1,22 @@
+The MIT License (MIT)
+
+Copyright (c) 2015 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.
+

kcp-go/crypt.go

@@ -0,0 +1,785 @@
+package kcp
+
+import (
+	"crypto/aes"
+	"crypto/cipher"
+	"crypto/des"
+	"crypto/sha1"
+
+	"github.com/templexxx/xor"
+	"github.com/tjfoc/gmsm/sm4"
+
+	"golang.org/x/crypto/blowfish"
+	"golang.org/x/crypto/cast5"
+	"golang.org/x/crypto/pbkdf2"
+	"golang.org/x/crypto/salsa20"
+	"golang.org/x/crypto/tea"
+	"golang.org/x/crypto/twofish"
+	"golang.org/x/crypto/xtea"
+)
+
+var (
+	initialVector = []byte{167, 115, 79, 156, 18, 172, 27, 1, 164, 21, 242, 193, 252, 120, 230, 107}
+	saltxor       = `sH3CIVoF#rWLtJo6`
+)
+
+// BlockCrypt defines encryption/decryption methods for a given byte slice.
+// Notes on implementing: the data to be encrypted contains a builtin
+// nonce at the first 16 bytes
+type BlockCrypt interface {
+	// Encrypt encrypts the whole block in src into dst.
+	// Dst and src may point at the same memory.
+	Encrypt(dst, src []byte)
+
+	// Decrypt decrypts the whole block in src into dst.
+	// Dst and src may point at the same memory.
+	Decrypt(dst, src []byte)
+}
+
+type salsa20BlockCrypt struct {
+	key [32]byte
+}
+
+// NewSalsa20BlockCrypt https://en.wikipedia.org/wiki/Salsa20
+func NewSalsa20BlockCrypt(key []byte) (BlockCrypt, error) {
+	c := new(salsa20BlockCrypt)
+	copy(c.key[:], key)
+	return c, nil
+}
+
+func (c *salsa20BlockCrypt) Encrypt(dst, src []byte) {
+	salsa20.XORKeyStream(dst[8:], src[8:], src[:8], &c.key)
+	copy(dst[:8], src[:8])
+}
+func (c *salsa20BlockCrypt) Decrypt(dst, src []byte) {
+	salsa20.XORKeyStream(dst[8:], src[8:], src[:8], &c.key)
+	copy(dst[:8], src[:8])
+}
+
+type sm4BlockCrypt struct {
+	encbuf [sm4.BlockSize]byte
+	decbuf [2 * sm4.BlockSize]byte
+	block  cipher.Block
+}
+
+// NewSM4BlockCrypt https://github.com/tjfoc/gmsm/tree/master/sm4
+func NewSM4BlockCrypt(key []byte) (BlockCrypt, error) {
+	c := new(sm4BlockCrypt)
+	block, err := sm4.NewCipher(key)
+	if err != nil {
+		return nil, err
+	}
+	c.block = block
+	return c, nil
+}
+
+func (c *sm4BlockCrypt) Encrypt(dst, src []byte) { encrypt(c.block, dst, src, c.encbuf[:]) }
+func (c *sm4BlockCrypt) Decrypt(dst, src []byte) { decrypt(c.block, dst, src, c.decbuf[:]) }
+
+type twofishBlockCrypt struct {
+	encbuf [twofish.BlockSize]byte
+	decbuf [2 * twofish.BlockSize]byte
+	block  cipher.Block
+}
+
+// NewTwofishBlockCrypt https://en.wikipedia.org/wiki/Twofish
+func NewTwofishBlockCrypt(key []byte) (BlockCrypt, error) {
+	c := new(twofishBlockCrypt)
+	block, err := twofish.NewCipher(key)
+	if err != nil {
+		return nil, err
+	}
+	c.block = block
+	return c, nil
+}
+
+func (c *twofishBlockCrypt) Encrypt(dst, src []byte) { encrypt(c.block, dst, src, c.encbuf[:]) }
+func (c *twofishBlockCrypt) Decrypt(dst, src []byte) { decrypt(c.block, dst, src, c.decbuf[:]) }
+
+type tripleDESBlockCrypt struct {
+	encbuf [des.BlockSize]byte
+	decbuf [2 * des.BlockSize]byte
+	block  cipher.Block
+}
+
+// NewTripleDESBlockCrypt https://en.wikipedia.org/wiki/Triple_DES
+func NewTripleDESBlockCrypt(key []byte) (BlockCrypt, error) {
+	c := new(tripleDESBlockCrypt)
+	block, err := des.NewTripleDESCipher(key)
+	if err != nil {
+		return nil, err
+	}
+	c.block = block
+	return c, nil
+}
+
+func (c *tripleDESBlockCrypt) Encrypt(dst, src []byte) { encrypt(c.block, dst, src, c.encbuf[:]) }
+func (c *tripleDESBlockCrypt) Decrypt(dst, src []byte) { decrypt(c.block, dst, src, c.decbuf[:]) }
+
+type cast5BlockCrypt struct {
+	encbuf [cast5.BlockSize]byte
+	decbuf [2 * cast5.BlockSize]byte
+	block  cipher.Block
+}
+
+// NewCast5BlockCrypt https://en.wikipedia.org/wiki/CAST-128
+func NewCast5BlockCrypt(key []byte) (BlockCrypt, error) {
+	c := new(cast5BlockCrypt)
+	block, err := cast5.NewCipher(key)
+	if err != nil {
+		return nil, err
+	}
+	c.block = block
+	return c, nil
+}
+
+func (c *cast5BlockCrypt) Encrypt(dst, src []byte) { encrypt(c.block, dst, src, c.encbuf[:]) }
+func (c *cast5BlockCrypt) Decrypt(dst, src []byte) { decrypt(c.block, dst, src, c.decbuf[:]) }
+
+type blowfishBlockCrypt struct {
+	encbuf [blowfish.BlockSize]byte
+	decbuf [2 * blowfish.BlockSize]byte
+	block  cipher.Block
+}
+
+// NewBlowfishBlockCrypt https://en.wikipedia.org/wiki/Blowfish_(cipher)
+func NewBlowfishBlockCrypt(key []byte) (BlockCrypt, error) {
+	c := new(blowfishBlockCrypt)
+	block, err := blowfish.NewCipher(key)
+	if err != nil {
+		return nil, err
+	}
+	c.block = block
+	return c, nil
+}
+
+func (c *blowfishBlockCrypt) Encrypt(dst, src []byte) { encrypt(c.block, dst, src, c.encbuf[:]) }
+func (c *blowfishBlockCrypt) Decrypt(dst, src []byte) { decrypt(c.block, dst, src, c.decbuf[:]) }
+
+type aesBlockCrypt struct {
+	encbuf [aes.BlockSize]byte
+	decbuf [2 * aes.BlockSize]byte
+	block  cipher.Block
+}
+
+// NewAESBlockCrypt https://en.wikipedia.org/wiki/Advanced_Encryption_Standard
+func NewAESBlockCrypt(key []byte) (BlockCrypt, error) {
+	c := new(aesBlockCrypt)
+	block, err := aes.NewCipher(key)
+	if err != nil {
+		return nil, err
+	}
+	c.block = block
+	return c, nil
+}
+
+func (c *aesBlockCrypt) Encrypt(dst, src []byte) { encrypt(c.block, dst, src, c.encbuf[:]) }
+func (c *aesBlockCrypt) Decrypt(dst, src []byte) { decrypt(c.block, dst, src, c.decbuf[:]) }
+
+type teaBlockCrypt struct {
+	encbuf [tea.BlockSize]byte
+	decbuf [2 * tea.BlockSize]byte
+	block  cipher.Block
+}
+
+// NewTEABlockCrypt https://en.wikipedia.org/wiki/Tiny_Encryption_Algorithm
+func NewTEABlockCrypt(key []byte) (BlockCrypt, error) {
+	c := new(teaBlockCrypt)
+	block, err := tea.NewCipherWithRounds(key, 16)
+	if err != nil {
+		return nil, err
+	}
+	c.block = block
+	return c, nil
+}
+
+func (c *teaBlockCrypt) Encrypt(dst, src []byte) { encrypt(c.block, dst, src, c.encbuf[:]) }
+func (c *teaBlockCrypt) Decrypt(dst, src []byte) { decrypt(c.block, dst, src, c.decbuf[:]) }
+
+type xteaBlockCrypt struct {
+	encbuf [xtea.BlockSize]byte
+	decbuf [2 * xtea.BlockSize]byte
+	block  cipher.Block
+}
+
+// NewXTEABlockCrypt https://en.wikipedia.org/wiki/XTEA
+func NewXTEABlockCrypt(key []byte) (BlockCrypt, error) {
+	c := new(xteaBlockCrypt)
+	block, err := xtea.NewCipher(key)
+	if err != nil {
+		return nil, err
+	}
+	c.block = block
+	return c, nil
+}
+
+func (c *xteaBlockCrypt) Encrypt(dst, src []byte) { encrypt(c.block, dst, src, c.encbuf[:]) }
+func (c *xteaBlockCrypt) Decrypt(dst, src []byte) { decrypt(c.block, dst, src, c.decbuf[:]) }
+
+type simpleXORBlockCrypt struct {
+	xortbl []byte
+}
+
+// NewSimpleXORBlockCrypt simple xor with key expanding
+func NewSimpleXORBlockCrypt(key []byte) (BlockCrypt, error) {
+	c := new(simpleXORBlockCrypt)
+	c.xortbl = pbkdf2.Key(key, []byte(saltxor), 32, mtuLimit, sha1.New)
+	return c, nil
+}
+
+func (c *simpleXORBlockCrypt) Encrypt(dst, src []byte) { xor.Bytes(dst, src, c.xortbl) }
+func (c *simpleXORBlockCrypt) Decrypt(dst, src []byte) { xor.Bytes(dst, src, c.xortbl) }
+
+type noneBlockCrypt struct{}
+
+// NewNoneBlockCrypt does nothing but copying
+func NewNoneBlockCrypt(key []byte) (BlockCrypt, error) {
+	return new(noneBlockCrypt), nil
+}
+
+func (c *noneBlockCrypt) Encrypt(dst, src []byte) { copy(dst, src) }
+func (c *noneBlockCrypt) Decrypt(dst, src []byte) { copy(dst, src) }
+
+// packet encryption with local CFB mode
+func encrypt(block cipher.Block, dst, src, buf []byte) {
+	switch block.BlockSize() {
+	case 8:
+		encrypt8(block, dst, src, buf)
+	case 16:
+		encrypt16(block, dst, src, buf)
+	default:
+		encryptVariant(block, dst, src, buf)
+	}
+}
+
+// optimized encryption for the ciphers which works in 8-bytes
+func encrypt8(block cipher.Block, dst, src, buf []byte) {
+	tbl := buf[:8]
+	block.Encrypt(tbl, initialVector)
+	n := len(src) / 8
+	base := 0
+	repeat := n / 8
+	left := n % 8
+	for i := 0; i < repeat; i++ {
+		s := src[base:][0:64]
+		d := dst[base:][0:64]
+		// 1
+		xor.BytesSrc1(d[0:8], s[0:8], tbl)
+		block.Encrypt(tbl, d[0:8])
+		// 2
+		xor.BytesSrc1(d[8:16], s[8:16], tbl)
+		block.Encrypt(tbl, d[8:16])
+		// 3
+		xor.BytesSrc1(d[16:24], s[16:24], tbl)
+		block.Encrypt(tbl, d[16:24])
+		// 4
+		xor.BytesSrc1(d[24:32], s[24:32], tbl)
+		block.Encrypt(tbl, d[24:32])
+		// 5
+		xor.BytesSrc1(d[32:40], s[32:40], tbl)
+		block.Encrypt(tbl, d[32:40])
+		// 6
+		xor.BytesSrc1(d[40:48], s[40:48], tbl)
+		block.Encrypt(tbl, d[40:48])
+		// 7
+		xor.BytesSrc1(d[48:56], s[48:56], tbl)
+		block.Encrypt(tbl, d[48:56])
+		// 8
+		xor.BytesSrc1(d[56:64], s[56:64], tbl)
+		block.Encrypt(tbl, d[56:64])
+		base += 64
+	}
+
+	switch left {
+	case 7:
+		xor.BytesSrc1(dst[base:], src[base:], tbl)
+		block.Encrypt(tbl, dst[base:])
+		base += 8
+		fallthrough
+	case 6:
+		xor.BytesSrc1(dst[base:], src[base:], tbl)
+		block.Encrypt(tbl, dst[base:])
+		base += 8
+		fallthrough
+	case 5:
+		xor.BytesSrc1(dst[base:], src[base:], tbl)
+		block.Encrypt(tbl, dst[base:])
+		base += 8
+		fallthrough
+	case 4:
+		xor.BytesSrc1(dst[base:], src[base:], tbl)
+		block.Encrypt(tbl, dst[base:])
+		base += 8
+		fallthrough
+	case 3:
+		xor.BytesSrc1(dst[base:], src[base:], tbl)
+		block.Encrypt(tbl, dst[base:])
+		base += 8
+		fallthrough
+	case 2:
+		xor.BytesSrc1(dst[base:], src[base:], tbl)
+		block.Encrypt(tbl, dst[base:])
+		base += 8
+		fallthrough
+	case 1:
+		xor.BytesSrc1(dst[base:], src[base:], tbl)
+		block.Encrypt(tbl, dst[base:])
+		base += 8
+		fallthrough
+	case 0:
+		xor.BytesSrc0(dst[base:], src[base:], tbl)
+	}
+}
+
+// optimized encryption for the ciphers which works in 16-bytes
+func encrypt16(block cipher.Block, dst, src, buf []byte) {
+	tbl := buf[:16]
+	block.Encrypt(tbl, initialVector)
+	n := len(src) / 16
+	base := 0
+	repeat := n / 8
+	left := n % 8
+	for i := 0; i < repeat; i++ {
+		s := src[base:][0:128]
+		d := dst[base:][0:128]
+		// 1
+		xor.BytesSrc1(d[0:16], s[0:16], tbl)
+		block.Encrypt(tbl, d[0:16])
+		// 2
+		xor.BytesSrc1(d[16:32], s[16:32], tbl)
+		block.Encrypt(tbl, d[16:32])
+		// 3
+		xor.BytesSrc1(d[32:48], s[32:48], tbl)
+		block.Encrypt(tbl, d[32:48])
+		// 4
+		xor.BytesSrc1(d[48:64], s[48:64], tbl)
+		block.Encrypt(tbl, d[48:64])
+		// 5
+		xor.BytesSrc1(d[64:80], s[64:80], tbl)
+		block.Encrypt(tbl, d[64:80])
+		// 6
+		xor.BytesSrc1(d[80:96], s[80:96], tbl)
+		block.Encrypt(tbl, d[80:96])
+		// 7
+		xor.BytesSrc1(d[96:112], s[96:112], tbl)
+		block.Encrypt(tbl, d[96:112])
+		// 8
+		xor.BytesSrc1(d[112:128], s[112:128], tbl)
+		block.Encrypt(tbl, d[112:128])
+		base += 128
+	}
+
+	switch left {
+	case 7:
+		xor.BytesSrc1(dst[base:], src[base:], tbl)
+		block.Encrypt(tbl, dst[base:])
+		base += 16
+		fallthrough
+	case 6:
+		xor.BytesSrc1(dst[base:], src[base:], tbl)
+		block.Encrypt(tbl, dst[base:])
+		base += 16
+		fallthrough
+	case 5:
+		xor.BytesSrc1(dst[base:], src[base:], tbl)
+		block.Encrypt(tbl, dst[base:])
+		base += 16
+		fallthrough
+	case 4:
+		xor.BytesSrc1(dst[base:], src[base:], tbl)
+		block.Encrypt(tbl, dst[base:])
+		base += 16
+		fallthrough
+	case 3:
+		xor.BytesSrc1(dst[base:], src[base:], tbl)
+		block.Encrypt(tbl, dst[base:])
+		base += 16
+		fallthrough
+	case 2:
+		xor.BytesSrc1(dst[base:], src[base:], tbl)
+		block.Encrypt(tbl, dst[base:])
+		base += 16
+		fallthrough
+	case 1:
+		xor.BytesSrc1(dst[base:], src[base:], tbl)
+		block.Encrypt(tbl, dst[base:])
+		base += 16
+		fallthrough
+	case 0:
+		xor.BytesSrc0(dst[base:], src[base:], tbl)
+	}
+}
+
+func encryptVariant(block cipher.Block, dst, src, buf []byte) {
+	blocksize := block.BlockSize()
+	tbl := buf[:blocksize]
+	block.Encrypt(tbl, initialVector)
+	n := len(src) / blocksize
+	base := 0
+	repeat := n / 8
+	left := n % 8
+	for i := 0; i < repeat; i++ {
+		// 1
+		xor.BytesSrc1(dst[base:], src[base:], tbl)
+		block.Encrypt(tbl, dst[base:])
+		base += blocksize
+
+		// 2
+		xor.BytesSrc1(dst[base:], src[base:], tbl)
+		block.Encrypt(tbl, dst[base:])
+		base += blocksize
+
+		// 3
+		xor.BytesSrc1(dst[base:], src[base:], tbl)
+		block.Encrypt(tbl, dst[base:])
+		base += blocksize
+
+		// 4
+		xor.BytesSrc1(dst[base:], src[base:], tbl)
+		block.Encrypt(tbl, dst[base:])
+		base += blocksize
+
+		// 5
+		xor.BytesSrc1(dst[base:], src[base:], tbl)
+		block.Encrypt(tbl, dst[base:])
+		base += blocksize
+
+		// 6
+		xor.BytesSrc1(dst[base:], src[base:], tbl)
+		block.Encrypt(tbl, dst[base:])
+		base += blocksize
+
+		// 7
+		xor.BytesSrc1(dst[base:], src[base:], tbl)
+		block.Encrypt(tbl, dst[base:])
+		base += blocksize
+
+		// 8
+		xor.BytesSrc1(dst[base:], src[base:], tbl)
+		block.Encrypt(tbl, dst[base:])
+		base += blocksize
+	}
+
+	switch left {
+	case 7:
+		xor.BytesSrc1(dst[base:], src[base:], tbl)
+		block.Encrypt(tbl, dst[base:])
+		base += blocksize
+		fallthrough
+	case 6:
+		xor.BytesSrc1(dst[base:], src[base:], tbl)
+		block.Encrypt(tbl, dst[base:])
+		base += blocksize
+		fallthrough
+	case 5:
+		xor.BytesSrc1(dst[base:], src[base:], tbl)
+		block.Encrypt(tbl, dst[base:])
+		base += blocksize
+		fallthrough
+	case 4:
+		xor.BytesSrc1(dst[base:], src[base:], tbl)
+		block.Encrypt(tbl, dst[base:])
+		base += blocksize
+		fallthrough
+	case 3:
+		xor.BytesSrc1(dst[base:], src[base:], tbl)
+		block.Encrypt(tbl, dst[base:])
+		base += blocksize
+		fallthrough
+	case 2:
+		xor.BytesSrc1(dst[base:], src[base:], tbl)
+		block.Encrypt(tbl, dst[base:])
+		base += blocksize
+		fallthrough
+	case 1:
+		xor.BytesSrc1(dst[base:], src[base:], tbl)
+		block.Encrypt(tbl, dst[base:])
+		base += blocksize
+		fallthrough
+	case 0:
+		xor.BytesSrc0(dst[base:], src[base:], tbl)
+	}
+}
+
+// decryption
+func decrypt(block cipher.Block, dst, src, buf []byte) {
+	switch block.BlockSize() {
+	case 8:
+		decrypt8(block, dst, src, buf)
+	case 16:
+		decrypt16(block, dst, src, buf)
+	default:
+		decryptVariant(block, dst, src, buf)
+	}
+}
+
+func decrypt8(block cipher.Block, dst, src, buf []byte) {
+	tbl := buf[0:8]
+	next := buf[8:16]
+	block.Encrypt(tbl, initialVector)
+	n := len(src) / 8
+	base := 0
+	repeat := n / 8
+	left := n % 8
+	for i := 0; i < repeat; i++ {
+		s := src[base:][0:64]
+		d := dst[base:][0:64]
+		// 1
+		block.Encrypt(next, s[0:8])
+		xor.BytesSrc1(d[0:8], s[0:8], tbl)
+		// 2
+		block.Encrypt(tbl, s[8:16])
+		xor.BytesSrc1(d[8:16], s[8:16], next)
+		// 3
+		block.Encrypt(next, s[16:24])
+		xor.BytesSrc1(d[16:24], s[16:24], tbl)
+		// 4
+		block.Encrypt(tbl, s[24:32])
+		xor.BytesSrc1(d[24:32], s[24:32], next)
+		// 5
+		block.Encrypt(next, s[32:40])
+		xor.BytesSrc1(d[32:40], s[32:40], tbl)
+		// 6
+		block.Encrypt(tbl, s[40:48])
+		xor.BytesSrc1(d[40:48], s[40:48], next)
+		// 7
+		block.Encrypt(next, s[48:56])
+		xor.BytesSrc1(d[48:56], s[48:56], tbl)
+		// 8
+		block.Encrypt(tbl, s[56:64])
+		xor.BytesSrc1(d[56:64], s[56:64], next)
+		base += 64
+	}
+
+	switch left {
+	case 7:
+		block.Encrypt(next, src[base:])
+		xor.BytesSrc1(dst[base:], src[base:], tbl)
+		tbl, next = next, tbl
+		base += 8
+		fallthrough
+	case 6:
+		block.Encrypt(next, src[base:])
+		xor.BytesSrc1(dst[base:], src[base:], tbl)
+		tbl, next = next, tbl
+		base += 8
+		fallthrough
+	case 5:
+		block.Encrypt(next, src[base:])
+		xor.BytesSrc1(dst[base:], src[base:], tbl)
+		tbl, next = next, tbl
+		base += 8
+		fallthrough
+	case 4:
+		block.Encrypt(next, src[base:])
+		xor.BytesSrc1(dst[base:], src[base:], tbl)
+		tbl, next = next, tbl
+		base += 8
+		fallthrough
+	case 3:
+		block.Encrypt(next, src[base:])
+		xor.BytesSrc1(dst[base:], src[base:], tbl)
+		tbl, next = next, tbl
+		base += 8
+		fallthrough
+	case 2:
+		block.Encrypt(next, src[base:])
+		xor.BytesSrc1(dst[base:], src[base:], tbl)
+		tbl, next = next, tbl
+		base += 8
+		fallthrough
+	case 1:
+		block.Encrypt(next, src[base:])
+		xor.BytesSrc1(dst[base:], src[base:], tbl)
+		tbl, next = next, tbl
+		base += 8
+		fallthrough
+	case 0:
+		xor.BytesSrc0(dst[base:], src[base:], tbl)
+	}
+}
+
+func decrypt16(block cipher.Block, dst, src, buf []byte) {
+	tbl := buf[0:16]
+	next := buf[16:32]
+	block.Encrypt(tbl, initialVector)
+	n := len(src) / 16
+	base := 0
+	repeat := n / 8
+	left := n % 8
+	for i := 0; i < repeat; i++ {
+		s := src[base:][0:128]
+		d := dst[base:][0:128]
+		// 1
+		block.Encrypt(next, s[0:16])
+		xor.BytesSrc1(d[0:16], s[0:16], tbl)
+		// 2
+		block.Encrypt(tbl, s[16:32])
+		xor.BytesSrc1(d[16:32], s[16:32], next)
+		// 3
+		block.Encrypt(next, s[32:48])
+		xor.BytesSrc1(d[32:48], s[32:48], tbl)
+		// 4
+		block.Encrypt(tbl, s[48:64])
+		xor.BytesSrc1(d[48:64], s[48:64], next)
+		// 5
+		block.Encrypt(next, s[64:80])
+		xor.BytesSrc1(d[64:80], s[64:80], tbl)
+		// 6
+		block.Encrypt(tbl, s[80:96])
+		xor.BytesSrc1(d[80:96], s[80:96], next)
+		// 7
+		block.Encrypt(next, s[96:112])
+		xor.BytesSrc1(d[96:112], s[96:112], tbl)
+		// 8
+		block.Encrypt(tbl, s[112:128])
+		xor.BytesSrc1(d[112:128], s[112:128], next)
+		base += 128
+	}
+
+	switch left {
+	case 7:
+		block.Encrypt(next, src[base:])
+		xor.BytesSrc1(dst[base:], src[base:], tbl)
+		tbl, next = next, tbl
+		base += 16
+		fallthrough
+	case 6:
+		block.Encrypt(next, src[base:])
+		xor.BytesSrc1(dst[base:], src[base:], tbl)
+		tbl, next = next, tbl
+		base += 16
+		fallthrough
+	case 5:
+		block.Encrypt(next, src[base:])
+		xor.BytesSrc1(dst[base:], src[base:], tbl)
+		tbl, next = next, tbl
+		base += 16
+		fallthrough
+	case 4:
+		block.Encrypt(next, src[base:])
+		xor.BytesSrc1(dst[base:], src[base:], tbl)
+		tbl, next = next, tbl
+		base += 16
+		fallthrough
+	case 3:
+		block.Encrypt(next, src[base:])
+		xor.BytesSrc1(dst[base:], src[base:], tbl)
+		tbl, next = next, tbl
+		base += 16
+		fallthrough
+	case 2:
+		block.Encrypt(next, src[base:])
+		xor.BytesSrc1(dst[base:], src[base:], tbl)
+		tbl, next = next, tbl
+		base += 16
+		fallthrough
+	case 1:
+		block.Encrypt(next, src[base:])
+		xor.BytesSrc1(dst[base:], src[base:], tbl)
+		tbl, next = next, tbl
+		base += 16
+		fallthrough
+	case 0:
+		xor.BytesSrc0(dst[base:], src[base:], tbl)
+	}
+}
+
+func decryptVariant(block cipher.Block, dst, src, buf []byte) {
+	blocksize := block.BlockSize()
+	tbl := buf[:blocksize]
+	next := buf[blocksize:]
+	block.Encrypt(tbl, initialVector)
+	n := len(src) / blocksize
+	base := 0
+	repeat := n / 8
+	left := n % 8
+	for i := 0; i < repeat; i++ {
+		// 1
+		block.Encrypt(next, src[base:])
+		xor.BytesSrc1(dst[base:], src[base:], tbl)
+		base += blocksize
+
+		// 2
+		block.Encrypt(tbl, src[base:])
+		xor.BytesSrc1(dst[base:], src[base:], next)
+		base += blocksize
+
+		// 3
+		block.Encrypt(next, src[base:])
+		xor.BytesSrc1(dst[base:], src[base:], tbl)
+		base += blocksize
+
+		// 4
+		block.Encrypt(tbl, src[base:])
+		xor.BytesSrc1(dst[base:], src[base:], next)
+		base += blocksize
+
+		// 5
+		block.Encrypt(next, src[base:])
+		xor.BytesSrc1(dst[base:], src[base:], tbl)
+		base += blocksize
+
+		// 6
+		block.Encrypt(tbl, src[base:])
+		xor.BytesSrc1(dst[base:], src[base:], next)
+		base += blocksize
+
+		// 7
+		block.Encrypt(next, src[base:])
+		xor.BytesSrc1(dst[base:], src[base:], tbl)
+		base += blocksize
+
+		// 8
+		block.Encrypt(tbl, src[base:])
+		xor.BytesSrc1(dst[base:], src[base:], next)
+		base += blocksize
+	}
+
+	switch left {
+	case 7:
+		block.Encrypt(next, src[base:])
+		xor.BytesSrc1(dst[base:], src[base:], tbl)
+		tbl, next = next, tbl
+		base += blocksize
+		fallthrough
+	case 6:
+		block.Encrypt(next, src[base:])
+		xor.BytesSrc1(dst[base:], src[base:], tbl)
+		tbl, next = next, tbl
+		base += blocksize
+		fallthrough
+	case 5:
+		block.Encrypt(next, src[base:])
+		xor.BytesSrc1(dst[base:], src[base:], tbl)
+		tbl, next = next, tbl
+		base += blocksize
+		fallthrough
+	case 4:
+		block.Encrypt(next, src[base:])
+		xor.BytesSrc1(dst[base:], src[base:], tbl)
+		tbl, next = next, tbl
+		base += blocksize
+		fallthrough
+	case 3:
+		block.Encrypt(next, src[base:])
+		xor.BytesSrc1(dst[base:], src[base:], tbl)
+		tbl, next = next, tbl
+		base += blocksize
+		fallthrough
+	case 2:
+		block.Encrypt(next, src[base:])
+		xor.BytesSrc1(dst[base:], src[base:], tbl)
+		tbl, next = next, tbl
+		base += blocksize
+		fallthrough
+	case 1:
+		block.Encrypt(next, src[base:])
+		xor.BytesSrc1(dst[base:], src[base:], tbl)
+		tbl, next = next, tbl
+		base += blocksize
+		fallthrough
+	case 0:
+		xor.BytesSrc0(dst[base:], src[base:], tbl)
+	}
+}

kcp-go/entropy.go

@@ -0,0 +1,51 @@
+package kcp
+
+import (
+	"crypto/aes"
+	"crypto/cipher"
+	"crypto/md5"
+	"crypto/rand"
+	"io"
+)
+
+type Entropy interface {
+	Init()
+	Fill(nonce []byte)
+}
+
+// nonceMD5 nonce generator for packet header
+type nonceMD5 struct {
+	seed [md5.Size]byte
+}
+
+func (n *nonceMD5) Init() { /*nothing required*/ }
+
+func (n *nonceMD5) Fill(nonce []byte) {
+	if n.seed[0] == 0 { // entropy update
+		io.ReadFull(rand.Reader, n.seed[:])
+	}
+	n.seed = md5.Sum(n.seed[:])
+	copy(nonce, n.seed[:])
+}
+
+// nonceAES128 nonce generator for packet headers
+type nonceAES128 struct {
+	seed  [aes.BlockSize]byte
+	block cipher.Block
+}
+
+func (n *nonceAES128) Init() {
+	var key [16]byte //aes-128
+	io.ReadFull(rand.Reader, key[:])
+	io.ReadFull(rand.Reader, n.seed[:])
+	block, _ := aes.NewCipher(key[:])
+	n.block = block
+}
+
+func (n *nonceAES128) Fill(nonce []byte) {
+	if n.seed[0] == 0 { // entropy update
+		io.ReadFull(rand.Reader, n.seed[:])
+	}
+	n.block.Encrypt(n.seed[:], n.seed[:])
+	copy(nonce, n.seed[:])
+}

kcp-go/fec.go

@@ -0,0 +1,311 @@
+package kcp
+
+import (
+	"encoding/binary"
+	"sync/atomic"
+
+	"github.com/klauspost/reedsolomon"
+)
+
+const (
+	fecHeaderSize      = 6
+	fecHeaderSizePlus2 = fecHeaderSize + 2 // plus 2B data size
+	typeData           = 0xf1
+	typeFEC            = 0xf2
+)
+
+type (
+	// fecPacket is a decoded FEC packet
+	fecPacket struct {
+		seqid uint32
+		flag  uint16
+		data  []byte
+	}
+
+	// fecDecoder for decoding incoming packets
+	fecDecoder struct {
+		rxlimit      int // queue size limit
+		dataShards   int
+		parityShards int
+		shardSize    int
+		rx           []fecPacket // ordered receive queue
+
+		// caches
+		decodeCache [][]byte
+		flagCache   []bool
+
+		// zeros
+		zeros []byte
+
+		// RS decoder
+		codec reedsolomon.Encoder
+	}
+)
+
+func newFECDecoder(rxlimit, dataShards, parityShards int) *fecDecoder {
+	if dataShards <= 0 || parityShards <= 0 {
+		return nil
+	}
+	if rxlimit < dataShards+parityShards {
+		return nil
+	}
+
+	dec := new(fecDecoder)
+	dec.rxlimit = rxlimit
+	dec.dataShards = dataShards
+	dec.parityShards = parityShards
+	dec.shardSize = dataShards + parityShards
+	codec, err := reedsolomon.New(dataShards, parityShards)
+	if err != nil {
+		return nil
+	}
+	dec.codec = codec
+	dec.decodeCache = make([][]byte, dec.shardSize)
+	dec.flagCache = make([]bool, dec.shardSize)
+	dec.zeros = make([]byte, mtuLimit)
+	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) {
+	// 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)
+			return nil
+		} else if _itimediff(pkt.seqid, dec.rx[i].seqid) > 0 { // insertion
+			insertIdx = i + 1
+			break
+		}
+	}
+
+	// insert into ordered rx queue
+	if insertIdx == n+1 {
+		dec.rx = append(dec.rx, pkt)
+	} else {
+		dec.rx = append(dec.rx, fecPacket{})
+		copy(dec.rx[insertIdx+1:], dec.rx[insertIdx:]) // shift right
+		dec.rx[insertIdx] = pkt
+	}
+
+	// shard range for current packet
+	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))
+	if searchBegin < 0 {
+		searchBegin = 0
+	}
+	searchEnd := searchBegin + dec.shardSize - 1
+	if searchEnd >= len(dec.rx) {
+		searchEnd = len(dec.rx) - 1
+	}
+
+	// re-construct datashards
+	if searchEnd-searchBegin+1 >= dec.dataShards {
+		var numshard, numDataShard, first, maxlen int
+
+		// zero caches
+		shards := dec.decodeCache
+		shardsflag := dec.flagCache
+		for k := range dec.decodeCache {
+			shards[k] = nil
+			shardsflag[k] = false
+		}
+
+		// shard assembly
+		for i := searchBegin; i <= searchEnd; i++ {
+			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
+				shardsflag[seqid%uint32(dec.shardSize)] = true
+				numshard++
+				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 numDataShard == dec.dataShards {
+			// case 1: no loss on data shards
+			dec.rx = dec.freeRange(first, numshard, dec.rx)
+		} else if numshard >= dec.dataShards {
+			// case 2: loss on data shards, but it's recoverable from parity shards
+			for k := range shards {
+				if shards[k] != nil {
+					dlen := len(shards[k])
+					shards[k] = shards[k][:maxlen]
+					copy(shards[k][dlen:], dec.zeros)
+				}
+			}
+			if err := dec.codec.ReconstructData(shards); err == nil {
+				for k := range shards[:dec.dataShards] {
+					if !shardsflag[k] {
+						recovered = append(recovered, shards[k])
+					}
+				}
+			}
+			dec.rx = dec.freeRange(first, numshard, dec.rx)
+		}
+	}
+
+	// keep rxlimit
+	if len(dec.rx) > dec.rxlimit {
+		if dec.rx[0].flag == typeData { // track the unrecoverable data
+			atomic.AddUint64(&DefaultSnmp.FECShortShards, 1)
+		}
+		dec.rx = dec.freeRange(0, 1, dec.rx)
+	}
+	return
+}
+
+// free a range of fecPacket, and zero for GC recycling
+func (dec *fecDecoder) freeRange(first, n int, q []fecPacket) []fecPacket {
+	for i := first; i < first+n; i++ { // recycle buffer
+		xmitBuf.Put(q[i].data)
+	}
+	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]
+}
+
+type (
+	// fecEncoder for encoding outgoing packets
+	fecEncoder struct {
+		dataShards   int
+		parityShards int
+		shardSize    int
+		paws         uint32 // Protect Against Wrapped Sequence numbers
+		next         uint32 // next seqid
+
+		shardCount int // count the number of datashards collected
+		maxSize    int // track maximum data length in datashard
+
+		headerOffset  int // FEC header offset
+		payloadOffset int // FEC payload offset
+
+		// caches
+		shardCache  [][]byte
+		encodeCache [][]byte
+
+		// zeros
+		zeros []byte
+
+		// RS encoder
+		codec reedsolomon.Encoder
+	}
+)
+
+func newFECEncoder(dataShards, parityShards, offset int) *fecEncoder {
+	if dataShards <= 0 || parityShards <= 0 {
+		return nil
+	}
+	enc := new(fecEncoder)
+	enc.dataShards = dataShards
+	enc.parityShards = parityShards
+	enc.shardSize = dataShards + parityShards
+	enc.paws = (0xffffffff/uint32(enc.shardSize) - 1) * uint32(enc.shardSize)
+	enc.headerOffset = offset
+	enc.payloadOffset = enc.headerOffset + fecHeaderSize
+
+	codec, err := reedsolomon.New(dataShards, parityShards)
+	if err != nil {
+		return nil
+	}
+	enc.codec = codec
+
+	// caches
+	enc.encodeCache = make([][]byte, enc.shardSize)
+	enc.shardCache = make([][]byte, enc.shardSize)
+	for k := range enc.shardCache {
+		enc.shardCache[k] = make([]byte, mtuLimit)
+	}
+	enc.zeros = make([]byte, mtuLimit)
+	return enc
+}
+
+// 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) {
+	enc.markData(b[enc.headerOffset:])
+	binary.LittleEndian.PutUint16(b[enc.payloadOffset:], uint16(len(b[enc.payloadOffset:])))
+
+	// copy data to fec datashards
+	sz := len(b)
+	enc.shardCache[enc.shardCount] = enc.shardCache[enc.shardCount][:sz]
+	copy(enc.shardCache[enc.shardCount], b)
+	enc.shardCount++
+
+	// track max datashard length
+	if sz > enc.maxSize {
+		enc.maxSize = sz
+	}
+
+	//  Generation of Reed-Solomon Erasure Code
+	if enc.shardCount == enc.dataShards {
+		// fill '0' into the tail of each datashard
+		for i := 0; i < enc.dataShards; i++ {
+			shard := enc.shardCache[i]
+			slen := len(shard)
+			copy(shard[slen:enc.maxSize], enc.zeros)
+		}
+
+		// construct equal-sized slice with stripped header
+		cache := enc.encodeCache
+		for k := range cache {
+			cache[k] = enc.shardCache[k][enc.payloadOffset:enc.maxSize]
+		}
+
+		// encoding
+		if err := enc.codec.Encode(cache); err == nil {
+			ps = enc.shardCache[enc.dataShards:]
+			for k := range ps {
+				enc.markFEC(ps[k][enc.headerOffset:])
+				ps[k] = ps[k][:enc.maxSize]
+			}
+		}
+
+		// counters resetting
+		enc.shardCount = 0
+		enc.maxSize = 0
+	}
+
+	return
+}
+
+func (enc *fecEncoder) markData(data []byte) {
+	binary.LittleEndian.PutUint32(data, enc.next)
+	binary.LittleEndian.PutUint16(data[4:], typeData)
+	enc.next++
+}
+
+func (enc *fecEncoder) markFEC(data []byte) {
+	binary.LittleEndian.PutUint32(data, enc.next)
+	binary.LittleEndian.PutUint16(data[4:], typeFEC)
+	enc.next = (enc.next + 1) % enc.paws
+}

kcp-go/sess.go

@@ -0,0 +1,990 @@
+package kcp
+
+import (
+	"crypto/rand"
+	"encoding/binary"
+	"github.com/pkg/errors"
+	"golang.org/x/net/ipv4"
+	"hash/crc32"
+	"log"
+	"net"
+	"sync"
+	"sync/atomic"
+	"time"
+)
+
+type errTimeout struct {
+	error
+}
+
+func (errTimeout) Timeout() bool   { return true }
+func (errTimeout) Temporary() bool { return true }
+func (errTimeout) Error() string   { return "i/o timeout" }
+
+const (
+	// 16-bytes nonce for each packet
+	nonceSize = 16
+
+	// 4-bytes packet checksum
+	crcSize = 4
+
+	// overall crypto header size
+	cryptHeaderSize = nonceSize + crcSize
+
+	// maximum packet size
+	mtuLimit = 1500
+
+	// FEC keeps rxFECMulti* (dataShard+parityShard) ordered packets in memory
+	rxFECMulti = 3
+
+	// accept backlog
+	acceptBacklog = 128
+
+	// prerouting(to session) queue
+	qlen = 128
+)
+
+const (
+	errBrokenPipe       = "broken pipe"
+	errInvalidOperation = "invalid operation"
+)
+
+var (
+	// a system-wide packet buffer shared among sending, receiving and FEC
+	// to mitigate high-frequency memory allocation for packets
+	xmitBuf sync.Pool
+)
+
+func init() {
+	xmitBuf.New = func() interface{} {
+		return make([]byte, mtuLimit)
+	}
+}
+
+type (
+	// UDPSession defines a KCP session implemented by UDP
+	UDPSession struct {
+		updaterIdx int            // record slice index in updater
+		conn       net.PacketConn // the underlying packet connection
+		kcp        *KCP           // KCP ARQ protocol
+		l          *Listener      // pointing to the Listener object if it's been accepted by a Listener
+		block      BlockCrypt     // block encryption object
+
+		// kcp receiving is based on packets
+		// recvbuf turns packets into stream
+		recvbuf []byte
+		bufptr  []byte
+		// header extended output buffer, if has header
+		ext []byte
+
+		// FEC codec
+		fecDecoder *fecDecoder
+		fecEncoder *fecEncoder
+
+		// settings
+		remote     net.Addr  // remote peer address
+		rd         time.Time // read deadline
+		wd         time.Time // write deadline
+		headerSize int       // the header size additional to a KCP frame
+		ackNoDelay bool      // send ack immediately for each incoming packet(testing purpose)
+		writeDelay bool      // delay kcp.flush() for Write() for bulk transfer
+		dup        int       // duplicate udp packets(testing purpose)
+
+		// notifications
+		die          chan struct{} // notify current session has Closed
+		chReadEvent  chan struct{} // notify Read() can be called without blocking
+		chWriteEvent chan struct{} // notify Write() can be called without blocking
+		chErrorEvent chan error    // notify Read() have an error
+
+		// nonce generator
+		nonce Entropy
+
+		isClosed bool // flag the session has Closed
+		mu       sync.Mutex
+	}
+
+	setReadBuffer interface {
+		SetReadBuffer(bytes int) error
+	}
+
+	setWriteBuffer interface {
+		SetWriteBuffer(bytes int) error
+	}
+)
+
+// newUDPSession create a new udp session for client or server
+func newUDPSession(conv uint32, dataShards, parityShards int, l *Listener, conn net.PacketConn, remote net.Addr, block BlockCrypt) *UDPSession {
+	sess := new(UDPSession)
+	sess.die = make(chan struct{})
+	sess.nonce = new(nonceAES128)
+	sess.nonce.Init()
+	sess.chReadEvent = make(chan struct{}, 1)
+	sess.chWriteEvent = make(chan struct{}, 1)
+	sess.chErrorEvent = make(chan error, 1)
+	sess.remote = remote
+	sess.conn = conn
+	sess.l = l
+	sess.block = block
+	sess.recvbuf = make([]byte, mtuLimit)
+
+	// FEC codec initialization
+	sess.fecDecoder = newFECDecoder(rxFECMulti*(dataShards+parityShards), dataShards, parityShards)
+	if sess.block != nil {
+		sess.fecEncoder = newFECEncoder(dataShards, parityShards, cryptHeaderSize)
+	} else {
+		sess.fecEncoder = newFECEncoder(dataShards, parityShards, 0)
+	}
+
+	// calculate additional header size introduced by FEC and encryption
+	if sess.block != nil {
+		sess.headerSize += cryptHeaderSize
+	}
+	if sess.fecEncoder != nil {
+		sess.headerSize += fecHeaderSizePlus2
+	}
+
+	// we only need to allocate extended packet buffer if we have the additional header
+	if sess.headerSize > 0 {
+		sess.ext = make([]byte, mtuLimit)
+	}
+
+	sess.kcp = NewKCP(conv, func(buf []byte, size int) {
+		if size >= IKCP_OVERHEAD {
+			sess.output(buf[:size])
+		}
+	})
+	sess.kcp.SetMtu(IKCP_MTU_DEF - sess.headerSize)
+
+	// register current session to the global updater,
+	// which call sess.update() periodically.
+	updater.addSession(sess)
+
+	if sess.l == nil { // it's a client connection
+		go sess.readLoop()
+		atomic.AddUint64(&DefaultSnmp.ActiveOpens, 1)
+	} else {
+		atomic.AddUint64(&DefaultSnmp.PassiveOpens, 1)
+	}
+	currestab := atomic.AddUint64(&DefaultSnmp.CurrEstab, 1)
+	maxconn := atomic.LoadUint64(&DefaultSnmp.MaxConn)
+	if currestab > maxconn {
+		atomic.CompareAndSwapUint64(&DefaultSnmp.MaxConn, maxconn, currestab)
+	}
+
+	return sess
+}
+
+// Read implements net.Conn
+func (s *UDPSession) Read(b []byte) (n int, err error) {
+	for {
+		s.mu.Lock()
+		if len(s.bufptr) > 0 { // copy from buffer into b
+			n = copy(b, s.bufptr)
+			s.bufptr = s.bufptr[n:]
+			s.mu.Unlock()
+			return n, nil
+		}
+
+		if s.isClosed {
+			s.mu.Unlock()
+			return 0, errors.New(errBrokenPipe)
+		}
+
+		if size := s.kcp.PeekSize(); size > 0 { // peek data size from kcp
+			atomic.AddUint64(&DefaultSnmp.BytesReceived, uint64(size))
+			if len(b) >= size { // receive data into 'b' directly
+				s.kcp.Recv(b)
+				s.mu.Unlock()
+				return size, nil
+			}
+
+			// if necessary resize the stream buffer to guarantee a sufficent buffer space
+			if cap(s.recvbuf) < size {
+				s.recvbuf = make([]byte, size)
+			}
+
+			// resize the length of recvbuf to correspond to data size
+			s.recvbuf = s.recvbuf[:size]
+			s.kcp.Recv(s.recvbuf)
+			n = copy(b, s.recvbuf)   // copy to 'b'
+			s.bufptr = s.recvbuf[n:] // pointer update
+			s.mu.Unlock()
+			return n, nil
+		}
+
+		// deadline for current reading operation
+		var timeout *time.Timer
+		var c <-chan time.Time
+		if !s.rd.IsZero() {
+			if time.Now().After(s.rd) {
+				s.mu.Unlock()
+				return 0, errTimeout{}
+			}
+
+			delay := s.rd.Sub(time.Now())
+			timeout = time.NewTimer(delay)
+			c = timeout.C
+		}
+		s.mu.Unlock()
+
+		// wait for read event or timeout
+		select {
+		case <-s.chReadEvent:
+		case <-c:
+		case <-s.die:
+		case err = <-s.chErrorEvent:
+			if timeout != nil {
+				timeout.Stop()
+			}
+			return n, err
+		}
+
+		if timeout != nil {
+			timeout.Stop()
+		}
+	}
+}
+
+// Write implements net.Conn
+func (s *UDPSession) Write(b []byte) (n int, err error) {
+	for {
+		s.mu.Lock()
+		if s.isClosed {
+			s.mu.Unlock()
+			return 0, errors.New(errBrokenPipe)
+		}
+
+		// controls how much data will be sent to kcp core
+		// to prevent the memory from exhuasting
+		if s.kcp.WaitSnd() < int(s.kcp.snd_wnd) {
+			n = len(b)
+			for {
+				if len(b) <= int(s.kcp.mss) {
+					s.kcp.Send(b)
+					break
+				} else {
+					s.kcp.Send(b[:s.kcp.mss])
+					b = b[s.kcp.mss:]
+				}
+			}
+
+			// flush immediately if the queue is full
+			if s.kcp.WaitSnd() >= int(s.kcp.snd_wnd) || !s.writeDelay {
+				s.kcp.flush(false)
+			}
+			s.mu.Unlock()
+			atomic.AddUint64(&DefaultSnmp.BytesSent, uint64(n))
+			return n, nil
+		}
+
+		// deadline for current writing operation
+		var timeout *time.Timer
+		var c <-chan time.Time
+		if !s.wd.IsZero() {
+			if time.Now().After(s.wd) {
+				s.mu.Unlock()
+				return 0, errTimeout{}
+			}
+			delay := s.wd.Sub(time.Now())
+			timeout = time.NewTimer(delay)
+			c = timeout.C
+		}
+		s.mu.Unlock()
+
+		// wait for write event or timeout
+		select {
+		case <-s.chWriteEvent:
+		case <-c:
+		case <-s.die:
+		}
+
+		if timeout != nil {
+			timeout.Stop()
+		}
+	}
+}
+
+// Close closes the connection.
+func (s *UDPSession) Close() error {
+	// remove current session from updater & listener(if necessary)
+	updater.removeSession(s)
+	if s.l != nil { // notify listener
+		s.l.closeSession(s.remote)
+	}
+
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	if s.isClosed {
+		return errors.New(errBrokenPipe)
+	}
+	close(s.die)
+	s.isClosed = true
+	atomic.AddUint64(&DefaultSnmp.CurrEstab, ^uint64(0))
+	if s.l == nil { // client socket close
+		//return s.conn.Close()
+	}
+	return nil
+}
+
+// LocalAddr returns the local network address. The Addr returned is shared by all invocations of LocalAddr, so do not modify it.
+func (s *UDPSession) LocalAddr() net.Addr { return s.conn.LocalAddr() }
+
+// RemoteAddr returns the remote network address. The Addr returned is shared by all invocations of RemoteAddr, so do not modify it.
+func (s *UDPSession) RemoteAddr() net.Addr { return s.remote }
+
+// SetDeadline sets the deadline associated with the listener. A zero time value disables the deadline.
+func (s *UDPSession) SetDeadline(t time.Time) error {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	s.rd = t
+	s.wd = t
+	s.notifyReadEvent()
+	s.notifyWriteEvent()
+	return nil
+}
+
+// SetReadDeadline implements the Conn SetReadDeadline method.
+func (s *UDPSession) SetReadDeadline(t time.Time) error {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	s.rd = t
+	s.notifyReadEvent()
+	return nil
+}
+
+// SetWriteDeadline implements the Conn SetWriteDeadline method.
+func (s *UDPSession) SetWriteDeadline(t time.Time) error {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	s.wd = t
+	s.notifyWriteEvent()
+	return nil
+}
+
+// SetWriteDelay delays write for bulk transfer until the next update interval
+func (s *UDPSession) SetWriteDelay(delay bool) {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	s.writeDelay = delay
+}
+
+// SetWindowSize set maximum window size
+func (s *UDPSession) SetWindowSize(sndwnd, rcvwnd int) {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	s.kcp.WndSize(sndwnd, rcvwnd)
+}
+
+// SetMtu sets the maximum transmission unit(not including UDP header)
+func (s *UDPSession) SetMtu(mtu int) bool {
+	if mtu > mtuLimit {
+		return false
+	}
+
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	s.kcp.SetMtu(mtu - s.headerSize)
+	return true
+}
+
+// SetStreamMode toggles the stream mode on/off
+func (s *UDPSession) SetStreamMode(enable bool) {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	if enable {
+		s.kcp.stream = 1
+	} else {
+		s.kcp.stream = 0
+	}
+}
+
+// SetACKNoDelay changes ack flush option, set true to flush ack immediately,
+func (s *UDPSession) SetACKNoDelay(nodelay bool) {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	s.ackNoDelay = nodelay
+}
+
+// SetDUP duplicates udp packets for kcp output, for testing purpose only
+func (s *UDPSession) SetDUP(dup int) {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	s.dup = dup
+}
+
+// SetNoDelay calls nodelay() of kcp
+// https://github.com/skywind3000/kcp/blob/master/README.en.md#protocol-configuration
+func (s *UDPSession) SetNoDelay(nodelay, interval, resend, nc int) {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	s.kcp.NoDelay(nodelay, interval, resend, nc)
+}
+
+// SetDSCP sets the 6bit DSCP field of IP header, no effect if it's accepted from Listener
+func (s *UDPSession) SetDSCP(dscp int) error {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	if s.l == nil {
+		if nc, ok := s.conn.(*connectedUDPConn); ok {
+			return ipv4.NewConn(nc.UDPConn).SetTOS(dscp << 2)
+		} else if nc, ok := s.conn.(net.Conn); ok {
+			return ipv4.NewConn(nc).SetTOS(dscp << 2)
+		}
+	}
+	return errors.New(errInvalidOperation)
+}
+
+// SetReadBuffer sets the socket read buffer, no effect if it's accepted from Listener
+func (s *UDPSession) SetReadBuffer(bytes int) error {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	if s.l == nil {
+		if nc, ok := s.conn.(setReadBuffer); ok {
+			return nc.SetReadBuffer(bytes)
+		}
+	}
+	return errors.New(errInvalidOperation)
+}
+
+// SetWriteBuffer sets the socket write buffer, no effect if it's accepted from Listener
+func (s *UDPSession) SetWriteBuffer(bytes int) error {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	if s.l == nil {
+		if nc, ok := s.conn.(setWriteBuffer); ok {
+			return nc.SetWriteBuffer(bytes)
+		}
+	}
+	return errors.New(errInvalidOperation)
+}
+
+// post-processing for sending a packet from kcp core
+// steps:
+// 0. Header extending
+// 1. FEC packet generation
+// 2. CRC32 integrity
+// 3. Encryption
+// 4. WriteTo kernel
+func (s *UDPSession) output(buf []byte) {
+	var ecc [][]byte
+
+	// 0. extend buf's header space(if necessary)
+	ext := buf
+	if s.headerSize > 0 {
+		ext = s.ext[:s.headerSize+len(buf)]
+		copy(ext[s.headerSize:], buf)
+	}
+
+	// 1. FEC encoding
+	if s.fecEncoder != nil {
+		ecc = s.fecEncoder.encode(ext)
+	}
+
+	// 2&3. crc32 & encryption
+	if s.block != nil {
+		s.nonce.Fill(ext[:nonceSize])
+		checksum := crc32.ChecksumIEEE(ext[cryptHeaderSize:])
+		binary.LittleEndian.PutUint32(ext[nonceSize:], checksum)
+		s.block.Encrypt(ext, ext)
+
+		for k := range ecc {
+			s.nonce.Fill(ecc[k][:nonceSize])
+			checksum := crc32.ChecksumIEEE(ecc[k][cryptHeaderSize:])
+			binary.LittleEndian.PutUint32(ecc[k][nonceSize:], checksum)
+			s.block.Encrypt(ecc[k], ecc[k])
+		}
+	}
+
+	// 4. WriteTo kernel
+	nbytes := 0
+	npkts := 0
+	for i := 0; i < s.dup+1; i++ {
+		if n, err := s.conn.WriteTo(ext, s.remote); err == nil {
+			nbytes += n
+			npkts++
+		}
+	}
+
+	for k := range ecc {
+		if n, err := s.conn.WriteTo(ecc[k], s.remote); err == nil {
+			nbytes += n
+			npkts++
+		}
+	}
+	atomic.AddUint64(&DefaultSnmp.OutPkts, uint64(npkts))
+	atomic.AddUint64(&DefaultSnmp.OutBytes, uint64(nbytes))
+}
+
+// kcp update, returns interval for next calling
+func (s *UDPSession) update() (interval time.Duration) {
+	s.mu.Lock()
+	waitsnd := s.kcp.WaitSnd()
+	interval = time.Duration(s.kcp.flush(false)) * time.Millisecond
+	if s.kcp.WaitSnd() < waitsnd {
+		s.notifyWriteEvent()
+	}
+	s.mu.Unlock()
+	return
+}
+
+// GetConv gets conversation id of a session
+func (s *UDPSession) GetConv() uint32 { return s.kcp.conv }
+
+func (s *UDPSession) notifyReadEvent() {
+	select {
+	case s.chReadEvent <- struct{}{}:
+	default:
+	}
+}
+
+func (s *UDPSession) notifyWriteEvent() {
+	select {
+	case s.chWriteEvent <- struct{}{}:
+	default:
+	}
+}
+
+func (s *UDPSession) kcpInput(data []byte) {
+	var kcpInErrors, fecErrs, fecRecovered, fecParityShards uint64
+
+	if s.fecDecoder != nil {
+		if len(data) > fecHeaderSize { // must be larger than fec header size
+			f := s.fecDecoder.decodeBytes(data)
+			if f.flag == typeData || f.flag == typeFEC { // header check
+				if f.flag == typeFEC {
+					fecParityShards++
+				}
+				recovers := s.fecDecoder.decode(f)
+
+				s.mu.Lock()
+				waitsnd := s.kcp.WaitSnd()
+				if f.flag == typeData {
+					if ret := s.kcp.Input(data[fecHeaderSizePlus2:], true, s.ackNoDelay); ret != 0 {
+						kcpInErrors++
+					}
+				}
+
+				for _, r := range recovers {
+					if len(r) >= 2 { // must be larger than 2bytes
+						sz := binary.LittleEndian.Uint16(r)
+						if int(sz) <= len(r) && sz >= 2 {
+							if ret := s.kcp.Input(r[2:sz], false, s.ackNoDelay); ret == 0 {
+								fecRecovered++
+							} else {
+								kcpInErrors++
+							}
+						} else {
+							fecErrs++
+						}
+					} else {
+						fecErrs++
+					}
+				}
+
+				// to notify the readers to receive the data
+				if n := s.kcp.PeekSize(); n > 0 {
+					s.notifyReadEvent()
+				}
+				// to notify the writers when queue is shorter(e.g. ACKed)
+				if s.kcp.WaitSnd() < waitsnd {
+					s.notifyWriteEvent()
+				}
+				s.mu.Unlock()
+			} else {
+				atomic.AddUint64(&DefaultSnmp.InErrs, 1)
+			}
+		} else {
+			atomic.AddUint64(&DefaultSnmp.InErrs, 1)
+		}
+	} else {
+		s.mu.Lock()
+		waitsnd := s.kcp.WaitSnd()
+		if ret := s.kcp.Input(data, true, s.ackNoDelay); ret != 0 {
+			kcpInErrors++
+		}
+		if n := s.kcp.PeekSize(); n > 0 {
+			s.notifyReadEvent()
+		}
+		if s.kcp.WaitSnd() < waitsnd {
+			s.notifyWriteEvent()
+		}
+		s.mu.Unlock()
+	}
+
+	atomic.AddUint64(&DefaultSnmp.InPkts, 1)
+	atomic.AddUint64(&DefaultSnmp.InBytes, uint64(len(data)))
+	if fecParityShards > 0 {
+		atomic.AddUint64(&DefaultSnmp.FECParityShards, fecParityShards)
+	}
+	if kcpInErrors > 0 {
+		atomic.AddUint64(&DefaultSnmp.KCPInErrors, kcpInErrors)
+	}
+	if fecErrs > 0 {
+		atomic.AddUint64(&DefaultSnmp.FECErrs, fecErrs)
+	}
+	if fecRecovered > 0 {
+		atomic.AddUint64(&DefaultSnmp.FECRecovered, fecRecovered)
+	}
+}
+
+func (s *UDPSession) receiver(ch chan<- inPacket) {
+	for {
+		data := xmitBuf.Get().([]byte)[:mtuLimit]
+		if n, from, err := s.conn.ReadFrom(data); err == nil && n >= s.headerSize+IKCP_OVERHEAD {
+			select {
+			case ch <- inPacket{from, data[:n]}:
+			case <-s.die:
+				return
+			}
+		} else if err != nil {
+			s.chErrorEvent <- err
+			return
+		} else {
+			atomic.AddUint64(&DefaultSnmp.InErrs, 1)
+		}
+	}
+}
+
+// the read loop for a client session
+func (s *UDPSession) readLoop() {
+	chPacket := make(chan inPacket, qlen)
+	go s.receiver(chPacket)
+	firstPacket := true
+	for {
+		select {
+		case p := <-chPacket:
+			raw  := p.data
+			data := p.data
+			from := p.from
+			dataValid := false
+			if firstPacket{
+				log.Println("firstPacket from", from.String())
+			}
+			if s.block != nil {
+				s.block.Decrypt(data, data)
+				data = data[nonceSize:]
+				checksum := crc32.ChecksumIEEE(data[crcSize:])
+				if checksum == binary.LittleEndian.Uint32(data) {
+					data = data[crcSize:]
+					dataValid = true
+				} else {
+					atomic.AddUint64(&DefaultSnmp.InCsumErrors, 1)
+				}
+			} else if s.block == nil {
+				dataValid = true
+			}
+
+			if dataValid {
+				if firstPacket{
+					log.Println("firstPacket valided", from.String())
+					firstPacket = false
+					//remote upd ip may change
+					s.remote = from
+				}
+				s.kcpInput(data)
+			}
+			xmitBuf.Put(raw)
+		case <-s.die:
+			return
+		}
+	}
+}
+
+type (
+	// Listener defines a server which will be waiting to accept incoming connections
+	Listener struct {
+		block        BlockCrypt     // block encryption
+		dataShards   int            // FEC data shard
+		parityShards int            // FEC parity shard
+		fecDecoder   *fecDecoder    // FEC mock initialization
+		conn         net.PacketConn // the underlying packet connection
+
+		sessions        map[string]*UDPSession // all sessions accepted by this Listener
+		chAccepts       chan *UDPSession       // Listen() backlog
+		chSessionClosed chan net.Addr          // session close queue
+		headerSize      int                    // the additional header to a KCP frame
+		die             chan struct{}          // notify the listener has closed
+		rd              atomic.Value           // read deadline for Accept()
+		wd              atomic.Value
+	}
+
+	// a incoming packet definition
+	inPacket struct {
+		from net.Addr
+		data []byte
+	}
+)
+
+// monitor incoming data for all connections of server
+func (l *Listener) monitor() {
+	// a cache for session object last used
+	var lastAddr string
+	var lastSession *UDPSession
+
+	chPacket := make(chan inPacket, qlen)
+	go l.receiver(chPacket)
+	for {
+		select {
+		case p := <-chPacket:
+			raw := p.data
+			data := p.data
+			from := p.from
+			dataValid := false
+			if l.block != nil {
+				l.block.Decrypt(data, data)
+				data = data[nonceSize:]
+				checksum := crc32.ChecksumIEEE(data[crcSize:])
+				if checksum == binary.LittleEndian.Uint32(data) {
+					data = data[crcSize:]
+					dataValid = true
+				} else {
+					atomic.AddUint64(&DefaultSnmp.InCsumErrors, 1)
+				}
+			} else if l.block == nil {
+				dataValid = true
+			}
+
+			if dataValid {
+				addr := from.String()
+				var s *UDPSession
+				var ok bool
+
+				// the packets received from an address always come in batch,
+				// cache the session for next packet, without querying map.
+				if addr == lastAddr {
+					s, ok = lastSession, true
+				} else if s, ok = l.sessions[addr]; ok {
+					lastSession = s
+					lastAddr = addr
+				}
+
+				if !ok { // new session
+					if len(l.chAccepts) < cap(l.chAccepts) { // do not let the new sessions overwhelm accept queue
+						var conv uint32
+						convValid := false
+						if l.fecDecoder != nil {
+							isfec := binary.LittleEndian.Uint16(data[4:])
+							if isfec == typeData {
+								conv = binary.LittleEndian.Uint32(data[fecHeaderSizePlus2:])
+								convValid = true
+							}
+						} else {
+							conv = binary.LittleEndian.Uint32(data)
+							convValid = true
+						}
+
+						if convValid { // creates a new session only if the 'conv' field in kcp is accessible
+							s := newUDPSession(conv, l.dataShards, l.parityShards, l, l.conn, from, l.block)
+							s.kcpInput(data)
+							l.sessions[addr] = s
+							l.chAccepts <- s
+						}
+					}
+				} else {
+					s.kcpInput(data)
+				}
+			}
+
+			xmitBuf.Put(raw)
+		case deadlink := <-l.chSessionClosed:
+			delete(l.sessions, deadlink.String())
+		case <-l.die:
+			return
+		}
+	}
+}
+
+func (l *Listener) receiver(ch chan<- inPacket) {
+	for {
+		data := xmitBuf.Get().([]byte)[:mtuLimit]
+		if n, from, err := l.conn.ReadFrom(data); err == nil && n >= l.headerSize+IKCP_OVERHEAD {
+			select {
+			case ch <- inPacket{from, data[:n]}:
+			case <-l.die:
+				return
+			}
+		} else if err != nil {
+			return
+		} else {
+			atomic.AddUint64(&DefaultSnmp.InErrs, 1)
+		}
+	}
+}
+
+// SetReadBuffer sets the socket read buffer for the Listener
+func (l *Listener) SetReadBuffer(bytes int) error {
+	if nc, ok := l.conn.(setReadBuffer); ok {
+		return nc.SetReadBuffer(bytes)
+	}
+	return errors.New(errInvalidOperation)
+}
+
+// SetWriteBuffer sets the socket write buffer for the Listener
+func (l *Listener) SetWriteBuffer(bytes int) error {
+	if nc, ok := l.conn.(setWriteBuffer); ok {
+		return nc.SetWriteBuffer(bytes)
+	}
+	return errors.New(errInvalidOperation)
+}
+
+// SetDSCP sets the 6bit DSCP field of IP header
+func (l *Listener) SetDSCP(dscp int) error {
+	if nc, ok := l.conn.(net.Conn); ok {
+		return ipv4.NewConn(nc).SetTOS(dscp << 2)
+	}
+	return errors.New(errInvalidOperation)
+}
+
+// Accept implements the Accept method in the Listener interface; it waits for the next call and returns a generic Conn.
+func (l *Listener) Accept() (net.Conn, error) {
+	return l.AcceptKCP()
+}
+
+// AcceptKCP accepts a KCP connection
+func (l *Listener) AcceptKCP() (*UDPSession, error) {
+	var timeout <-chan time.Time
+	if tdeadline, ok := l.rd.Load().(time.Time); ok && !tdeadline.IsZero() {
+		timeout = time.After(tdeadline.Sub(time.Now()))
+	}
+
+	select {
+	case <-timeout:
+		return nil, &errTimeout{}
+	case c := <-l.chAccepts:
+		return c, nil
+	case <-l.die:
+		return nil, errors.New(errBrokenPipe)
+	}
+}
+
+// SetDeadline sets the deadline associated with the listener. A zero time value disables the deadline.
+func (l *Listener) SetDeadline(t time.Time) error {
+	l.SetReadDeadline(t)
+	l.SetWriteDeadline(t)
+	return nil
+}
+
+// SetReadDeadline implements the Conn SetReadDeadline method.
+func (l *Listener) SetReadDeadline(t time.Time) error {
+	l.rd.Store(t)
+	return nil
+}
+
+// SetWriteDeadline implements the Conn SetWriteDeadline method.
+func (l *Listener) SetWriteDeadline(t time.Time) error {
+	l.wd.Store(t)
+	return nil
+}
+
+// Close stops listening on the UDP address. Already Accepted connections are not closed.
+func (l *Listener) Close() error {
+	close(l.die)
+	return l.conn.Close()
+}
+
+// closeSession notify the listener that a session has closed
+func (l *Listener) closeSession(remote net.Addr) bool {
+	select {
+	case l.chSessionClosed <- remote:
+		return true
+	case <-l.die:
+		return false
+	}
+}
+
+// Addr returns the listener's network address, The Addr returned is shared by all invocations of Addr, so do not modify it.
+func (l *Listener) Addr() net.Addr { return l.conn.LocalAddr() }
+
+// Listen listens for incoming KCP packets addressed to the local address laddr on the network "udp",
+func Listen(laddr string) (net.Listener, error) { return ListenWithOptions(laddr, nil, 0, 0) }
+
+// ListenWithOptions listens for incoming KCP packets addressed to the local address laddr on the network "udp" with packet encryption,
+// dataShards, parityShards defines Reed-Solomon Erasure Coding parameters
+func ListenWithOptions(laddr string, block BlockCrypt, dataShards, parityShards int) (*Listener, error) {
+	udpaddr, err := net.ResolveUDPAddr("udp", laddr)
+	if err != nil {
+		return nil, errors.Wrap(err, "net.ResolveUDPAddr")
+	}
+	conn, err := net.ListenUDP("udp", udpaddr)
+	if err != nil {
+		return nil, errors.Wrap(err, "net.ListenUDP")
+	}
+
+	return ServeConn(block, dataShards, parityShards, conn)
+}
+
+// ServeConn serves KCP protocol for a single packet connection.
+func ServeConn(block BlockCrypt, dataShards, parityShards int, conn net.PacketConn) (*Listener, error) {
+	l := new(Listener)
+	l.conn = conn
+	l.sessions = make(map[string]*UDPSession)
+	l.chAccepts = make(chan *UDPSession, acceptBacklog)
+	l.chSessionClosed = make(chan net.Addr)
+	l.die = make(chan struct{})
+	l.dataShards = dataShards
+	l.parityShards = parityShards
+	l.block = block
+	l.fecDecoder = newFECDecoder(rxFECMulti*(dataShards+parityShards), dataShards, parityShards)
+
+	// calculate header size
+	if l.block != nil {
+		l.headerSize += cryptHeaderSize
+	}
+	if l.fecDecoder != nil {
+		l.headerSize += fecHeaderSizePlus2
+	}
+
+	go l.monitor()
+	return l, nil
+}
+
+// Dial connects to the remote address "raddr" on the network "udp"
+func Dial(raddr string) (net.Conn, error) { return DialWithOptions(raddr, nil, 0, 0) }
+
+// DialWithOptions connects to the remote address "raddr" on the network "udp" with packet encryption
+func DialWithOptions(raddr string, block BlockCrypt, dataShards, parityShards int) (*UDPSession, error) {
+	udpaddr, err := net.ResolveUDPAddr("udp", raddr)
+	if err != nil {
+		return nil, errors.Wrap(err, "net.ResolveUDPAddr")
+	}
+
+	udpconn, err := net.DialUDP("udp", nil, udpaddr)
+	if err != nil {
+		return nil, errors.Wrap(err, "net.DialUDP")
+	}
+
+	return NewConn(raddr, block, dataShards, parityShards, &connectedUDPConn{udpconn})
+}
+
+// NewConn establishes a session and talks KCP protocol over a packet connection.
+func NewConn(raddr string, block BlockCrypt, dataShards, parityShards int, conn net.PacketConn) (*UDPSession, error) {
+	udpaddr, err := net.ResolveUDPAddr("udp", raddr)
+	if err != nil {
+		return nil, errors.Wrap(err, "net.ResolveUDPAddr")
+	}
+
+	var convid uint32
+	binary.Read(rand.Reader, binary.LittleEndian, &convid)
+	return newUDPSession(convid, dataShards, parityShards, nil, conn, udpaddr, block), nil
+}
+func NewP2pConn(udpConn net.PacketConn, raddr string, block BlockCrypt, dataShards, parityShards int) (*UDPSession, error){
+	udpaddr, err := net.ResolveUDPAddr("udp", raddr)
+	if err != nil {
+		return nil, errors.Wrap(err, "net.ResolveUDPAddr")
+	}
+
+	return newUDPSession(0x1, dataShards, parityShards, nil, udpConn, udpaddr, block), nil
+}
+// monotonic reference time point
+var refTime time.Time = time.Now()
+
+// currentMs returns current elasped monotonic milliseconds since program startup
+func currentMs() uint32 { return uint32(time.Now().Sub(refTime) / time.Millisecond) }
+
+// connectedUDPConn is a wrapper for net.UDPConn which converts WriteTo syscalls
+// to Write syscalls that are 4 times faster on some OS'es. This should only be
+// used for connections that were produced by a net.Dial* call.
+type connectedUDPConn struct{ *net.UDPConn }
+
+// WriteTo redirects all writes to the Write syscall, which is 4 times faster.
+func (c *connectedUDPConn) WriteTo(b []byte, addr net.Addr) (int, error) { return c.Write(b) }

kcp-go/snmp.go

@@ -0,0 +1,164 @@
+package kcp
+
+import (
+	"fmt"
+	"sync/atomic"
+)
+
+// Snmp defines network statistics indicator
+type Snmp struct {
+	BytesSent        uint64 // bytes sent from upper level
+	BytesReceived    uint64 // bytes received to upper level
+	MaxConn          uint64 // max number of connections ever reached
+	ActiveOpens      uint64 // accumulated active open connections
+	PassiveOpens     uint64 // accumulated passive open connections
+	CurrEstab        uint64 // current number of established connections
+	InErrs           uint64 // UDP read errors reported from net.PacketConn
+	InCsumErrors     uint64 // checksum errors from CRC32
+	KCPInErrors      uint64 // packet iput errors reported from KCP
+	InPkts           uint64 // incoming packets count
+	OutPkts          uint64 // outgoing packets count
+	InSegs           uint64 // incoming KCP segments
+	OutSegs          uint64 // outgoing KCP segments
+	InBytes          uint64 // UDP bytes received
+	OutBytes         uint64 // UDP bytes sent
+	RetransSegs      uint64 // accmulated retransmited segments
+	FastRetransSegs  uint64 // accmulated fast retransmitted segments
+	EarlyRetransSegs uint64 // accmulated early retransmitted segments
+	LostSegs         uint64 // number of segs infered as lost
+	RepeatSegs       uint64 // number of segs duplicated
+	FECRecovered     uint64 // correct packets recovered from FEC
+	FECErrs          uint64 // incorrect packets recovered from FEC
+	FECParityShards  uint64 // FEC segments received
+	FECShortShards   uint64 // number of data shards that's not enough for recovery
+}
+
+func newSnmp() *Snmp {
+	return new(Snmp)
+}
+
+// Header returns all field names
+func (s *Snmp) Header() []string {
+	return []string{
+		"BytesSent",
+		"BytesReceived",
+		"MaxConn",
+		"ActiveOpens",
+		"PassiveOpens",
+		"CurrEstab",
+		"InErrs",
+		"InCsumErrors",
+		"KCPInErrors",
+		"InPkts",
+		"OutPkts",
+		"InSegs",
+		"OutSegs",
+		"InBytes",
+		"OutBytes",
+		"RetransSegs",
+		"FastRetransSegs",
+		"EarlyRetransSegs",
+		"LostSegs",
+		"RepeatSegs",
+		"FECParityShards",
+		"FECErrs",
+		"FECRecovered",
+		"FECShortShards",
+	}
+}
+
+// ToSlice returns current snmp info as slice
+func (s *Snmp) ToSlice() []string {
+	snmp := s.Copy()
+	return []string{
+		fmt.Sprint(snmp.BytesSent),
+		fmt.Sprint(snmp.BytesReceived),
+		fmt.Sprint(snmp.MaxConn),
+		fmt.Sprint(snmp.ActiveOpens),
+		fmt.Sprint(snmp.PassiveOpens),
+		fmt.Sprint(snmp.CurrEstab),
+		fmt.Sprint(snmp.InErrs),
+		fmt.Sprint(snmp.InCsumErrors),
+		fmt.Sprint(snmp.KCPInErrors),
+		fmt.Sprint(snmp.InPkts),
+		fmt.Sprint(snmp.OutPkts),
+		fmt.Sprint(snmp.InSegs),
+		fmt.Sprint(snmp.OutSegs),
+		fmt.Sprint(snmp.InBytes),
+		fmt.Sprint(snmp.OutBytes),
+		fmt.Sprint(snmp.RetransSegs),
+		fmt.Sprint(snmp.FastRetransSegs),
+		fmt.Sprint(snmp.EarlyRetransSegs),
+		fmt.Sprint(snmp.LostSegs),
+		fmt.Sprint(snmp.RepeatSegs),
+		fmt.Sprint(snmp.FECParityShards),
+		fmt.Sprint(snmp.FECErrs),
+		fmt.Sprint(snmp.FECRecovered),
+		fmt.Sprint(snmp.FECShortShards),
+	}
+}
+
+// Copy make a copy of current snmp snapshot
+func (s *Snmp) Copy() *Snmp {
+	d := newSnmp()
+	d.BytesSent = atomic.LoadUint64(&s.BytesSent)
+	d.BytesReceived = atomic.LoadUint64(&s.BytesReceived)
+	d.MaxConn = atomic.LoadUint64(&s.MaxConn)
+	d.ActiveOpens = atomic.LoadUint64(&s.ActiveOpens)
+	d.PassiveOpens = atomic.LoadUint64(&s.PassiveOpens)
+	d.CurrEstab = atomic.LoadUint64(&s.CurrEstab)
+	d.InErrs = atomic.LoadUint64(&s.InErrs)
+	d.InCsumErrors = atomic.LoadUint64(&s.InCsumErrors)
+	d.KCPInErrors = atomic.LoadUint64(&s.KCPInErrors)
+	d.InPkts = atomic.LoadUint64(&s.InPkts)
+	d.OutPkts = atomic.LoadUint64(&s.OutPkts)
+	d.InSegs = atomic.LoadUint64(&s.InSegs)
+	d.OutSegs = atomic.LoadUint64(&s.OutSegs)
+	d.InBytes = atomic.LoadUint64(&s.InBytes)
+	d.OutBytes = atomic.LoadUint64(&s.OutBytes)
+	d.RetransSegs = atomic.LoadUint64(&s.RetransSegs)
+	d.FastRetransSegs = atomic.LoadUint64(&s.FastRetransSegs)
+	d.EarlyRetransSegs = atomic.LoadUint64(&s.EarlyRetransSegs)
+	d.LostSegs = atomic.LoadUint64(&s.LostSegs)
+	d.RepeatSegs = atomic.LoadUint64(&s.RepeatSegs)
+	d.FECParityShards = atomic.LoadUint64(&s.FECParityShards)
+	d.FECErrs = atomic.LoadUint64(&s.FECErrs)
+	d.FECRecovered = atomic.LoadUint64(&s.FECRecovered)
+	d.FECShortShards = atomic.LoadUint64(&s.FECShortShards)
+	return d
+}
+
+// Reset values to zero
+func (s *Snmp) Reset() {
+	atomic.StoreUint64(&s.BytesSent, 0)
+	atomic.StoreUint64(&s.BytesReceived, 0)
+	atomic.StoreUint64(&s.MaxConn, 0)
+	atomic.StoreUint64(&s.ActiveOpens, 0)
+	atomic.StoreUint64(&s.PassiveOpens, 0)
+	atomic.StoreUint64(&s.CurrEstab, 0)
+	atomic.StoreUint64(&s.InErrs, 0)
+	atomic.StoreUint64(&s.InCsumErrors, 0)
+	atomic.StoreUint64(&s.KCPInErrors, 0)
+	atomic.StoreUint64(&s.InPkts, 0)
+	atomic.StoreUint64(&s.OutPkts, 0)
+	atomic.StoreUint64(&s.InSegs, 0)
+	atomic.StoreUint64(&s.OutSegs, 0)
+	atomic.StoreUint64(&s.InBytes, 0)
+	atomic.StoreUint64(&s.OutBytes, 0)
+	atomic.StoreUint64(&s.RetransSegs, 0)
+	atomic.StoreUint64(&s.FastRetransSegs, 0)
+	atomic.StoreUint64(&s.EarlyRetransSegs, 0)
+	atomic.StoreUint64(&s.LostSegs, 0)
+	atomic.StoreUint64(&s.RepeatSegs, 0)
+	atomic.StoreUint64(&s.FECParityShards, 0)
+	atomic.StoreUint64(&s.FECErrs, 0)
+	atomic.StoreUint64(&s.FECRecovered, 0)
+	atomic.StoreUint64(&s.FECShortShards, 0)
+}
+
+// DefaultSnmp is the global KCP connection statistics collector
+var DefaultSnmp *Snmp
+
+func init() {
+	DefaultSnmp = newSnmp()
+}

kcp-go/updater.go

@@ -0,0 +1,104 @@
+package kcp
+
+import (
+	"container/heap"
+	"sync"
+	"time"
+)
+
+var updater updateHeap
+
+func init() {
+	updater.init()
+	go updater.updateTask()
+}
+
+// entry contains a session update info
+type entry struct {
+	ts time.Time
+	s  *UDPSession
+}
+
+// a global heap managed kcp.flush() caller
+type updateHeap struct {
+	entries  []entry
+	mu       sync.Mutex
+	chWakeUp chan struct{}
+}
+
+func (h *updateHeap) Len() int           { return len(h.entries) }
+func (h *updateHeap) Less(i, j int) bool { return h.entries[i].ts.Before(h.entries[j].ts) }
+func (h *updateHeap) Swap(i, j int) {
+	h.entries[i], h.entries[j] = h.entries[j], h.entries[i]
+	h.entries[i].s.updaterIdx = i
+	h.entries[j].s.updaterIdx = j
+}
+
+func (h *updateHeap) Push(x interface{}) {
+	h.entries = append(h.entries, x.(entry))
+	n := len(h.entries)
+	h.entries[n-1].s.updaterIdx = n - 1
+}
+
+func (h *updateHeap) Pop() interface{} {
+	n := len(h.entries)
+	x := h.entries[n-1]
+	h.entries[n-1].s.updaterIdx = -1
+	h.entries[n-1] = entry{} // manual set nil for GC
+	h.entries = h.entries[0 : n-1]
+	return x
+}
+
+func (h *updateHeap) init() {
+	h.chWakeUp = make(chan struct{}, 1)
+}
+
+func (h *updateHeap) addSession(s *UDPSession) {
+	h.mu.Lock()
+	heap.Push(h, entry{time.Now(), s})
+	h.mu.Unlock()
+	h.wakeup()
+}
+
+func (h *updateHeap) removeSession(s *UDPSession) {
+	h.mu.Lock()
+	if s.updaterIdx != -1 {
+		heap.Remove(h, s.updaterIdx)
+	}
+	h.mu.Unlock()
+}
+
+func (h *updateHeap) wakeup() {
+	select {
+	case h.chWakeUp <- struct{}{}:
+	default:
+	}
+}
+
+func (h *updateHeap) updateTask() {
+	var timer <-chan time.Time
+	for {
+		select {
+		case <-timer:
+		case <-h.chWakeUp:
+		}
+
+		h.mu.Lock()
+		hlen := h.Len()
+		for i := 0; i < hlen; i++ {
+			entry := &h.entries[0]
+			if time.Now().After(entry.ts) {
+				interval := entry.s.update()
+				entry.ts = time.Now().Add(interval)
+				heap.Fix(h, 0)
+			} else {
+				break
+			}
+		}
+
+		if hlen > 0 {
+			timer = time.After(h.entries[0].ts.Sub(time.Now()))
+		}
+		h.mu.Unlock()
+	}
+}

p2pclient/config.go

@@ -0,0 +1,48 @@
+package main
+
+import (
+	"encoding/json"
+	"os"
+)
+
+// Config for client
+type Config struct {
+	ListenTcp    string `json:"listentcp"`
+	TargetTcp    string `json:"targettcp"`
+	BindUdp      string `json:"bindudp"`
+	RemoteUdp    string `json:"remoteudp"`
+	IsServer     bool   `json:"isserver"`
+	Key          string `json:"key"`
+	Passwd		 string `json:"passwd"`
+	Crypt        string `json:"crypt"`
+	Mode         string `json:"mode"`
+	AutoExpire   int    `json:"autoexpire"`
+	MTU          int    `json:"mtu"`
+	SndWnd       int    `json:"sndwnd"`
+	RcvWnd       int    `json:"rcvwnd"`
+	DataShard    int    `json:"datashard"`
+	ParityShard  int    `json:"parityshard"`
+	DSCP         int    `json:"dscp"`
+	NoComp       bool   `json:"nocomp"`
+	AckNodelay   bool   `json:"acknodelay"`
+	NoDelay      int    `json:"nodelay"`
+	Interval     int    `json:"interval"`
+	Resend       int    `json:"resend"`
+	NoCongestion int    `json:"nc"`
+	SockBuf      int    `json:"sockbuf"`
+	KeepAlive    int    `json:"keepalive"`
+	Log          string `json:"log"`
+	SnmpLog      string `json:"snmplog"`
+	SnmpPeriod   int    `json:"snmpperiod"`
+	Quiet        bool   `json:"quiet"`
+}
+
+func parseJSONConfig(config *Config, path string) error {
+	file, err := os.Open(path) // For read access.
+	if err != nil {
+		return err
+	}
+	defer file.Close()
+
+	return json.NewDecoder(file).Decode(config)
+}

p2pclient/main.go

@@ -0,0 +1,632 @@
+package main
+
+import (
+	"bufio"
+	"crypto/sha1"
+	"encoding/csv"
+	"encoding/json"
+	"fmt"
+	"io"
+	"log"
+	"math/rand"
+	"net"
+	"os"
+	"sync/atomic"
+	"time"
+
+	"golang.org/x/crypto/pbkdf2"
+
+	"github.com/golang/snappy"
+	"github.com/urfave/cli"
+	"github.com/hikaricai/p2p_tun/kcp-go"
+	"github.com/xtaci/smux"
+
+	"path/filepath"
+)
+
+var (
+	// VERSION is injected by buildflags
+	VERSION = "SELFBUILD"
+	// SALT is use for pbkdf2 key expansion
+	SALT = "kcp-go"
+)
+
+type compStream struct {
+	conn net.Conn
+	w    *snappy.Writer
+	r    *snappy.Reader
+}
+
+func (c *compStream) Read(p []byte) (n int, err error) {
+	return c.r.Read(p)
+}
+
+func (c *compStream) Write(p []byte) (n int, err error) {
+	n, err = c.w.Write(p)
+	err = c.w.Flush()
+	return n, err
+}
+
+func (c *compStream) Close() error {
+	return c.conn.Close()
+}
+
+func newCompStream(conn net.Conn) *compStream {
+	c := new(compStream)
+	c.conn = conn
+	c.w = snappy.NewBufferedWriter(conn)
+	c.r = snappy.NewReader(conn)
+	return c
+}
+
+func handleLocalTcp(sess *smux.Session, p1 io.ReadWriteCloser, quiet bool) {
+	if !quiet {
+		log.Println("stream opened")
+		defer log.Println("stream closed")
+	}
+
+	defer p1.Close()
+	p2, err := sess.OpenStream()
+	if err != nil {
+		return
+	}
+	defer p2.Close()
+
+	// start tunnel
+	p1die := make(chan struct{})
+	buf1 := make([]byte, 65535)
+	go func() { io.CopyBuffer(p1, p2, buf1); close(p1die) }()
+
+	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:
+	}
+}
+
+func checkError(err error) {
+	if err != nil {
+		log.Printf("%+v\n", err)
+		os.Exit(-1)
+	}
+}
+
+func main() {
+	rand.Seed(int64(time.Now().Nanosecond()))
+	if VERSION == "SELFBUILD" {
+		// add more log flags for debugging
+		log.SetFlags(log.LstdFlags | log.Lshortfile)
+	}
+	myApp := cli.NewApp()
+	myApp.Name = "kcptun"
+	myApp.Usage = "client(with SMUX)"
+	myApp.Version = VERSION
+	myApp.Flags = []cli.Flag{
+		cli.StringFlag{
+			Name:  "targettcp, t",
+			Value: "127.0.0.1:22",
+			Usage: "target server address",
+		},
+		cli.StringFlag{
+			Name:  "listentcp,l",
+			Value: ":12948",
+			Usage: "local listen address",
+		},
+		cli.StringFlag{
+			Name:  "remoteudp, r",
+			Value: "vps:29900",
+			Usage: "kcp server address",
+		},
+		cli.StringFlag{
+			Name:  "bindudp, b",
+			Value: ":29900",
+			Usage: "bind local udp",
+		},
+		cli.StringFlag{
+			Name:  "key, k",
+			Value: "1234",
+			Usage: "p2p pair key",
+		},
+		cli.BoolFlag{
+			Name:  "server, s",
+			Usage: "bind local udp",
+		},
+		cli.StringFlag{
+			Name:   "passwd",
+			Value:  "1234",
+			Usage:  "pre-shared secret between client and server",
+			EnvVar: "KCPTUN_KEY",
+		},
+		cli.StringFlag{
+			Name:  "crypt",
+			Value: "aes",
+			Usage: "aes, aes-128, aes-192, salsa20, blowfish, twofish, cast5, 3des, tea, xtea, xor, sm4, none",
+		},
+		cli.StringFlag{
+			Name:  "mode",
+			Value: "fast",
+			Usage: "profiles: fast3, fast2, fast, normal, manual",
+		},
+		cli.IntFlag{
+			Name:  "conn",
+			Value: 1,
+			Usage: "set num of UDP connections to server",
+		},
+		cli.IntFlag{
+			Name:  "autoexpire",
+			Value: 0,
+			Usage: "set auto expiration time(in seconds) for a single UDP connection, 0 to disable",
+		},
+		cli.IntFlag{
+			Name:  "mtu",
+			Value: 1350,
+			Usage: "set maximum transmission unit for UDP packets",
+		},
+		cli.IntFlag{
+			Name:  "sndwnd",
+			Value: 1024,
+			Usage: "set send window size(num of packets)",
+		},
+		cli.IntFlag{
+			Name:  "rcvwnd",
+			Value: 1024,
+			Usage: "set receive window size(num of packets)",
+		},
+		cli.IntFlag{
+			Name:  "datashard,ds",
+			Value: 10,
+			Usage: "set reed-solomon erasure coding - datashard",
+		},
+		cli.IntFlag{
+			Name:  "parityshard,ps",
+			Value: 3,
+			Usage: "set reed-solomon erasure coding - parityshard",
+		},
+		cli.IntFlag{
+			Name:  "dscp",
+			Value: 0,
+			Usage: "set DSCP(6bit)",
+		},
+		cli.BoolFlag{
+			Name:  "nocomp",
+			Usage: "disable compression",
+		},
+		cli.BoolFlag{
+			Name:   "acknodelay",
+			Usage:  "flush ack immediately when a packet is received",
+			Hidden: true,
+		},
+		cli.IntFlag{
+			Name:   "nodelay",
+			Value:  0,
+			Hidden: true,
+		},
+		cli.IntFlag{
+			Name:   "interval",
+			Value:  50,
+			Hidden: true,
+		},
+		cli.IntFlag{
+			Name:   "resend",
+			Value:  0,
+			Hidden: true,
+		},
+		cli.IntFlag{
+			Name:   "nc",
+			Value:  0,
+			Hidden: true,
+		},
+		cli.IntFlag{
+			Name:  "sockbuf",
+			Value: 4194304, // socket buffer size in bytes
+			Usage: "per-socket buffer in bytes",
+		},
+		cli.IntFlag{
+			Name:  "keepalive",
+			Value: 10, // nat keepalive interval in seconds
+			Usage: "seconds between heartbeats",
+		},
+		cli.StringFlag{
+			Name:  "snmplog",
+			Value: "",
+			Usage: "collect snmp to file, aware of timeformat in golang, like: ./snmp-20060102.log",
+		},
+		cli.IntFlag{
+			Name:  "snmpperiod",
+			Value: 60,
+			Usage: "snmp collect period, in seconds",
+		},
+		cli.StringFlag{
+			Name:  "log",
+			Value: "",
+			Usage: "specify a log file to output, default goes to stderr",
+		},
+		cli.BoolFlag{
+			Name:  "quiet",
+			Usage: "to suppress the 'stream open/close' messages",
+		},
+		cli.StringFlag{
+			Name:  "c",
+			Value: "", // when the value is not empty, the config path must exists
+			Usage: "config from json file, which will override the command from shell",
+		},
+	}
+	myApp.Action = func(c *cli.Context) error {
+		config := Config{}
+		config.ListenTcp = c.String("listentcp")
+		config.RemoteUdp = c.String("remoteudp")
+		config.TargetTcp = c.String("targettcp")
+		config.BindUdp = c.String("bindudp")
+		config.IsServer = c.Bool("server")
+		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")
+		config.DataShard = c.Int("datashard")
+		config.ParityShard = c.Int("parityshard")
+		config.DSCP = c.Int("dscp")
+		config.NoComp = c.Bool("nocomp")
+		config.AckNodelay = c.Bool("acknodelay")
+		config.NoDelay = c.Int("nodelay")
+		config.Interval = c.Int("interval")
+		config.Resend = c.Int("resend")
+		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") != "" {
+			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
+		case "fast":
+			config.NoDelay, config.Interval, config.Resend, config.NoCongestion = 0, 30, 2, 1
+		case "fast2":
+			config.NoDelay, config.Interval, config.Resend, config.NoCongestion = 1, 20, 2, 1
+		case "fast3":
+			config.NoDelay, config.Interval, config.Resend, config.NoCongestion = 1, 10, 2, 1
+		}
+
+		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)
+			if err == nil{
+				p2pHandle(&config, peerAddr, chTCPConn)
+			}
+		}
+	}
+	myApp.Run(os.Args)
+}
+
+func p2pHandle(config *Config, peerAddr string, chTCPConn chan *net.TCPConn){
+	udpAddr, err := net.ResolveUDPAddr("udp", config.BindUdp)
+	checkError(err)
+	udpconn, err := net.ListenUDP("udp", udpAddr)
+	checkError(err)
+	defer udpconn.Close()
+
+	smuxSession, err := newSmuxSession(udpconn, config, peerAddr)
+	checkError(err)
+
+	go handleTargetTcp(config.TargetTcp, smuxSession, config.Quiet)
+	tickerCheck := time.NewTicker(10*time.Second)
+	defer tickerCheck.Stop()
+	for {
+		select {
+		case p1 := <-chTCPConn:
+			go handleLocalTcp(smuxSession, p1, config.Quiet)
+		case <-tickerCheck.C:
+			if smuxSession.IsClosed(){
+				log.Println("p2p session closed")
+				return
+			}
+		}
+	}
+}
+
+func tcpListener(chTCPConn chan *net.TCPConn, config *Config){
+	listenTcpAddr, err := net.ResolveTCPAddr("tcp", config.ListenTcp)
+	checkError(err)
+	listener, err := net.ListenTCP("tcp", listenTcpAddr)
+	checkError(err)
+	log.Println("listening on:", listener.Addr())
+	for{
+		p1, err := listener.AcceptTCP()
+		if err != nil {
+			log.Fatalln(err)
+			checkError(err)
+		}
+		chTCPConn <- p1
+	}
+}
+
+func getPeerAddr(config *Config)(string, error){
+
+	udpAddr, err := net.ResolveUDPAddr("udp", config.BindUdp)
+	checkError(err)
+	udpconn, err := net.ListenUDP("udp", udpAddr)
+	checkError(err)
+	defer udpconn.Close()
+
+	kcpConn, err := newKcpConn(udpconn, config, config.RemoteUdp)
+	defer kcpConn.Close()
+
+	var dataReady int32
+	var chPing = make(chan struct{})
+	defer close(chPing)
+
+	go pingCheck(kcpConn, &dataReady, chPing)
+	reader := bufio.NewReader(kcpConn)
+	pairMess := phaseJsonMess("login", config.Key)
+	finMess := phaseJsonMess("fin", "good bye")
+	var peerAddr string
+	log.Println("writing mess")
+	n, err := kcpConn.Write(pairMess)
+	if err != nil {
+		log.Println("kcpConn.Write", err)
+		return "", err
+	}
+	log.Println("writen ", n)
+	for {
+		log.Println("waiting for server")
+		line, err := reader.ReadString('\n')
+		if err != nil {
+			log.Println("reader.ReadString", err)
+			return "", err
+		}
+		var mess DigHoleMess
+		json.Unmarshal([]byte(line), &mess)
+		log.Println("rcv Cmd", mess.Cmd)
+		switch mess.Cmd {
+		case "ping":
+			atomic.StoreInt32(&dataReady, 1)
+			log.Println("rcv ping")
+		case "pair":
+			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)
+			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)
+	defer tickerPing.Stop()
+	defer 	log.Println("pingCheck return")
+	for {
+		select {
+		case <-tickerDie.C:
+			if !atomic.CompareAndSwapInt32(dataReady, 1, 0) {
+				log.Println("ping timeout")
+				conn.Close()
+				return
+			}
+		case <-tickerPing.C:
+			conn.Write(jsonPingMess)
+		case <- chPing:
+			return
+		}
+	}
+}
+
+type DigHoleMess struct {
+	Cmd  string
+	Data string
+}
+
+func phaseJsonMess(cmd string, data string) []byte {
+	mess := DigHoleMess{cmd, data}
+	jsonMess, err := json.Marshal(mess)
+	if err != nil {
+		log.Println(err)
+	}
+	return append(jsonMess, '\n')
+}
+
+func newSmuxSession(udpconn net.PacketConn, config *Config, remoteAddr string) (*smux.Session, error) {
+
+	kcpconn, err := newKcpConn(udpconn, config, remoteAddr)
+	if err != nil {
+		return nil, err
+	}
+	smuxConfig := smux.DefaultConfig()
+	smuxConfig.MaxReceiveBuffer = config.SockBuf
+	smuxConfig.KeepAliveInterval = time.Duration(config.KeepAlive) * time.Second
+	// stream multiplex
+	var smuxSession *smux.Session
+
+	if config.IsServer {
+		smuxSession, err = smux.Server(kcpconn, smuxConfig)
+	} else {
+		smuxSession, err = smux.Client(kcpconn, smuxConfig)
+	}
+	if err == nil {
+		log.Println("connection:", kcpconn.LocalAddr(), "->", kcpconn.RemoteAddr())
+	}
+	return smuxSession, err
+}
+
+func newKcpConn(udpconn net.PacketConn, config *Config, remoteAddr string) (*kcp.UDPSession, error) {
+	log.Println("initiating key derivation")
+	pass := pbkdf2.Key([]byte(config.Passwd), []byte(SALT), 4096, 32, sha1.New)
+	var block kcp.BlockCrypt
+	switch config.Crypt {
+	case "sm4":
+		block, _ = kcp.NewSM4BlockCrypt(pass[:16])
+	case "tea":
+		block, _ = kcp.NewTEABlockCrypt(pass[:16])
+	case "xor":
+		block, _ = kcp.NewSimpleXORBlockCrypt(pass)
+	case "none":
+		block, _ = kcp.NewNoneBlockCrypt(pass)
+	case "aes-128":
+		block, _ = kcp.NewAESBlockCrypt(pass[:16])
+	case "aes-192":
+		block, _ = kcp.NewAESBlockCrypt(pass[:24])
+	case "blowfish":
+		block, _ = kcp.NewBlowfishBlockCrypt(pass)
+	case "twofish":
+		block, _ = kcp.NewTwofishBlockCrypt(pass)
+	case "cast5":
+		block, _ = kcp.NewCast5BlockCrypt(pass[:16])
+	case "3des":
+		block, _ = kcp.NewTripleDESBlockCrypt(pass[:24])
+	case "xtea":
+		block, _ = kcp.NewXTEABlockCrypt(pass[:16])
+	case "salsa20":
+		block, _ = kcp.NewSalsa20BlockCrypt(pass)
+	default:
+		config.Crypt = "aes"
+		block, _ = kcp.NewAESBlockCrypt(pass)
+	}
+
+	kcpconn, err := kcp.NewP2pConn(udpconn, remoteAddr, block, config.DataShard, config.ParityShard)
+	if err != nil {
+		return nil, err
+	}
+
+	kcpconn.SetStreamMode(true)
+	kcpconn.SetWriteDelay(false)
+	kcpconn.SetNoDelay(config.NoDelay, config.Interval, config.Resend, config.NoCongestion)
+	kcpconn.SetWindowSize(config.SndWnd, config.RcvWnd)
+	kcpconn.SetMtu(config.MTU)
+	kcpconn.SetACKNoDelay(config.AckNodelay)
+
+	if err := kcpconn.SetDSCP(config.DSCP); err != nil {
+		log.Println("SetDSCP:", err)
+	}
+	if err := kcpconn.SetReadBuffer(config.SockBuf); err != nil {
+		log.Println("SetReadBuffer:", err)
+	}
+	if err := kcpconn.SetWriteBuffer(config.SockBuf); err != nil {
+		log.Println("SetWriteBuffer:", err)
+	}
+	return kcpconn, err
+}
+
+func handleTargetTcp(addr string, session *smux.Session, quiet bool) {
+	for {
+		p1, err := session.AcceptStream()
+		if err != nil {
+			log.Println(err)
+			return
+		}
+		p2, err := net.DialTimeout("tcp", addr, 5*time.Second)
+		if err != nil {
+			p1.Close()
+			log.Println(err)
+			continue
+		}
+		go func() {
+			if !quiet {
+				log.Println("tcp client opened")
+				defer log.Println("tcp client closed")
+			}
+			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) }()
+
+			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:
+			}
+		}()
+	}
+}
+
+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()
+		}
+	}
+}

p2pclient/signal.go

@@ -0,0 +1,29 @@
+// +build linux darwin freebsd
+
+package main
+
+import (
+	"log"
+	"os"
+	"os/signal"
+	"syscall"
+
+	kcp "github.com/hikaricai/p2p_tun/kcp-go"
+)
+
+func init() {
+	go sigHandler()
+}
+
+func sigHandler() {
+	ch := make(chan os.Signal, 1)
+	signal.Notify(ch, syscall.SIGUSR1)
+	signal.Ignore(syscall.SIGPIPE)
+
+	for {
+		switch <-ch {
+		case syscall.SIGUSR1:
+			log.Printf("KCP SNMP:%+v", kcp.DefaultSnmp.Copy())
+		}
+	}
+}

p2pserver/config.go

@@ -0,0 +1,43 @@
+package main
+
+import (
+	"encoding/json"
+	"os"
+)
+
+// Config for server
+type Config struct {
+	Listen       string `json:"listen"`
+	Key          string `json:"key"`
+	Crypt        string `json:"crypt"`
+	Mode         string `json:"mode"`
+	MTU          int    `json:"mtu"`
+	SndWnd       int    `json:"sndwnd"`
+	RcvWnd       int    `json:"rcvwnd"`
+	DataShard    int    `json:"datashard"`
+	ParityShard  int    `json:"parityshard"`
+	DSCP         int    `json:"dscp"`
+	NoComp       bool   `json:"nocomp"`
+	AckNodelay   bool   `json:"acknodelay"`
+	NoDelay      int    `json:"nodelay"`
+	Interval     int    `json:"interval"`
+	Resend       int    `json:"resend"`
+	NoCongestion int    `json:"nc"`
+	SockBuf      int    `json:"sockbuf"`
+	KeepAlive    int    `json:"keepalive"`
+	Log          string `json:"log"`
+	SnmpLog      string `json:"snmplog"`
+	SnmpPeriod   int    `json:"snmpperiod"`
+	Pprof        bool   `json:"pprof"`
+	Quiet        bool   `json:"quiet"`
+}
+
+func parseJSONConfig(config *Config, path string) error {
+	file, err := os.Open(path) // For read access.
+	if err != nil {
+		return err
+	}
+	defer file.Close()
+
+	return json.NewDecoder(file).Decode(config)
+}

p2pserver/main.go

@@ -0,0 +1,486 @@
+package main
+
+import (
+	"bufio"
+	"crypto/sha1"
+	"encoding/csv"
+	"encoding/json"
+	"fmt"
+	"log"
+	"math/rand"
+	"net"
+	"net/http"
+	_ "net/http/pprof"
+	"os"
+	"sync"
+	"sync/atomic"
+	"time"
+
+	"golang.org/x/crypto/pbkdf2"
+
+	"path/filepath"
+
+	"github.com/golang/snappy"
+	"github.com/urfave/cli"
+	"github.com/hikaricai/p2p_tun/kcp-go"
+)
+
+var (
+	// VERSION is injected by buildflags
+	VERSION = "SELFBUILD"
+	// SALT is use for pbkdf2 key expansion
+	SALT = "kcp-go"
+)
+
+type compStream struct {
+	conn net.Conn
+	w    *snappy.Writer
+	r    *snappy.Reader
+}
+
+func (c *compStream) Read(p []byte) (n int, err error) {
+	return c.r.Read(p)
+}
+
+func (c *compStream) Write(p []byte) (n int, err error) {
+	n, err = c.w.Write(p)
+	err = c.w.Flush()
+	return n, err
+}
+
+func (c *compStream) Close() error {
+	return c.conn.Close()
+}
+
+func newCompStream(conn net.Conn) *compStream {
+	c := new(compStream)
+	c.conn = conn
+	c.w = snappy.NewBufferedWriter(conn)
+	c.r = snappy.NewReader(conn)
+	return c
+}
+
+func checkError(err error) {
+	if err != nil {
+		log.Printf("%+v\n", err)
+		os.Exit(-1)
+	}
+}
+
+func main() {
+	rand.Seed(int64(time.Now().Nanosecond()))
+	if VERSION == "SELFBUILD" {
+		// add more log flags for debugging
+		log.SetFlags(log.LstdFlags | log.Lshortfile)
+	}
+	myApp := cli.NewApp()
+	myApp.Name = "kcptun"
+	myApp.Usage = "server(with SMUX)"
+	myApp.Version = VERSION
+	myApp.Flags = []cli.Flag{
+		cli.StringFlag{
+			Name:  "bind,b",
+			Value: ":12948",
+			Usage: "local listen address",
+		},
+		cli.StringFlag{
+			Name:  "listen,l",
+			Value: ":29900",
+			Usage: "kcp server listen address",
+		},
+		cli.StringFlag{
+			Name:  "target, t",
+			Value: "127.0.0.1:12948",
+			Usage: "target server address",
+		},
+		cli.StringFlag{
+			Name:   "key",
+			Value:  "1234",
+			Usage:  "pre-shared secret between client and server",
+			EnvVar: "KCPTUN_KEY",
+		},
+		cli.StringFlag{
+			Name:  "crypt",
+			Value: "aes",
+			Usage: "aes, aes-128, aes-192, salsa20, blowfish, twofish, cast5, 3des, tea, xtea, xor, sm4, none",
+		},
+		cli.StringFlag{
+			Name:  "mode",
+			Value: "fast",
+			Usage: "profiles: fast3, fast2, fast, normal, manual",
+		},
+		cli.IntFlag{
+			Name:  "mtu",
+			Value: 1350,
+			Usage: "set maximum transmission unit for UDP packets",
+		},
+		cli.IntFlag{
+			Name:  "sndwnd",
+			Value: 1024,
+			Usage: "set send window size(num of packets)",
+		},
+		cli.IntFlag{
+			Name:  "rcvwnd",
+			Value: 1024,
+			Usage: "set receive window size(num of packets)",
+		},
+		cli.IntFlag{
+			Name:  "datashard,ds",
+			Value: 10,
+			Usage: "set reed-solomon erasure coding - datashard",
+		},
+		cli.IntFlag{
+			Name:  "parityshard,ps",
+			Value: 3,
+			Usage: "set reed-solomon erasure coding - parityshard",
+		},
+		cli.IntFlag{
+			Name:  "dscp",
+			Value: 0,
+			Usage: "set DSCP(6bit)",
+		},
+		cli.BoolFlag{
+			Name:  "nocomp",
+			Usage: "disable compression",
+		},
+		cli.BoolFlag{
+			Name:   "acknodelay",
+			Usage:  "flush ack immediately when a packet is received",
+			Hidden: true,
+		},
+		cli.IntFlag{
+			Name:   "nodelay",
+			Value:  0,
+			Hidden: true,
+		},
+		cli.IntFlag{
+			Name:   "interval",
+			Value:  50,
+			Hidden: true,
+		},
+		cli.IntFlag{
+			Name:   "resend",
+			Value:  0,
+			Hidden: true,
+		},
+		cli.IntFlag{
+			Name:   "nc",
+			Value:  0,
+			Hidden: true,
+		},
+		cli.IntFlag{
+			Name:  "sockbuf",
+			Value: 4194304, // socket buffer size in bytes
+			Usage: "per-socket buffer in bytes",
+		},
+		cli.IntFlag{
+			Name:  "keepalive",
+			Value: 10, // nat keepalive interval in seconds
+			Usage: "seconds between heartbeats",
+		},
+		cli.StringFlag{
+			Name:  "snmplog",
+			Value: "",
+			Usage: "collect snmp to file, aware of timeformat in golang, like: ./snmp-20060102.log",
+		},
+		cli.IntFlag{
+			Name:  "snmpperiod",
+			Value: 60,
+			Usage: "snmp collect period, in seconds",
+		},
+		cli.BoolFlag{
+			Name:  "pprof",
+			Usage: "start profiling server on :6060",
+		},
+		cli.StringFlag{
+			Name:  "log",
+			Value: "",
+			Usage: "specify a log file to output, default goes to stderr",
+		},
+		cli.BoolFlag{
+			Name:  "quiet",
+			Usage: "to suppress the 'stream open/close' messages",
+		},
+		cli.StringFlag{
+			Name:  "c",
+			Value: "", // when the value is not empty, the config path must exists
+			Usage: "config from json file, which will override the command from shell",
+		},
+	}
+	myApp.Action = func(c *cli.Context) error {
+		config := Config{}
+		config.Listen = c.String("listen")
+		config.Key = c.String("key")
+		config.Crypt = c.String("crypt")
+		config.Mode = c.String("mode")
+		config.MTU = c.Int("mtu")
+		config.SndWnd = c.Int("sndwnd")
+		config.RcvWnd = c.Int("rcvwnd")
+		config.DataShard = c.Int("datashard")
+		config.ParityShard = c.Int("parityshard")
+		config.DSCP = c.Int("dscp")
+		config.NoComp = c.Bool("nocomp")
+		config.AckNodelay = c.Bool("acknodelay")
+		config.NoDelay = c.Int("nodelay")
+		config.Interval = c.Int("interval")
+		config.Resend = c.Int("resend")
+		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")
+		config.Quiet = c.Bool("quiet")
+
+		if c.String("c") != "" {
+			//Now only support json config file
+			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
+		case "fast":
+			config.NoDelay, config.Interval, config.Resend, config.NoCongestion = 0, 30, 2, 1
+		case "fast2":
+			config.NoDelay, config.Interval, config.Resend, config.NoCongestion = 1, 20, 2, 1
+		case "fast3":
+			config.NoDelay, config.Interval, config.Resend, config.NoCongestion = 1, 10, 2, 1
+		}
+
+		log.Println("version:", VERSION)
+		log.Println("initiating key derivation")
+		pass := pbkdf2.Key([]byte(config.Key), []byte(SALT), 4096, 32, sha1.New)
+		var block kcp.BlockCrypt
+		switch config.Crypt {
+		case "sm4":
+			block, _ = kcp.NewSM4BlockCrypt(pass[:16])
+		case "tea":
+			block, _ = kcp.NewTEABlockCrypt(pass[:16])
+		case "xor":
+			block, _ = kcp.NewSimpleXORBlockCrypt(pass)
+		case "none":
+			block, _ = kcp.NewNoneBlockCrypt(pass)
+		case "aes-128":
+			block, _ = kcp.NewAESBlockCrypt(pass[:16])
+		case "aes-192":
+			block, _ = kcp.NewAESBlockCrypt(pass[:24])
+		case "blowfish":
+			block, _ = kcp.NewBlowfishBlockCrypt(pass)
+		case "twofish":
+			block, _ = kcp.NewTwofishBlockCrypt(pass)
+		case "cast5":
+			block, _ = kcp.NewCast5BlockCrypt(pass[:16])
+		case "3des":
+			block, _ = kcp.NewTripleDESBlockCrypt(pass[:24])
+		case "xtea":
+			block, _ = kcp.NewXTEABlockCrypt(pass[:16])
+		case "salsa20":
+			block, _ = kcp.NewSalsa20BlockCrypt(pass)
+		default:
+			config.Crypt = "aes"
+			block, _ = kcp.NewAESBlockCrypt(pass)
+		}
+
+		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)
+		}
+		if err := lis.SetReadBuffer(config.SockBuf); err != nil {
+			log.Println("SetReadBuffer:", err)
+		}
+		if err := lis.SetWriteBuffer(config.SockBuf); err != nil {
+			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 {
+				log.Println("remote address:", conn.RemoteAddr())
+				conn.SetStreamMode(true)
+				conn.SetWriteDelay(false)
+				conn.SetNoDelay(config.NoDelay, config.Interval, config.Resend, config.NoCongestion)
+				conn.SetMtu(config.MTU)
+				conn.SetWindowSize(config.SndWnd, config.RcvWnd)
+				conn.SetACKNoDelay(config.AckNodelay)
+
+				go handleClient(conn)
+			} else {
+				log.Printf("%+v", err)
+			}
+		}
+	}
+	myApp.Run(os.Args)
+}
+
+type DigHoleMess struct {
+	Cmd  string
+	Data string
+}
+
+type Peer struct {
+	addr   string
+	chPair chan string
+}
+
+var keymap = make(map[string]*Peer)
+
+var keymu  sync.Mutex
+
+func handleClient(conn *kcp.UDPSession) {
+	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)
+	for {
+		line, err := reader.ReadString('\n')
+		if err != nil {
+			log.Println("reader.ReadString", err)
+			return
+		}
+		var mess DigHoleMess
+		json.Unmarshal([]byte(line), &mess)
+		switch mess.Cmd {
+		case "login":
+			remoteAddr := conn.RemoteAddr().String()
+			log.Println("login from ", remoteAddr)
+			key := mess.Data
+
+			log.Println("key is ", key)
+			keymu.Lock()
+			peer, 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", peerAddr)
+				conn.Write(jsonPairMess)
+			} else {
+				log.Println("no peer, registed")
+				peer := Peer{remoteAddr, make(chan string)}
+				keymap[key] = &peer
+				keymu.Unlock()
+				go notifyAddr(conn, &peer, chThreadDie)
+			}
+		case "ping":
+			atomic.StoreInt32(&dataReady, 1)
+			jsonPingMess := phaseJsonMess("ping", "hello")
+			conn.Write(jsonPingMess)
+			log.Println("rcv ping from ", conn.RemoteAddr().String())
+		case "fin":
+			log.Println("fin from", conn.RemoteAddr().String())
+			time.Sleep(time.Second)
+			return
+		}
+	}
+}
+
+func timeout(conn *kcp.UDPSession, dataReady *int32, chThreadDie chan struct{}){
+	tickerDie := time.NewTicker(30*time.Second)
+	defer tickerDie.Stop()
+
+	for {
+		select {
+		case <-tickerDie.C:
+			if !atomic.CompareAndSwapInt32(dataReady, 1, 0) {
+				log.Println("ping timeout")
+				conn.Close()
+				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", peerAddr)
+			conn.Write(jsonPairMess)
+			return
+		}
+	}
+}
+
+func phaseJsonMess(cmd string, data string) []byte {
+	mess := DigHoleMess{cmd, data}
+	jsonMess, err := json.Marshal(mess)
+	if err != nil {
+		log.Println(err)
+	}
+	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()
+		}
+	}
+}

p2pserver/signal.go

@@ -0,0 +1,29 @@
+// +build linux darwin freebsd
+
+package main
+
+import (
+	"log"
+	"os"
+	"os/signal"
+	"syscall"
+
+	kcp "github.com/hikaricai/p2p_tun/kcp-go"
+)
+
+func init() {
+	go sigHandler()
+}
+
+func sigHandler() {
+	ch := make(chan os.Signal, 1)
+	signal.Notify(ch, syscall.SIGUSR1)
+	signal.Ignore(syscall.SIGPIPE)
+
+	for {
+		switch <-ch {
+		case syscall.SIGUSR1:
+			log.Printf("KCP SNMP:%+v", kcp.DefaultSnmp.Copy())
+		}
+	}
+}

smux/LICENSE

@@ -0,0 +1,21 @@
+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

@@ -0,0 +1,60 @@
+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

@@ -0,0 +1,80 @@
+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

@@ -0,0 +1,350 @@
+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

@@ -0,0 +1,262 @@
+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 }