lint

.gitignore

@@ -0,0 +1,24 @@
+# Compiled Object files, Static and Dynamic libs (Shared Objects)
+*.o
+*.a
+*.so
+
+# Folders
+_obj
+_test
+
+# Architecture specific extensions/prefixes
+*.[568vq]
+[568vq].out
+
+*.cgo1.go
+*.cgo2.c
+_cgo_defun.c
+_cgo_gotypes.go
+_cgo_export.*
+
+_testmain.go
+
+*.exe
+*.test
+*.prof

.travis.yml

@@ -0,0 +1,12 @@
+language: go
+go:
+    - 1.6
+before_install:
+    - go get github.com/mattn/goveralls
+    - go get golang.org/x/tools/cmd/cover
+install:
+    - go get github.com/xtaci/kcp-go
+before_script:
+script:
+    - $HOME/gopath/bin/goveralls -service=travis-ci
+    - exit 0

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.
+

README.md

@@ -0,0 +1,60 @@
+# ***kcp-go*** :pisces:
+[![GoDoc][1]][2] [![Powered][9]][10] [![Build Status][3]][4] [![Go Report Card][5]][6] [![Coverage Statusd][7]][8] 
+
+[1]: https://godoc.org/github.com/xtaci/kcp-go?status.svg
+[2]: https://godoc.org/github.com/xtaci/kcp-go
+[3]: https://travis-ci.org/xtaci/kcp-go.svg?branch=master
+[4]: https://travis-ci.org/xtaci/kcp-go
+[5]: https://goreportcard.com/badge/github.com/xtaci/kcp-go
+[6]: https://goreportcard.com/report/github.com/xtaci/kcp-go
+[7]: https://coveralls.io/repos/github/xtaci/kcp-go/badge.svg?branch=master
+[8]: https://coveralls.io/github/xtaci/kcp-go?branch=master
+[9]: https://img.shields.io/badge/KCP-Powered-blue.svg
+[10]: https://github.com/skywind3000/kcp
+
+***A full-featured reliable UDP communication library for various usage***
+
+# ***Features*** :zap:
+1. Optimized for ***game development***.
+1. Compatible with [skywind3000's](https://github.com/skywind3000) C version with modificiations.
+1. ***Cache friendly*** and ***Memory optimized*** design in golang.
+1. A [session manager](https://github.com/xtaci/kcp-go/blob/master/sess.go) has provided with compatiablity for [net.Conn](https://golang.org/pkg/net/#Conn) and [net.Listener](https://golang.org/pkg/net/#Listener).
+1. Support [FEC(Forward Error Correction)](https://en.wikipedia.org/wiki/Forward_error_correction) with [Reed-Solomon Codes](https://en.wikipedia.org/wiki/Reed%E2%80%93Solomon_error_correction)
+1. Support packet level encryption with [AES](https://en.wikipedia.org/wiki/Advanced_Encryption_Standard), [TEA](https://en.wikipedia.org/wiki/Tiny_Encryption_Algorithm).
+
+# ***Conventions*** :zap:
+1. [UDP](https://en.wikipedia.org/wiki/User_Datagram_Protocol)  for packet delivery.
+2. ```conv uint32``` in session manager is a ***random number*** initiated by client.
+3. KCP doesn't define control messages like SYN/ACK/FIN/RST in TCP, a real world example is to use some ***multiplexing*** protocol over session, such as [yamux](https://github.com/hashicorp/yamux), see [kcptun](https://github.com/xtaci/kcptun) for example.
+
+# ***Examples*** :zap:
+Client:   [full demo](https://github.com/xtaci/kcptun/blob/master/client/main.go#L231)
+```go
+kcpconn, err := kcp.DialWithOptions("192.168.0.1:10000", nil, 10, 3)
+```
+Server:   [full demo](https://github.com/xtaci/kcptun/blob/master/server/main.go#L235)
+```go
+lis, err := kcp.ListenWithOptions(":10000", nil, 10, 3)
+```
+
+# ***Performance*** :zap:
+```
+  型号名称：	MacBook Pro
+  型号标识符：	MacBookPro12,1
+  处理器名称：	Intel Core i5
+  处理器速度：	2.7 GHz
+  处理器数目：	1
+  核总数：	2
+  L2 缓存（每个核）：	256 KB
+  L3 缓存：	3 MB
+  内存：	8 GB
+```
+```
+$ go test -run Speed
+new client 127.0.0.1:61165
+total recv: 16777216
+time for 16MB rtt with encryption 570.41176ms
+&{BytesSent:33554432 BytesReceived:33554432 MaxConn:2 ActiveOpens:1 PassiveOpens:1 CurrEstab:1 InErrs:0 InCsumErrors:0 InSegs:42577 OutSegs:42641 OutBytes:48111336 RetransSegs:92 FastRetransSegs:92 LostSegs:0 RepeatSegs:0 FECRecovered:1 FECErrs:0 FECSegs:8514}
+PASS
+ok  	github.com/xtaci/kcp-go	0.600s
+```

crypt.go

@@ -0,0 +1,155 @@
+package kcp
+
+import (
+	"crypto/aes"
+	"crypto/cipher"
+	"crypto/sha1"
+
+	"golang.org/x/crypto/pbkdf2"
+	"golang.org/x/crypto/tea"
+)
+
+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
+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)
+}
+
+// AESBlockCrypt implements BlockCrypt with AES
+type AESBlockCrypt struct {
+	encbuf []byte
+	decbuf []byte
+	block  cipher.Block
+}
+
+// NewAESBlockCrypt initates AES BlockCrypt by the given key
+func NewAESBlockCrypt(key []byte) (BlockCrypt, error) {
+	c := new(AESBlockCrypt)
+	block, err := aes.NewCipher(key)
+	if err != nil {
+		return nil, err
+	}
+	c.block = block
+	c.encbuf = make([]byte, aes.BlockSize)
+	c.decbuf = make([]byte, 2*aes.BlockSize)
+	return c, nil
+}
+
+// Encrypt implements Encrypt interface
+func (c *AESBlockCrypt) Encrypt(dst, src []byte) {
+	encrypt(c.block, dst, src, c.encbuf)
+}
+
+// Decrypt implements Decrypt interface
+func (c *AESBlockCrypt) Decrypt(dst, src []byte) {
+	decrypt(c.block, dst, src, c.decbuf)
+}
+
+// TEABlockCrypt implements BlockCrypt with TEA
+type TEABlockCrypt struct {
+	encbuf []byte
+	decbuf []byte
+	block  cipher.Block
+}
+
+// NewTEABlockCrypt initate TEA BlockCrypt by the given key
+func NewTEABlockCrypt(key []byte) (BlockCrypt, error) {
+	c := new(TEABlockCrypt)
+	block, err := tea.NewCipherWithRounds(key, 16)
+	if err != nil {
+		return nil, err
+	}
+	c.block = block
+	c.encbuf = make([]byte, tea.BlockSize)
+	c.decbuf = make([]byte, 2*tea.BlockSize)
+	return c, nil
+}
+
+// Encrypt implements Encrypt interface
+func (c *TEABlockCrypt) Encrypt(dst, src []byte) {
+	encrypt(c.block, dst, src, c.encbuf)
+}
+
+// Decrypt implements Decrypt interface
+func (c *TEABlockCrypt) Decrypt(dst, src []byte) {
+	decrypt(c.block, dst, src, c.decbuf)
+}
+
+// SimpleXORBlockCrypt implements BlockCrypt with simple xor to a table
+type SimpleXORBlockCrypt struct {
+	xortbl []byte
+}
+
+// NewSimpleXORBlockCrypt initate SimpleXORBlockCrypt by the given key
+func NewSimpleXORBlockCrypt(key []byte) (BlockCrypt, error) {
+	c := new(SimpleXORBlockCrypt)
+	c.xortbl = pbkdf2.Key(key, []byte(saltxor), 32, mtuLimit, sha1.New)
+	return c, nil
+}
+
+// Encrypt implements Encrypt interface
+func (c *SimpleXORBlockCrypt) Encrypt(dst, src []byte) {
+	xorBytes(dst, src, c.xortbl)
+}
+
+// Decrypt implements Decrypt interface
+func (c *SimpleXORBlockCrypt) Decrypt(dst, src []byte) {
+	xorBytes(dst, src, c.xortbl)
+}
+
+// NoneBlockCrypt simple returns the plaintext
+type NoneBlockCrypt struct {
+	xortbl []byte
+}
+
+// NewNoneBlockCrypt initate NoneBlockCrypt by the given key
+func NewNoneBlockCrypt(key []byte) (BlockCrypt, error) {
+	return new(NoneBlockCrypt), nil
+}
+
+// Encrypt implements Encrypt interface
+func (c *NoneBlockCrypt) Encrypt(dst, src []byte) {}
+
+// Decrypt implements Decrypt interface
+func (c *NoneBlockCrypt) Decrypt(dst, src []byte) {}
+
+// packet encryption with local CFB mode
+func encrypt(block cipher.Block, dst, src, buf []byte) {
+	blocksize := block.BlockSize()
+	tbl := buf[:blocksize]
+	block.Encrypt(tbl, initialVector)
+	n := len(src) / blocksize
+	base := 0
+	for i := 0; i < n; i++ {
+		xorWords(dst[base:], src[base:], tbl)
+		block.Encrypt(tbl, dst[base:])
+		base += blocksize
+	}
+	xorBytes(dst[base:], src[base:], tbl)
+}
+
+func decrypt(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
+	for i := 0; i < n; i++ {
+		block.Encrypt(next, src[base:])
+		xorWords(dst[base:], src[base:], tbl)
+		tbl, next = next, tbl
+		base += blocksize
+	}
+	xorBytes(dst[base:], src[base:], tbl)
+}

crypt_test.go

@@ -0,0 +1,56 @@
+package kcp
+
+import (
+	"crypto/sha1"
+	"testing"
+
+	"golang.org/x/crypto/pbkdf2"
+)
+
+const crypt_key = "testkey"
+const crypt_salt = "kcptest"
+
+func TestAES(t *testing.T) {
+	pass := pbkdf2.Key(key, []byte(salt), 4096, 32, sha1.New)
+	bc, err := NewAESBlockCrypt(pass)
+	if err != nil {
+		t.Fatal(err)
+	}
+	data := make([]byte, 4096)
+	for i := 0; i < 4096; i++ {
+		data[i] = byte(i & 0xff)
+	}
+	bc.Encrypt(data, data)
+	bc.Decrypt(data, data)
+	t.Log(data)
+}
+
+func TestTEA(t *testing.T) {
+	pass := pbkdf2.Key(key, []byte(salt), 4096, 16, sha1.New)
+	bc, err := NewTEABlockCrypt(pass)
+	if err != nil {
+		t.Fatal(err)
+	}
+	data := make([]byte, 4096)
+	for i := 0; i < 4096; i++ {
+		data[i] = byte(i & 0xff)
+	}
+	bc.Encrypt(data, data)
+	bc.Decrypt(data, data)
+	t.Log(data)
+}
+
+func TestSimpleXOR(t *testing.T) {
+	pass := pbkdf2.Key(key, []byte(salt), 4096, 16, sha1.New)
+	bc, err := NewSimpleXORBlockCrypt(pass)
+	if err != nil {
+		t.Fatal(err)
+	}
+	data := make([]byte, 4096)
+	for i := 0; i < 4096; i++ {
+		data[i] = byte(i & 0xff)
+	}
+	bc.Encrypt(data, data)
+	bc.Decrypt(data, data)
+	t.Log(data)
+}

fec.go

@@ -0,0 +1,239 @@
+package kcp
+
+import (
+	"encoding/binary"
+	"log"
+	"sync"
+
+	"github.com/klauspost/reedsolomon"
+)
+
+const (
+	fecHeaderSize      = 6
+	fecHeaderSizePlus2 = fecHeaderSize + 2 // plus 2B data size
+	typeData           = 0xf1
+	typeFEC            = 0xf2
+	fecExpire          = 30000 // 30s
+)
+
+type (
+	// FEC defines forward error correction for packets
+	FEC struct {
+		rx           []fecPacket // ordered rx queue
+		rxlimit      int         // queue size limit
+		dataShards   int
+		parityShards int
+		shardSize    int
+		next         uint32 // next seqid
+		enc          reedsolomon.Encoder
+		shards       [][]byte
+		shardsflag   []bool
+		paws         uint32 // Protect Against Wrapped Sequence numbers
+		lastCheck    uint32
+		xmitBuf      sync.Pool
+	}
+
+	fecPacket struct {
+		seqid uint32
+		flag  uint16
+		data  []byte
+		ts    uint32
+	}
+)
+
+func newFEC(rxlimit, dataShards, parityShards int) *FEC {
+	if rxlimit < dataShards+parityShards {
+		return nil
+	}
+
+	fec := new(FEC)
+	fec.rxlimit = rxlimit
+	fec.dataShards = dataShards
+	fec.parityShards = parityShards
+	fec.shardSize = dataShards + parityShards
+	fec.paws = (0xffffffff/uint32(fec.shardSize) - 1) * uint32(fec.shardSize)
+	enc, err := reedsolomon.New(dataShards, parityShards)
+	if err != nil {
+		log.Println(err)
+		return nil
+	}
+	fec.enc = enc
+	fec.shards = make([][]byte, fec.shardSize)
+	fec.shardsflag = make([]bool, fec.shardSize)
+	fec.xmitBuf.New = func() interface{} {
+		return make([]byte, mtuLimit)
+	}
+
+	return fec
+}
+
+// decode a fec packet
+func (fec *FEC) decode(data []byte) fecPacket {
+	var pkt fecPacket
+	pkt.seqid = binary.LittleEndian.Uint32(data)
+	pkt.flag = binary.LittleEndian.Uint16(data[4:])
+	pkt.ts = currentMs()
+	// alloc & copy
+	buf := fec.xmitBuf.Get().([]byte)
+	xorBytes(buf, buf, buf)
+	copy(buf, data[6:])
+	pkt.data = buf
+	return pkt
+}
+
+func (fec *FEC) markData(data []byte) {
+	binary.LittleEndian.PutUint32(data, fec.next)
+	binary.LittleEndian.PutUint16(data[4:], typeData)
+	fec.next++
+	if fec.next >= fec.paws {
+		fec.next = 0
+	}
+}
+
+func (fec *FEC) markFEC(data []byte) {
+	binary.LittleEndian.PutUint32(data, fec.next)
+	binary.LittleEndian.PutUint16(data[4:], typeFEC)
+	fec.next++
+	if fec.next >= fec.paws {
+		fec.next = 0
+	}
+}
+
+// input a fec packet
+func (fec *FEC) input(pkt fecPacket) (recovered [][]byte) {
+	// expiration
+	now := currentMs()
+	if now-fec.lastCheck >= fecExpire {
+		var rx []fecPacket
+		for k := range fec.rx {
+			if now-fec.rx[k].ts < fecExpire {
+				rx = append(rx, fec.rx[k])
+			} else {
+				fec.xmitBuf.Put(fec.rx[k].data)
+			}
+		}
+		fec.rx = rx
+		fec.lastCheck = now
+	}
+
+	// insertion
+	n := len(fec.rx) - 1
+	insertIdx := 0
+	for i := n; i >= 0; i-- {
+		if pkt.seqid == fec.rx[i].seqid { // de-duplicate
+			fec.xmitBuf.Put(pkt.data)
+			return nil
+		} else if pkt.seqid > fec.rx[i].seqid { // insertion
+			insertIdx = i + 1
+			break
+		}
+	}
+
+	// insert into ordered rx queue
+	if insertIdx == n+1 {
+		fec.rx = append(fec.rx, pkt)
+	} else {
+		fec.rx = append(fec.rx, fecPacket{})
+		copy(fec.rx[insertIdx+1:], fec.rx[insertIdx:])
+		fec.rx[insertIdx] = pkt
+	}
+
+	shardBegin := pkt.seqid - pkt.seqid%uint32(fec.shardSize)
+	shardEnd := shardBegin + uint32(fec.shardSize) - 1
+
+	searchBegin := insertIdx - fec.shardSize
+	if searchBegin < 0 {
+		searchBegin = 0
+	}
+
+	searchEnd := insertIdx + fec.shardSize
+	if searchEnd >= len(fec.rx) {
+		searchEnd = len(fec.rx) - 1
+	}
+
+	if len(fec.rx) >= fec.dataShards && shardBegin < shardEnd {
+		numshard := 0
+		numDataShard := 0
+		first := -1
+		maxlen := 0
+		shards := fec.shards
+		shardsflag := fec.shardsflag
+		for k := range fec.shards {
+			shards[k] = nil
+			shardsflag[k] = false
+		}
+
+		for i := searchBegin; i <= searchEnd; i++ {
+			seqid := fec.rx[i].seqid
+			if seqid > shardEnd {
+				break
+			} else if seqid >= shardBegin {
+				shards[seqid%uint32(fec.shardSize)] = fec.rx[i].data
+				shardsflag[seqid%uint32(fec.shardSize)] = true
+				numshard++
+				if fec.rx[i].flag == typeData {
+					numDataShard++
+				}
+				if numshard == 1 {
+					first = i
+				}
+				if len(fec.rx[i].data) > maxlen {
+					maxlen = len(fec.rx[i].data)
+				}
+			}
+		}
+
+		if numDataShard == fec.dataShards { // no lost
+			for i := first; i < first+numshard; i++ { // free
+				fec.xmitBuf.Put(fec.rx[i].data)
+			}
+			copy(fec.rx[first:], fec.rx[first+numshard:])
+			fec.rx = fec.rx[:len(fec.rx)-numshard]
+		} else if numshard >= fec.dataShards { // recoverable
+			for k := range shards {
+				if shards[k] != nil {
+					shards[k] = shards[k][:maxlen]
+				}
+			}
+			if err := fec.enc.Reconstruct(shards); err == nil {
+				for k := range shards[:fec.dataShards] {
+					if !shardsflag[k] {
+						recovered = append(recovered, shards[k])
+					}
+				}
+			} else {
+				log.Println(err)
+			}
+
+			for i := first; i < first+numshard; i++ { // free
+				fec.xmitBuf.Put(fec.rx[i].data)
+			}
+			copy(fec.rx[first:], fec.rx[first+numshard:])
+			fec.rx = fec.rx[:len(fec.rx)-numshard]
+		}
+	}
+
+	// keep rxlen
+	if len(fec.rx) > fec.rxlimit {
+		fec.xmitBuf.Put(fec.rx[0].data) // free
+		fec.rx = fec.rx[1:]
+	}
+	return
+}
+
+func (fec *FEC) calcECC(data [][]byte, offset, maxlen int) (ecc [][]byte) {
+	if len(data) != fec.shardSize {
+		println("mismatch", len(data), fec.shardSize)
+		return nil
+	}
+	shards := make([][]byte, fec.shardSize)
+	for k := range shards {
+		shards[k] = data[k][offset:maxlen]
+	}
+
+	if err := fec.enc.Encode(shards); err != nil {
+		log.Println(err)
+		return nil
+	}
+	return data[fec.dataShards:]
+}

fec_test.go

@@ -0,0 +1,103 @@
+package kcp
+
+import (
+	"encoding/binary"
+	"math/rand"
+	"testing"
+)
+
+func TestFECOther(t *testing.T) {
+	t.Log(newFEC(128, 0, 1))
+}
+
+func TestFECNoLost(t *testing.T) {
+	fec := newFEC(128, 10, 3)
+	for i := 0; i < 100; i += 10 {
+		data := makefecgroup(i, 13)
+		for k := range data[fec.dataShards] {
+			fec.markData(data[k])
+			t.Log("input:", data[k])
+		}
+
+		ecc := fec.calcECC(data, fecHeaderSize, fecHeaderSize+4)
+		for k := range ecc {
+			fec.markFEC(ecc[k])
+		}
+		t.Log("  ecc:", ecc)
+		data = append(data, ecc...)
+		for k := range data {
+			f := fec.decode(data[k])
+			if recovered := fec.input(f); recovered != nil {
+				for k := range recovered {
+					t.Log("recovered:", binary.LittleEndian.Uint32(recovered[k]))
+				}
+			}
+		}
+	}
+}
+
+func TestFECLost1(t *testing.T) {
+	fec := newFEC(128, 10, 3)
+	for i := 0; i < 100; i += 10 {
+		data := makefecgroup(i, 13)
+		for k := range data[fec.dataShards] {
+			fec.markData(data[k])
+			t.Log("input:", data[k])
+		}
+		ecc := fec.calcECC(data, fecHeaderSize, fecHeaderSize+4)
+		for k := range ecc {
+			fec.markFEC(ecc[k])
+		}
+		t.Log("ecc:", ecc)
+		lost := rand.Intn(13)
+		t.Log("lost:", data[lost])
+		for k := range data {
+			if k != lost {
+				f := fec.decode(data[k])
+				if recovered := fec.input(f); recovered != nil {
+					for i := range recovered {
+						t.Log("recovered:", binary.LittleEndian.Uint32(recovered[i]))
+					}
+				}
+			}
+		}
+	}
+}
+
+func TestFECLost2(t *testing.T) {
+	fec := newFEC(128, 10, 3)
+	for i := 0; i < 100; i += 10 {
+		data := makefecgroup(i, 13)
+		for k := range data[fec.dataShards] {
+			fec.markData(data[k])
+			t.Log("input:", data[k])
+		}
+		ecc := fec.calcECC(data, fecHeaderSize, fecHeaderSize+4)
+		for k := range ecc {
+			fec.markFEC(ecc[k])
+		}
+		t.Log("ecc:", ecc)
+		lost1, lost2 := rand.Intn(13), rand.Intn(13)
+		t.Log(" lost1:", data[lost1])
+		t.Log(" lost2:", data[lost2])
+		for k := range data {
+			if k != lost1 && k != lost2 {
+				f := fec.decode(data[k])
+				if recovered := fec.input(f); recovered != nil {
+					for i := range recovered {
+						t.Log("recovered:", binary.LittleEndian.Uint32(recovered[i]))
+					}
+				}
+			}
+		}
+	}
+}
+
+func makefecgroup(start, size int) (group [][]byte) {
+	for i := 0; i < size; i++ {
+		data := make([]byte, fecHeaderSize+4)
+		binary.LittleEndian.PutUint32(data[fecHeaderSize:], uint32(start+i))
+		group = append(group, data)
+	}
+	return
+}

kcp.go

@@ -0,0 +1,903 @@
+// Package kcp - A Fast and Reliable ARQ Protocol
+package kcp
+
+import (
+	"encoding/binary"
+	"sync/atomic"
+)
+
+const (
+	IKCP_RTO_NDL     = 30  // no delay min rto
+	IKCP_RTO_MIN     = 100 // normal min rto
+	IKCP_RTO_DEF     = 200
+	IKCP_RTO_MAX     = 60000
+	IKCP_CMD_PUSH    = 81 // cmd: push data
+	IKCP_CMD_ACK     = 82 // cmd: ack
+	IKCP_CMD_WASK    = 83 // cmd: window probe (ask)
+	IKCP_CMD_WINS    = 84 // cmd: window size (tell)
+	IKCP_ASK_SEND    = 1  // need to send IKCP_CMD_WASK
+	IKCP_ASK_TELL    = 2  // need to send IKCP_CMD_WINS
+	IKCP_WND_SND     = 32
+	IKCP_WND_RCV     = 32
+	IKCP_MTU_DEF     = 1400
+	IKCP_ACK_FAST    = 3
+	IKCP_INTERVAL    = 100
+	IKCP_OVERHEAD    = 24
+	IKCP_DEADLINK    = 20
+	IKCP_THRESH_INIT = 2
+	IKCP_THRESH_MIN  = 2
+	IKCP_PROBE_INIT  = 7000   // 7 secs to probe window size
+	IKCP_PROBE_LIMIT = 120000 // up to 120 secs to probe window
+)
+
+// Output is a closure which captures conn and calls conn.Write
+type Output func(buf []byte, size int)
+
+/* encode 8 bits unsigned int */
+func ikcp_encode8u(p []byte, c byte) []byte {
+	p[0] = c
+	return p[1:]
+}
+
+/* decode 8 bits unsigned int */
+func ikcp_decode8u(p []byte, c *byte) []byte {
+	*c = p[0]
+	return p[1:]
+}
+
+/* encode 16 bits unsigned int (lsb) */
+func ikcp_encode16u(p []byte, w uint16) []byte {
+	binary.LittleEndian.PutUint16(p, w)
+	return p[2:]
+}
+
+/* decode 16 bits unsigned int (lsb) */
+func ikcp_decode16u(p []byte, w *uint16) []byte {
+	*w = binary.LittleEndian.Uint16(p)
+	return p[2:]
+}
+
+/* encode 32 bits unsigned int (lsb) */
+func ikcp_encode32u(p []byte, l uint32) []byte {
+	binary.LittleEndian.PutUint32(p, l)
+	return p[4:]
+}
+
+/* decode 32 bits unsigned int (lsb) */
+func ikcp_decode32u(p []byte, l *uint32) []byte {
+	*l = binary.LittleEndian.Uint32(p)
+	return p[4:]
+}
+
+func _imin_(a, b uint32) uint32 {
+	if a <= b {
+		return a
+	} else {
+		return b
+	}
+}
+
+func _imax_(a, b uint32) uint32 {
+	if a >= b {
+		return a
+	} else {
+		return b
+	}
+}
+
+func _ibound_(lower, middle, upper uint32) uint32 {
+	return _imin_(_imax_(lower, middle), upper)
+}
+
+func _itimediff(later, earlier uint32) int32 {
+	return (int32)(later - earlier)
+}
+
+// Segment defines a KCP segment
+type Segment struct {
+	conv     uint32
+	cmd      uint32
+	frg      uint32
+	wnd      uint32
+	ts       uint32
+	sn       uint32
+	una      uint32
+	resendts uint32
+	rto      uint32
+	fastack  uint32
+	xmit     uint32
+	data     []byte
+}
+
+// encode a segment into buffer
+func (seg *Segment) encode(ptr []byte) []byte {
+	ptr = ikcp_encode32u(ptr, seg.conv)
+	ptr = ikcp_encode8u(ptr, uint8(seg.cmd))
+	ptr = ikcp_encode8u(ptr, uint8(seg.frg))
+	ptr = ikcp_encode16u(ptr, uint16(seg.wnd))
+	ptr = ikcp_encode32u(ptr, seg.ts)
+	ptr = ikcp_encode32u(ptr, seg.sn)
+	ptr = ikcp_encode32u(ptr, seg.una)
+	ptr = ikcp_encode32u(ptr, uint32(len(seg.data)))
+	return ptr
+}
+
+// NewSegment creates a KCP segment
+func NewSegment(size int) *Segment {
+	seg := new(Segment)
+	seg.data = make([]byte, size)
+	return seg
+}
+
+// KCP defines a single KCP connection
+type KCP struct {
+	conv, mtu, mss, state                  uint32
+	snd_una, snd_nxt, rcv_nxt              uint32
+	ts_recent, ts_lastack, ssthresh        uint32
+	rx_rttval, rx_srtt, rx_rto, rx_minrto  uint32
+	snd_wnd, rcv_wnd, rmt_wnd, cwnd, probe uint32
+	current, interval, ts_flush, xmit      uint32
+	nodelay, updated                       uint32
+	ts_probe, probe_wait                   uint32
+	dead_link, incr                        uint32
+
+	snd_queue []Segment
+	rcv_queue []Segment
+	snd_buf   []Segment
+	rcv_buf   []Segment
+
+	acklist []uint32
+
+	buffer     []byte
+	fastresend int32
+	nocwnd     int32
+	logmask    int32
+	output     Output
+}
+
+// NewKCP create a new kcp control object, 'conv' must equal in two endpoint
+// from the same connection.
+func NewKCP(conv uint32, output Output) *KCP {
+	kcp := new(KCP)
+	kcp.conv = conv
+	kcp.snd_wnd = IKCP_WND_SND
+	kcp.rcv_wnd = IKCP_WND_RCV
+	kcp.rmt_wnd = IKCP_WND_RCV
+	kcp.mtu = IKCP_MTU_DEF
+	kcp.mss = kcp.mtu - IKCP_OVERHEAD
+	kcp.buffer = make([]byte, (kcp.mtu+IKCP_OVERHEAD)*3)
+	kcp.rx_rto = IKCP_RTO_DEF
+	kcp.rx_minrto = IKCP_RTO_MIN
+	kcp.interval = IKCP_INTERVAL
+	kcp.ts_flush = IKCP_INTERVAL
+	kcp.ssthresh = IKCP_THRESH_INIT
+	kcp.dead_link = IKCP_DEADLINK
+	kcp.output = output
+	return kcp
+}
+
+// PeekSize checks the size of next message in the recv queue
+func (kcp *KCP) PeekSize() (length int) {
+	if len(kcp.rcv_queue) == 0 {
+		return -1
+	}
+
+	seg := &kcp.rcv_queue[0]
+	if seg.frg == 0 {
+		return len(seg.data)
+	}
+
+	if len(kcp.rcv_queue) < int(seg.frg+1) {
+		return -1
+	}
+
+	for k := range kcp.rcv_queue {
+		seg := &kcp.rcv_queue[k]
+		length += len(seg.data)
+		if seg.frg == 0 {
+			break
+		}
+	}
+	return
+}
+
+// Recv is user/upper level recv: returns size, returns below zero for EAGAIN
+func (kcp *KCP) Recv(buffer []byte) (n int) {
+	if len(kcp.rcv_queue) == 0 {
+		return -1
+	}
+
+	peeksize := kcp.PeekSize()
+	if peeksize < 0 {
+		return -2
+	}
+
+	if peeksize > len(buffer) {
+		return -3
+	}
+
+	var fast_recover bool
+	if len(kcp.rcv_queue) >= int(kcp.rcv_wnd) {
+		fast_recover = true
+	}
+
+	// merge fragment
+	count := 0
+	for k := range kcp.rcv_queue {
+		seg := &kcp.rcv_queue[k]
+		copy(buffer, seg.data)
+		buffer = buffer[len(seg.data):]
+		n += len(seg.data)
+		count++
+		if seg.frg == 0 {
+			break
+		}
+	}
+	kcp.rcv_queue = kcp.rcv_queue[count:]
+
+	// move available data from rcv_buf -> rcv_queue
+	count = 0
+	for k := range kcp.rcv_buf {
+		seg := &kcp.rcv_buf[k]
+		if seg.sn == kcp.rcv_nxt && len(kcp.rcv_queue) < int(kcp.rcv_wnd) {
+			kcp.rcv_queue = append(kcp.rcv_queue, *seg)
+			kcp.rcv_nxt++
+			count++
+		} else {
+			break
+		}
+	}
+	kcp.rcv_buf = kcp.rcv_buf[count:]
+
+	// fast recover
+	if len(kcp.rcv_queue) < int(kcp.rcv_wnd) && fast_recover {
+		// ready to send back IKCP_CMD_WINS in ikcp_flush
+		// tell remote my window size
+		kcp.probe |= IKCP_ASK_TELL
+	}
+	return
+}
+
+// Send is user/upper level send, returns below zero for error
+func (kcp *KCP) Send(buffer []byte) int {
+	var count int
+	if len(buffer) == 0 {
+		return -1
+	}
+
+	if len(buffer) < int(kcp.mss) {
+		count = 1
+	} else {
+		count = (len(buffer) + int(kcp.mss) - 1) / int(kcp.mss)
+	}
+
+	if count > 255 {
+		return -2
+	}
+
+	if count == 0 {
+		count = 1
+	}
+
+	for i := 0; i < count; i++ {
+		var size int
+		if len(buffer) > int(kcp.mss) {
+			size = int(kcp.mss)
+		} else {
+			size = len(buffer)
+		}
+		seg := NewSegment(size)
+		copy(seg.data, buffer[:size])
+		seg.frg = uint32(count - i - 1)
+		kcp.snd_queue = append(kcp.snd_queue, *seg)
+		buffer = buffer[size:]
+	}
+	return 0
+}
+
+// https://tools.ietf.org/html/rfc6298
+func (kcp *KCP) update_ack(rtt int32) {
+	var rto uint32 = 0
+	if kcp.rx_srtt == 0 {
+		kcp.rx_srtt = uint32(rtt)
+		kcp.rx_rttval = uint32(rtt) / 2
+	} else {
+		delta := rtt - int32(kcp.rx_srtt)
+		if delta < 0 {
+			delta = -delta
+		}
+		kcp.rx_rttval = (3*kcp.rx_rttval + uint32(delta)) / 4
+		kcp.rx_srtt = (7*kcp.rx_srtt + uint32(rtt)) / 8
+		if kcp.rx_srtt < 1 {
+			kcp.rx_srtt = 1
+		}
+	}
+	rto = kcp.rx_srtt + _imax_(1, 4*kcp.rx_rttval)
+	kcp.rx_rto = _ibound_(kcp.rx_minrto, rto, IKCP_RTO_MAX)
+}
+
+func (kcp *KCP) shrink_buf() {
+	if len(kcp.snd_buf) > 0 {
+		seg := &kcp.snd_buf[0]
+		kcp.snd_una = seg.sn
+	} else {
+		kcp.snd_una = kcp.snd_nxt
+	}
+}
+
+func (kcp *KCP) parse_ack(sn uint32) {
+	if _itimediff(sn, kcp.snd_una) < 0 || _itimediff(sn, kcp.snd_nxt) >= 0 {
+		return
+	}
+
+	for k := range kcp.snd_buf {
+		seg := &kcp.snd_buf[k]
+		if sn == seg.sn {
+			kcp.snd_buf = append(kcp.snd_buf[:k], kcp.snd_buf[k+1:]...)
+			break
+		}
+		if _itimediff(sn, seg.sn) < 0 {
+			break
+		}
+	}
+}
+
+func (kcp *KCP) parse_fastack(sn uint32) {
+	if _itimediff(sn, kcp.snd_una) < 0 || _itimediff(sn, kcp.snd_nxt) >= 0 {
+		return
+	}
+
+	for k := range kcp.snd_buf {
+		seg := &kcp.snd_buf[k]
+		if _itimediff(sn, seg.sn) < 0 {
+			break
+		} else if sn != seg.sn {
+			seg.fastack++
+		}
+	}
+}
+
+func (kcp *KCP) parse_una(una uint32) {
+	count := 0
+	for k := range kcp.snd_buf {
+		seg := &kcp.snd_buf[k]
+		if _itimediff(una, seg.sn) > 0 {
+			count++
+		} else {
+			break
+		}
+	}
+	kcp.snd_buf = kcp.snd_buf[count:]
+}
+
+// ack append
+func (kcp *KCP) ack_push(sn, ts uint32) {
+	kcp.acklist = append(kcp.acklist, sn, ts)
+}
+
+func (kcp *KCP) ack_get(p int) (sn, ts uint32) {
+	return kcp.acklist[p*2+0], kcp.acklist[p*2+1]
+}
+
+func (kcp *KCP) parse_data(newseg *Segment) {
+	sn := newseg.sn
+	if _itimediff(sn, kcp.rcv_nxt+kcp.rcv_wnd) >= 0 ||
+		_itimediff(sn, kcp.rcv_nxt) < 0 {
+		atomic.AddUint64(&DefaultSnmp.RepeatSegs, 1)
+		return
+	}
+
+	n := len(kcp.rcv_buf) - 1
+	insert_idx := 0
+	repeat := false
+	for i := n; i >= 0; i-- {
+		seg := &kcp.rcv_buf[i]
+		if seg.sn == sn {
+			repeat = true
+			atomic.AddUint64(&DefaultSnmp.RepeatSegs, 1)
+			break
+		}
+		if _itimediff(sn, seg.sn) > 0 {
+			insert_idx = i + 1
+			break
+		}
+	}
+
+	if !repeat {
+		if insert_idx == n+1 {
+			kcp.rcv_buf = append(kcp.rcv_buf, *newseg)
+		} else {
+			kcp.rcv_buf = append(kcp.rcv_buf, Segment{})
+			copy(kcp.rcv_buf[insert_idx+1:], kcp.rcv_buf[insert_idx:])
+			kcp.rcv_buf[insert_idx] = *newseg
+		}
+	}
+
+	// move available data from rcv_buf -> rcv_queue
+	count := 0
+	for k := range kcp.rcv_buf {
+		seg := &kcp.rcv_buf[k]
+		if seg.sn == kcp.rcv_nxt && len(kcp.rcv_queue) < int(kcp.rcv_wnd) {
+			kcp.rcv_queue = append(kcp.rcv_queue, kcp.rcv_buf[k])
+			kcp.rcv_nxt++
+			count++
+		} else {
+			break
+		}
+	}
+	kcp.rcv_buf = kcp.rcv_buf[count:]
+}
+
+// Input when you received a low level packet (eg. UDP packet), call it
+func (kcp *KCP) Input(data []byte) int {
+	una := kcp.snd_una
+	if len(data) < IKCP_OVERHEAD {
+		return -1
+	}
+
+	var maxack uint32
+	var flag int
+	for {
+		var ts, sn, length, una, conv uint32
+		var wnd uint16
+		var cmd, frg uint8
+
+		if len(data) < int(IKCP_OVERHEAD) {
+			break
+		}
+
+		data = ikcp_decode32u(data, &conv)
+		if conv != kcp.conv {
+			return -1
+		}
+
+		data = ikcp_decode8u(data, &cmd)
+		data = ikcp_decode8u(data, &frg)
+		data = ikcp_decode16u(data, &wnd)
+		data = ikcp_decode32u(data, &ts)
+		data = ikcp_decode32u(data, &sn)
+		data = ikcp_decode32u(data, &una)
+		data = ikcp_decode32u(data, &length)
+		if len(data) < int(length) {
+			return -2
+		}
+
+		if cmd != IKCP_CMD_PUSH && cmd != IKCP_CMD_ACK &&
+			cmd != IKCP_CMD_WASK && cmd != IKCP_CMD_WINS {
+			return -3
+		}
+
+		kcp.rmt_wnd = uint32(wnd)
+		kcp.parse_una(una)
+		kcp.shrink_buf()
+
+		if cmd == IKCP_CMD_ACK {
+			if _itimediff(kcp.current, ts) >= 0 {
+				kcp.update_ack(_itimediff(kcp.current, ts))
+			}
+			kcp.parse_ack(sn)
+			kcp.shrink_buf()
+			if flag == 0 {
+				flag = 1
+				maxack = sn
+			} else if _itimediff(sn, maxack) > 0 {
+				maxack = sn
+			}
+		} else if cmd == IKCP_CMD_PUSH {
+			if _itimediff(sn, kcp.rcv_nxt+kcp.rcv_wnd) < 0 {
+				kcp.ack_push(sn, ts)
+				if _itimediff(sn, kcp.rcv_nxt) >= 0 {
+					seg := NewSegment(int(length))
+					seg.conv = conv
+					seg.cmd = uint32(cmd)
+					seg.frg = uint32(frg)
+					seg.wnd = uint32(wnd)
+					seg.ts = ts
+					seg.sn = sn
+					seg.una = una
+					copy(seg.data, data[:length])
+					kcp.parse_data(seg)
+				}
+			}
+		} else if cmd == IKCP_CMD_WASK {
+			// ready to send back IKCP_CMD_WINS in Ikcp_flush
+			// tell remote my window size
+			kcp.probe |= IKCP_ASK_TELL
+		} else if cmd == IKCP_CMD_WINS {
+			// do nothing
+		} else {
+			return -3
+		}
+
+		data = data[length:]
+	}
+
+	if flag != 0 {
+		kcp.parse_fastack(maxack)
+	}
+
+	if _itimediff(kcp.snd_una, una) > 0 {
+		if kcp.cwnd < kcp.rmt_wnd {
+			mss := kcp.mss
+			if kcp.cwnd < kcp.ssthresh {
+				kcp.cwnd++
+				kcp.incr += mss
+			} else {
+				if kcp.incr < mss {
+					kcp.incr = mss
+				}
+				kcp.incr += (mss*mss)/kcp.incr + (mss / 16)
+				if (kcp.cwnd+1)*mss <= kcp.incr {
+					kcp.cwnd++
+				}
+			}
+			if kcp.cwnd > kcp.rmt_wnd {
+				kcp.cwnd = kcp.rmt_wnd
+				kcp.incr = kcp.rmt_wnd * mss
+			}
+		}
+	}
+
+	return 0
+}
+
+func (kcp *KCP) wnd_unused() int32 {
+	if len(kcp.rcv_queue) < int(kcp.rcv_wnd) {
+		return int32(int(kcp.rcv_wnd) - len(kcp.rcv_queue))
+	}
+	return 0
+}
+
+// flush pending data
+func (kcp *KCP) flush() {
+	current := kcp.current
+	buffer := kcp.buffer
+	change := 0
+	lost := false
+
+	if kcp.updated == 0 {
+		return
+	}
+	var seg Segment
+	seg.conv = kcp.conv
+	seg.cmd = IKCP_CMD_ACK
+	seg.wnd = uint32(kcp.wnd_unused())
+	seg.una = kcp.rcv_nxt
+
+	// flush acknowledges
+	count := len(kcp.acklist) / 2
+	ptr := buffer
+	for i := 0; i < count; i++ {
+		size := len(buffer) - len(ptr)
+		if size+IKCP_OVERHEAD > int(kcp.mtu) {
+			kcp.output(buffer, size)
+			ptr = buffer
+		}
+		seg.sn, seg.ts = kcp.ack_get(i)
+		ptr = seg.encode(ptr)
+	}
+	kcp.acklist = nil
+
+	// probe window size (if remote window size equals zero)
+	if kcp.rmt_wnd == 0 {
+		if kcp.probe_wait == 0 {
+			kcp.probe_wait = IKCP_PROBE_INIT
+			kcp.ts_probe = kcp.current + kcp.probe_wait
+		} else {
+			if _itimediff(kcp.current, kcp.ts_probe) >= 0 {
+				if kcp.probe_wait < IKCP_PROBE_INIT {
+					kcp.probe_wait = IKCP_PROBE_INIT
+				}
+				kcp.probe_wait += kcp.probe_wait / 2
+				if kcp.probe_wait > IKCP_PROBE_LIMIT {
+					kcp.probe_wait = IKCP_PROBE_LIMIT
+				}
+				kcp.ts_probe = kcp.current + kcp.probe_wait
+				kcp.probe |= IKCP_ASK_SEND
+			}
+		}
+	} else {
+		kcp.ts_probe = 0
+		kcp.probe_wait = 0
+	}
+
+	// flush window probing commands
+	if (kcp.probe & IKCP_ASK_SEND) != 0 {
+		seg.cmd = IKCP_CMD_WASK
+		size := len(buffer) - len(ptr)
+		if size+IKCP_OVERHEAD > int(kcp.mtu) {
+			kcp.output(buffer, size)
+			ptr = buffer
+		}
+		ptr = seg.encode(ptr)
+	}
+
+	// flush window probing commands
+	if (kcp.probe & IKCP_ASK_TELL) != 0 {
+		seg.cmd = IKCP_CMD_WINS
+		size := len(buffer) - len(ptr)
+		if size+IKCP_OVERHEAD > int(kcp.mtu) {
+			kcp.output(buffer, size)
+			ptr = buffer
+		}
+		ptr = seg.encode(ptr)
+	}
+
+	kcp.probe = 0
+
+	// calculate window size
+	cwnd := _imin_(kcp.snd_wnd, kcp.rmt_wnd)
+	if kcp.nocwnd == 0 {
+		cwnd = _imin_(kcp.cwnd, cwnd)
+	}
+
+	count = 0
+	for k := range kcp.snd_queue {
+		if _itimediff(kcp.snd_nxt, kcp.snd_una+cwnd) >= 0 {
+			break
+		}
+		newseg := kcp.snd_queue[k]
+		newseg.conv = kcp.conv
+		newseg.cmd = IKCP_CMD_PUSH
+		newseg.wnd = seg.wnd
+		newseg.ts = current
+		newseg.sn = kcp.snd_nxt
+		newseg.una = kcp.rcv_nxt
+		newseg.resendts = current
+		newseg.rto = kcp.rx_rto
+		newseg.fastack = 0
+		newseg.xmit = 0
+		kcp.snd_buf = append(kcp.snd_buf, newseg)
+		kcp.snd_nxt++
+		count++
+	}
+	kcp.snd_queue = kcp.snd_queue[count:]
+
+	// calculate resent
+	resent := uint32(kcp.fastresend)
+	if kcp.fastresend <= 0 {
+		resent = 0xffffffff
+	}
+	rtomin := (kcp.rx_rto >> 3)
+	if kcp.nodelay != 0 {
+		rtomin = 0
+	}
+
+	// flush data segments
+	for k := range kcp.snd_buf {
+		segment := &kcp.snd_buf[k]
+		needsend := false
+		if segment.xmit == 0 {
+			needsend = true
+			segment.xmit++
+			segment.rto = kcp.rx_rto
+			segment.resendts = current + segment.rto + rtomin
+		} else if _itimediff(current, segment.resendts) >= 0 {
+			needsend = true
+			segment.xmit++
+			kcp.xmit++
+			if kcp.nodelay == 0 {
+				segment.rto += kcp.rx_rto
+			} else {
+				segment.rto += kcp.rx_rto / 2
+			}
+			segment.resendts = current + segment.rto
+			lost = true
+			atomic.AddUint64(&DefaultSnmp.RetransSegs, 1)
+			atomic.AddUint64(&DefaultSnmp.LostSegs, 1)
+		} else if segment.fastack >= resent {
+			needsend = true
+			segment.xmit++
+			segment.fastack = 0
+			segment.resendts = current + segment.rto
+			change++
+			atomic.AddUint64(&DefaultSnmp.RetransSegs, 1)
+			atomic.AddUint64(&DefaultSnmp.FastRetransSegs, 1)
+		} else if segment.fastack > 0 && len(kcp.snd_queue) == 0 {
+			// early retransmit
+			needsend = true
+			segment.xmit++
+			segment.fastack = 0
+			segment.resendts = current + segment.rto
+			change++
+			atomic.AddUint64(&DefaultSnmp.RetransSegs, 1)
+			atomic.AddUint64(&DefaultSnmp.EarlyRetransSegs, 1)
+		}
+
+		if needsend {
+			segment.ts = current
+			segment.wnd = seg.wnd
+			segment.una = kcp.rcv_nxt
+
+			size := len(buffer) - len(ptr)
+			need := IKCP_OVERHEAD + len(segment.data)
+
+			if size+need >= int(kcp.mtu) {
+				kcp.output(buffer, size)
+				ptr = buffer
+			}
+
+			ptr = segment.encode(ptr)
+			copy(ptr, segment.data)
+			ptr = ptr[len(segment.data):]
+
+			if segment.xmit >= kcp.dead_link {
+				kcp.state = 0xFFFFFFFF
+			}
+		}
+	}
+
+	// flash remain segments
+	size := len(buffer) - len(ptr)
+	if size > 0 {
+		kcp.output(buffer, size)
+	}
+
+	// update ssthresh
+	// rate halving, https://tools.ietf.org/html/rfc6937
+	if change != 0 {
+		inflight := kcp.snd_nxt - kcp.snd_una
+		kcp.ssthresh = inflight / 2
+		if kcp.ssthresh < IKCP_THRESH_MIN {
+			kcp.ssthresh = IKCP_THRESH_MIN
+		}
+		kcp.cwnd = kcp.ssthresh + resent
+		kcp.incr = kcp.cwnd * kcp.mss
+	}
+
+	// congestion control, https://tools.ietf.org/html/rfc5681
+	if lost {
+		kcp.ssthresh = cwnd / 2
+		if kcp.ssthresh < IKCP_THRESH_MIN {
+			kcp.ssthresh = IKCP_THRESH_MIN
+		}
+		kcp.cwnd = 1
+		kcp.incr = kcp.mss
+	}
+
+	if kcp.cwnd < 1 {
+		kcp.cwnd = 1
+		kcp.incr = kcp.mss
+	}
+}
+
+// Update updates state (call it repeatedly, every 10ms-100ms), or you can ask
+// ikcp_check when to call it again (without ikcp_input/_send calling).
+// 'current' - current timestamp in millisec.
+func (kcp *KCP) Update(current uint32) {
+	var slap int32
+
+	kcp.current = current
+
+	if kcp.updated == 0 {
+		kcp.updated = 1
+		kcp.ts_flush = kcp.current
+	}
+
+	slap = _itimediff(kcp.current, kcp.ts_flush)
+
+	if slap >= 10000 || slap < -10000 {
+		kcp.ts_flush = kcp.current
+		slap = 0
+	}
+
+	if slap >= 0 {
+		kcp.ts_flush += kcp.interval
+		if _itimediff(kcp.current, kcp.ts_flush) >= 0 {
+			kcp.ts_flush = kcp.current + kcp.interval
+		}
+		kcp.flush()
+	}
+}
+
+// Check determines when should you invoke ikcp_update:
+// returns when you should invoke ikcp_update in millisec, if there
+// is no ikcp_input/_send calling. you can call ikcp_update in that
+// time, instead of call update repeatly.
+// Important to reduce unnacessary ikcp_update invoking. use it to
+// schedule ikcp_update (eg. implementing an epoll-like mechanism,
+// or optimize ikcp_update when handling massive kcp connections)
+func (kcp *KCP) Check(current uint32) uint32 {
+	ts_flush := kcp.ts_flush
+	tm_flush := int32(0x7fffffff)
+	tm_packet := int32(0x7fffffff)
+	minimal := uint32(0)
+	if kcp.updated == 0 {
+		return current
+	}
+
+	if _itimediff(current, ts_flush) >= 10000 ||
+		_itimediff(current, ts_flush) < -10000 {
+		ts_flush = current
+	}
+
+	if _itimediff(current, ts_flush) >= 0 {
+		return current
+	}
+
+	tm_flush = _itimediff(ts_flush, current)
+
+	for k := range kcp.snd_buf {
+		seg := &kcp.snd_buf[k]
+		diff := _itimediff(seg.resendts, current)
+		if diff <= 0 {
+			return current
+		}
+		if diff < tm_packet {
+			tm_packet = diff
+		}
+	}
+
+	minimal = uint32(tm_packet)
+	if tm_packet >= tm_flush {
+		minimal = uint32(tm_flush)
+	}
+	if minimal >= kcp.interval {
+		minimal = kcp.interval
+	}
+
+	return current + minimal
+}
+
+// SetMtu changes MTU size, default is 1400
+func (kcp *KCP) SetMtu(mtu int) int {
+	if mtu < 50 || mtu < IKCP_OVERHEAD {
+		return -1
+	}
+	buffer := make([]byte, (mtu+IKCP_OVERHEAD)*3)
+	if buffer == nil {
+		return -2
+	}
+	kcp.mtu = uint32(mtu)
+	kcp.mss = kcp.mtu - IKCP_OVERHEAD
+	kcp.buffer = buffer
+	return 0
+}
+
+// NoDelay options
+// fastest: ikcp_nodelay(kcp, 1, 20, 2, 1)
+// nodelay: 0:disable(default), 1:enable
+// interval: internal update timer interval in millisec, default is 100ms
+// resend: 0:disable fast resend(default), 1:enable fast resend
+// nc: 0:normal congestion control(default), 1:disable congestion control
+func (kcp *KCP) NoDelay(nodelay, interval, resend, nc int) int {
+	if nodelay >= 0 {
+		kcp.nodelay = uint32(nodelay)
+		if nodelay != 0 {
+			kcp.rx_minrto = IKCP_RTO_NDL
+		} else {
+			kcp.rx_minrto = IKCP_RTO_MIN
+		}
+	}
+	if interval >= 0 {
+		if interval > 5000 {
+			interval = 5000
+		} else if interval < 10 {
+			interval = 10
+		}
+		kcp.interval = uint32(interval)
+	}
+	if resend >= 0 {
+		kcp.fastresend = int32(resend)
+	}
+	if nc >= 0 {
+		kcp.nocwnd = int32(nc)
+	}
+	return 0
+}
+
+// WndSize sets maximum window size: sndwnd=32, rcvwnd=32 by default
+func (kcp *KCP) WndSize(sndwnd, rcvwnd int) int {
+	if sndwnd > 0 {
+		kcp.snd_wnd = uint32(sndwnd)
+	}
+	if rcvwnd > 0 {
+		kcp.rcv_wnd = uint32(rcvwnd)
+	}
+	return 0
+}
+
+// WaitSnd gets how many packet is waiting to be sent
+func (kcp *KCP) WaitSnd() int {
+	return len(kcp.snd_buf) + len(kcp.snd_queue)
+}

kcp_test.go

@@ -0,0 +1,285 @@
+package kcp
+
+import (
+	"bytes"
+	"container/list"
+	"encoding/binary"
+	"fmt"
+	"math/rand"
+	"testing"
+	"time"
+)
+
+func iclock() int32 {
+	return int32((time.Now().UnixNano() / 1000000) & 0xffffffff)
+}
+
+type DelayPacket struct {
+	_ptr  []byte
+	_size int
+	_ts   int32
+}
+
+func (p *DelayPacket) Init(size int, src []byte) {
+	p._ptr = make([]byte, size)
+	p._size = size
+	copy(p._ptr, src[:size])
+}
+
+func (p *DelayPacket) ptr() []byte    { return p._ptr }
+func (p *DelayPacket) size() int      { return p._size }
+func (p *DelayPacket) ts() int32      { return p._ts }
+func (p *DelayPacket) setts(ts int32) { p._ts = ts }
+
+type DelayTunnel struct{ *list.List }
+type Random *rand.Rand
+type LatencySimulator struct {
+	current                        int32
+	lostrate, rttmin, rttmax, nmax int
+	p12                            DelayTunnel
+	p21                            DelayTunnel
+	r12                            *rand.Rand
+	r21                            *rand.Rand
+}
+
+// lostrate: 往返一周丢包率的百分比，默认 10%
+// rttmin：rtt最小值，默认 60
+// rttmax：rtt最大值，默认 125
+//func (p *LatencySimulator)Init(int lostrate = 10, int rttmin = 60, int rttmax = 125, int nmax = 1000):
+func (p *LatencySimulator) Init(lostrate, rttmin, rttmax, nmax int) {
+	p.r12 = rand.New(rand.NewSource(9))
+	p.r21 = rand.New(rand.NewSource(99))
+	p.p12 = DelayTunnel{list.New()}
+	p.p21 = DelayTunnel{list.New()}
+	p.current = iclock()
+	p.lostrate = lostrate / 2 // 上面数据是往返丢包率，单程除以2
+	p.rttmin = rttmin / 2
+	p.rttmax = rttmax / 2
+	p.nmax = nmax
+}
+
+// 发送数据
+// peer - 端点0/1，从0发送，从1接收；从1发送从0接收
+func (p *LatencySimulator) send(peer int, data []byte, size int) int {
+	rnd := 0
+	if peer == 0 {
+		rnd = p.r12.Intn(100)
+	} else {
+		rnd = p.r21.Intn(100)
+	}
+	//println("!!!!!!!!!!!!!!!!!!!!", rnd, p.lostrate, peer)
+	if rnd < p.lostrate {
+		return 0
+	}
+	pkt := &DelayPacket{}
+	pkt.Init(size, data)
+	p.current = iclock()
+	delay := p.rttmin
+	if p.rttmax > p.rttmin {
+		delay += rand.Int() % (p.rttmax - p.rttmin)
+	}
+	pkt.setts(p.current + int32(delay))
+	if peer == 0 {
+		p.p12.PushBack(pkt)
+	} else {
+		p.p21.PushBack(pkt)
+	}
+	return 1
+}
+
+// 接收数据
+func (p *LatencySimulator) recv(peer int, data []byte, maxsize int) int32 {
+	var it *list.Element
+	if peer == 0 {
+		it = p.p21.Front()
+		if p.p21.Len() == 0 {
+			return -1
+		}
+	} else {
+		it = p.p12.Front()
+		if p.p12.Len() == 0 {
+			return -1
+		}
+	}
+	pkt := it.Value.(*DelayPacket)
+	p.current = iclock()
+	if p.current < pkt.ts() {
+		return -2
+	}
+	if maxsize < pkt.size() {
+		return -3
+	}
+	if peer == 0 {
+		p.p21.Remove(it)
+	} else {
+		p.p12.Remove(it)
+	}
+	maxsize = pkt.size()
+	copy(data, pkt.ptr()[:maxsize])
+	return int32(maxsize)
+}
+
+//=====================================================================
+//=====================================================================
+
+// 模拟网络
+var vnet *LatencySimulator
+
+// 测试用例
+func test(mode int) {
+	// 创建模拟网络：丢包率10%，Rtt 60ms~125ms
+	vnet = &LatencySimulator{}
+	vnet.Init(10, 60, 125, 1000)
+
+	// 创建两个端点的 kcp对象，第一个参数 conv是会话编号，同一个会话需要相同
+	// 最后一个是 user参数，用来传递标识
+	output1 := func(buf []byte, size int) {
+		if vnet.send(0, buf, size) != 1 {
+		}
+	}
+	output2 := func(buf []byte, size int) {
+		if vnet.send(1, buf, size) != 1 {
+		}
+	}
+	kcp1 := NewKCP(0x11223344, output1)
+	kcp2 := NewKCP(0x11223344, output2)
+
+	current := uint32(iclock())
+	slap := current + 20
+	index := 0
+	next := 0
+	var sumrtt uint32 = 0
+	count := 0
+	maxrtt := 0
+
+	// 配置窗口大小：平均延迟200ms，每20ms发送一个包，
+	// 而考虑到丢包重发，设置最大收发窗口为128
+	kcp1.WndSize(128, 128)
+	kcp2.WndSize(128, 128)
+
+	// 判断测试用例的模式
+	if mode == 0 {
+		// 默认模式
+		kcp1.NoDelay(0, 10, 0, 0)
+		kcp2.NoDelay(0, 10, 0, 0)
+	} else if mode == 1 {
+		// 普通模式，关闭流控等
+		kcp1.NoDelay(0, 10, 0, 1)
+		kcp2.NoDelay(0, 10, 0, 1)
+	} else {
+		// 启动快速模式
+		// 第二个参数 nodelay-启用以后若干常规加速将启动
+		// 第三个参数 interval为内部处理时钟，默认设置为 10ms
+		// 第四个参数 resend为快速重传指标，设置为2
+		// 第五个参数 为是否禁用常规流控，这里禁止
+		kcp1.NoDelay(1, 10, 2, 1)
+		kcp2.NoDelay(1, 10, 2, 1)
+	}
+
+	var buffer []byte = make([]byte, 2000)
+	var hr int32
+
+	ts1 := iclock()
+
+	for {
+		time.Sleep(100 * time.Millisecond)
+		current = uint32(iclock())
+		kcp1.Update(uint32(iclock()))
+		kcp2.Update(uint32(iclock()))
+
+		// 每隔 20ms，kcp1发送数据
+		for ; current >= slap; slap += 20 {
+			buf := new(bytes.Buffer)
+			binary.Write(buf, binary.LittleEndian, uint32(index))
+			index++
+			binary.Write(buf, binary.LittleEndian, uint32(current))
+			// 发送上层协议包
+			kcp1.Send(buf.Bytes())
+			//println("now", iclock())
+		}
+
+		// 处理虚拟网络：检测是否有udp包从p1->p2
+		for {
+			hr = vnet.recv(1, buffer, 2000)
+			if hr < 0 {
+				break
+			}
+			// 如果 p2收到udp，则作为下层协议输入到kcp2
+			kcp2.Input(buffer[:hr])
+		}
+
+		// 处理虚拟网络：检测是否有udp包从p2->p1
+		for {
+			hr = vnet.recv(0, buffer, 2000)
+			if hr < 0 {
+				break
+			}
+			// 如果 p1收到udp，则作为下层协议输入到kcp1
+			kcp1.Input(buffer[:hr])
+			//println("@@@@", hr, r)
+		}
+
+		// kcp2接收到任何包都返回回去
+		for {
+			hr = int32(kcp2.Recv(buffer[:10]))
+			// 没有收到包就退出
+			if hr < 0 {
+				break
+			}
+			// 如果收到包就回射
+			buf := bytes.NewReader(buffer)
+			var sn uint32
+			binary.Read(buf, binary.LittleEndian, &sn)
+			kcp2.Send(buffer[:hr])
+		}
+
+		// kcp1收到kcp2的回射数据
+		for {
+			hr = int32(kcp1.Recv(buffer[:10]))
+			buf := bytes.NewReader(buffer)
+			// 没有收到包就退出
+			if hr < 0 {
+				break
+			}
+			var sn uint32
+			var ts, rtt uint32
+			binary.Read(buf, binary.LittleEndian, &sn)
+			binary.Read(buf, binary.LittleEndian, &ts)
+			rtt = uint32(current) - ts
+
+			if sn != uint32(next) {
+				// 如果收到的包不连续
+				//for i:=0;i<8 ;i++ {
+				//println("---", i, buffer[i])
+				//}
+				println("ERROR sn ", count, "<->", next, sn)
+				return
+			}
+
+			next++
+			sumrtt += rtt
+			count++
+			if rtt > uint32(maxrtt) {
+				maxrtt = int(rtt)
+			}
+
+			println("[RECV] mode=", mode, " sn=", sn, " rtt=", rtt)
+		}
+
+		if next > 100 {
+			break
+		}
+	}
+
+	ts1 = iclock() - ts1
+
+	names := []string{"default", "normal", "fast"}
+	fmt.Printf("%s mode result (%dms):\n", names[mode], ts1)
+	fmt.Printf("avgrtt=%d maxrtt=%d\n", int(sumrtt/uint32(count)), maxrtt)
+}
+
+func TestNetwork(t *testing.T) {
+	test(0) // 默认模式，类似 TCP：正常模式，无快速重传，常规流控
+	test(1) // 普通模式，关闭流控等
+	test(2) // 快速模式，所有开关都打开，且关闭流控
+}

sess.go

@@ -0,0 +1,768 @@
+package kcp
+
+import (
+	crand "crypto/rand"
+	"encoding/binary"
+	"errors"
+	"hash/crc32"
+	"io"
+	"log"
+	"math/rand"
+	"net"
+	"sync"
+	"sync/atomic"
+	"time"
+
+	"golang.org/x/net/ipv4"
+)
+
+var (
+	errTimeout    = errors.New("i/o timeout")
+	errBrokenPipe = errors.New("broken pipe")
+	rng           = rand.New(rand.NewSource(time.Now().UnixNano()))
+)
+
+const (
+	basePort        = 20000 // minimum port for listening
+	maxPort         = 65535 // maximum port for listening
+	defaultWndSize  = 128   // default window size, in packet
+	nonceSize       = 16    // magic number
+	crcSize         = 4     // 4bytes packet checksum
+	cryptHeaderSize = nonceSize + crcSize
+	connTimeout     = 60 * time.Second
+	mtuLimit        = 2048
+	txQueueLimit    = 8192
+	rxFecLimit      = 2048
+	soBuffer        = 16777216
+)
+
+type (
+	// UDPSession defines a KCP session implemented by UDP
+	UDPSession struct {
+		kcp           *KCP         // the core ARQ
+		fec           *FEC         // forward error correction
+		conn          *net.UDPConn // the underlying UDP socket
+		block         BlockCrypt
+		needUpdate    bool
+		l             *Listener // point to server listener if it's a server socket
+		local, remote net.Addr
+		rd            time.Time // read deadline
+		wd            time.Time // write deadline
+		sockbuff      []byte    // kcp receiving is based on packet, I turn it into stream
+		die           chan struct{}
+		isClosed      bool
+		mu            sync.Mutex
+		chReadEvent   chan struct{}
+		chWriteEvent  chan struct{}
+		chTicker      chan time.Time
+		chUDPOutput   chan []byte
+		headerSize    int
+		ackNoDelay    bool
+		xmitBuf       sync.Pool
+	}
+)
+
+// newUDPSession create a new udp session for client or server
+func newUDPSession(conv uint32, dataShards, parityShards int, l *Listener, conn *net.UDPConn, remote *net.UDPAddr, block BlockCrypt) *UDPSession {
+	sess := new(UDPSession)
+	sess.chTicker = make(chan time.Time, 1)
+	sess.chUDPOutput = make(chan []byte, txQueueLimit)
+	sess.die = make(chan struct{})
+	sess.local = conn.LocalAddr()
+	sess.chReadEvent = make(chan struct{}, 1)
+	sess.chWriteEvent = make(chan struct{}, 1)
+	sess.remote = remote
+	sess.conn = conn
+	sess.l = l
+	sess.block = block
+	sess.fec = newFEC(rxFecLimit, dataShards, parityShards)
+	sess.xmitBuf.New = func() interface{} {
+		return make([]byte, mtuLimit)
+	}
+	// caculate header size
+	if sess.block != nil {
+		sess.headerSize += cryptHeaderSize
+	}
+	if sess.fec != nil {
+		sess.headerSize += fecHeaderSizePlus2
+	}
+
+	sess.kcp = NewKCP(conv, func(buf []byte, size int) {
+		if size >= IKCP_OVERHEAD {
+			ext := sess.xmitBuf.Get().([]byte)[:sess.headerSize+size]
+			copy(ext[sess.headerSize:], buf)
+			select {
+			case sess.chUDPOutput <- ext:
+			case <-sess.die:
+			}
+		}
+	})
+	sess.kcp.WndSize(defaultWndSize, defaultWndSize)
+	sess.kcp.SetMtu(IKCP_MTU_DEF - sess.headerSize)
+
+	go sess.updateTask()
+	go sess.outputTask()
+	if l == nil { // it's a client connection
+		go sess.readLoop()
+	}
+
+	if l == nil {
+		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 the Conn Read method.
+func (s *UDPSession) Read(b []byte) (n int, err error) {
+	for {
+		s.mu.Lock()
+		if len(s.sockbuff) > 0 { // copy from buffer
+			n = copy(b, s.sockbuff)
+			s.sockbuff = s.sockbuff[n:]
+			s.mu.Unlock()
+			return n, nil
+		}
+
+		if s.isClosed {
+			s.mu.Unlock()
+			return 0, errBrokenPipe
+		}
+
+		if !s.rd.IsZero() {
+			if time.Now().After(s.rd) { // timeout
+				s.mu.Unlock()
+				return 0, errTimeout
+			}
+		}
+
+		if n := s.kcp.PeekSize(); n > 0 { // data arrived
+			if len(b) >= n {
+				s.kcp.Recv(b)
+			} else {
+				buf := make([]byte, n)
+				s.kcp.Recv(buf)
+				n = copy(b, buf)
+				s.sockbuff = buf[n:] // store remaining bytes into sockbuff for next read
+			}
+			s.mu.Unlock()
+			atomic.AddUint64(&DefaultSnmp.BytesReceived, uint64(n))
+			return n, nil
+		}
+
+		var timeout <-chan time.Time
+		if !s.rd.IsZero() {
+			delay := s.rd.Sub(time.Now())
+			timeout = time.After(delay)
+		}
+		s.mu.Unlock()
+
+		// wait for read event or timeout
+		select {
+		case <-s.chReadEvent:
+		case <-timeout:
+		case <-s.die:
+		}
+	}
+}
+
+// Write implements the Conn Write method.
+func (s *UDPSession) Write(b []byte) (n int, err error) {
+	for {
+		s.mu.Lock()
+		if s.isClosed {
+			s.mu.Unlock()
+			return 0, errBrokenPipe
+		}
+
+		if !s.wd.IsZero() {
+			if time.Now().After(s.wd) { // timeout
+				s.mu.Unlock()
+				return 0, errTimeout
+			}
+		}
+
+		if s.kcp.WaitSnd() < 2*int(s.kcp.snd_wnd) {
+			n = len(b)
+			max := s.kcp.mss << 8
+			for {
+				if len(b) <= int(max) { // in most cases
+					s.kcp.Send(b)
+					break
+				} else {
+					s.kcp.Send(b[:max])
+					b = b[max:]
+				}
+			}
+			s.kcp.current = currentMs()
+			s.kcp.flush()
+			s.mu.Unlock()
+			atomic.AddUint64(&DefaultSnmp.BytesSent, uint64(n))
+			return n, nil
+		}
+
+		var timeout <-chan time.Time
+		if !s.wd.IsZero() {
+			delay := s.wd.Sub(time.Now())
+			timeout = time.After(delay)
+		}
+		s.mu.Unlock()
+
+		// wait for write event or timeout
+		select {
+		case <-s.chWriteEvent:
+		case <-timeout:
+		case <-s.die:
+		}
+	}
+}
+
+// Close closes the connection.
+func (s *UDPSession) Close() error {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	if s.isClosed {
+		return errBrokenPipe
+	}
+	close(s.die)
+	s.isClosed = true
+	if s.l == nil { // client socket close
+		s.conn.Close()
+	}
+
+	atomic.AddUint64(&DefaultSnmp.CurrEstab, ^uint64(0))
+	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.local
+}
+
+// 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
+	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
+	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
+	return nil
+}
+
+// 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
+func (s *UDPSession) SetMtu(mtu int) {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	s.kcp.SetMtu(mtu - s.headerSize)
+}
+
+// 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
+}
+
+// SetNoDelay calls nodelay() of kcp
+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 DSCP field of IP header
+func (s *UDPSession) SetDSCP(tos int) {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	if err := ipv4.NewConn(s.conn).SetTOS(tos << 2); err != nil {
+		log.Println("set tos:", err)
+	}
+}
+
+func (s *UDPSession) outputTask() {
+	// offset pre-compute
+	fecOffset := 0
+	if s.block != nil {
+		fecOffset = cryptHeaderSize
+	}
+	szOffset := fecOffset + fecHeaderSize
+
+	// fec data group
+	var fecGroup [][]byte
+	var fecCnt int
+	var fecMaxSize int
+	if s.fec != nil {
+		fecGroup = make([][]byte, s.fec.shardSize)
+		for k := range fecGroup {
+			fecGroup[k] = make([]byte, mtuLimit)
+		}
+	}
+
+	// ping
+	ticker := time.NewTicker(5 * time.Second)
+	defer ticker.Stop()
+	for {
+		select {
+		case ext := <-s.chUDPOutput:
+			var ecc [][]byte
+			if s.fec != nil {
+				s.fec.markData(ext[fecOffset:])
+				// explicit size
+				binary.LittleEndian.PutUint16(ext[szOffset:], uint16(len(ext[szOffset:])))
+
+				// copy data to fec group
+				xorBytes(fecGroup[fecCnt], fecGroup[fecCnt], fecGroup[fecCnt])
+				copy(fecGroup[fecCnt], ext)
+				fecCnt++
+				if len(ext) > fecMaxSize {
+					fecMaxSize = len(ext)
+				}
+
+				//  calculate Reed-Solomon Erasure Code
+				if fecCnt == s.fec.dataShards {
+					ecc = s.fec.calcECC(fecGroup, szOffset, fecMaxSize)
+					for k := range ecc {
+						s.fec.markFEC(ecc[k][fecOffset:])
+						ecc[k] = ecc[k][:fecMaxSize]
+					}
+					fecCnt = 0
+					fecMaxSize = 0
+				}
+			}
+
+			if s.block != nil {
+				io.ReadFull(crand.Reader, ext[:nonceSize])
+				checksum := crc32.ChecksumIEEE(ext[cryptHeaderSize:])
+				binary.LittleEndian.PutUint32(ext[nonceSize:], checksum)
+				s.block.Encrypt(ext, ext)
+
+				if ecc != nil {
+					for k := range ecc {
+						io.ReadFull(crand.Reader, ecc[k][:nonceSize])
+						checksum := crc32.ChecksumIEEE(ecc[k][cryptHeaderSize:])
+						binary.LittleEndian.PutUint32(ecc[k][nonceSize:], checksum)
+						s.block.Encrypt(ecc[k], ecc[k])
+					}
+				}
+			}
+
+			//if rand.Intn(100) < 80 {
+			n, err := s.conn.WriteTo(ext, s.remote)
+			if err != nil {
+				log.Println(err, n)
+			}
+			atomic.AddUint64(&DefaultSnmp.OutSegs, 1)
+			atomic.AddUint64(&DefaultSnmp.OutBytes, uint64(n))
+			//}
+
+			if ecc != nil {
+				for k := range ecc {
+					n, err := s.conn.WriteTo(ecc[k], s.remote)
+					if err != nil {
+						log.Println(err, n)
+					}
+					atomic.AddUint64(&DefaultSnmp.OutSegs, 1)
+					atomic.AddUint64(&DefaultSnmp.OutBytes, uint64(n))
+				}
+			}
+			xorBytes(ext, ext, ext)
+			s.xmitBuf.Put(ext)
+		case <-ticker.C: // only for NAT keep purpose
+			sz := rng.Intn(IKCP_MTU_DEF - s.headerSize - IKCP_OVERHEAD)
+			sz += s.headerSize + IKCP_OVERHEAD
+			ping := make([]byte, sz)
+			io.ReadFull(crand.Reader, ping)
+			n, err := s.conn.WriteTo(ping, s.remote)
+			if err != nil {
+				log.Println(err, n)
+			}
+		case <-s.die:
+			return
+		}
+	}
+}
+
+// kcp update, input loop
+func (s *UDPSession) updateTask() {
+	var tc <-chan time.Time
+	if s.l == nil { // client
+		ticker := time.NewTicker(10 * time.Millisecond)
+		tc = ticker.C
+		defer ticker.Stop()
+	} else {
+		tc = s.chTicker
+	}
+
+	var nextupdate uint32
+	for {
+		select {
+		case <-tc:
+			s.mu.Lock()
+			current := currentMs()
+			if current >= nextupdate || s.needUpdate {
+				s.kcp.Update(current)
+				nextupdate = s.kcp.Check(current)
+			}
+			if s.kcp.WaitSnd() < 2*int(s.kcp.snd_wnd) {
+				s.notifyWriteEvent()
+			}
+			s.needUpdate = false
+			s.mu.Unlock()
+		case <-s.die:
+			if s.l != nil { // has listener
+				s.l.chDeadlinks <- s.remote
+			}
+			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) {
+	atomic.AddUint64(&DefaultSnmp.InSegs, 1)
+	s.mu.Lock()
+	if s.fec != nil {
+		f := s.fec.decode(data)
+		if f.flag == typeData || f.flag == typeFEC {
+			if f.flag == typeFEC {
+				atomic.AddUint64(&DefaultSnmp.FECSegs, 1)
+			}
+
+			if recovers := s.fec.input(f); recovers != nil {
+				for k := range recovers {
+					sz := binary.LittleEndian.Uint16(recovers[k])
+					if int(sz) <= len(recovers[k]) && sz >= 2 {
+						s.kcp.current = currentMs()
+						s.kcp.Input(recovers[k][2:sz])
+						atomic.AddUint64(&DefaultSnmp.FECRecovered, 1)
+					} else {
+						atomic.AddUint64(&DefaultSnmp.FECErrs, 1)
+					}
+				}
+			}
+		}
+		if f.flag == typeData {
+			s.kcp.current = currentMs()
+			s.kcp.Input(data[fecHeaderSizePlus2:])
+		}
+
+	} else {
+		s.kcp.current = currentMs()
+		s.kcp.Input(data)
+	}
+
+	if s.ackNoDelay {
+		s.kcp.current = currentMs()
+		s.kcp.flush()
+	} else {
+		s.needUpdate = true
+	}
+	s.mu.Unlock()
+	s.notifyReadEvent()
+}
+
+func (s *UDPSession) receiver(ch chan []byte) {
+	for {
+		data := s.xmitBuf.Get().([]byte)[:mtuLimit]
+		if n, _, err := s.conn.ReadFromUDP(data); err == nil && n >= s.headerSize+IKCP_OVERHEAD {
+			select {
+			case ch <- data[:n]:
+			case <-s.die:
+			}
+		} else if err != nil {
+			return
+		} else {
+			atomic.AddUint64(&DefaultSnmp.InErrs, 1)
+		}
+	}
+}
+
+// read loop for client session
+func (s *UDPSession) readLoop() {
+	chPacket := make(chan []byte, txQueueLimit)
+	go s.receiver(chPacket)
+
+	for {
+		select {
+		case data := <-chPacket:
+			raw := data
+			dataValid := false
+			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 {
+				s.kcpInput(data)
+			}
+			xorBytes(raw, raw, raw)
+			s.xmitBuf.Put(raw)
+		case <-s.die:
+			return
+		}
+	}
+}
+
+type (
+	// Listener defines a server listening for connections
+	Listener struct {
+		block                    BlockCrypt
+		dataShards, parityShards int
+		fec                      *FEC // for fec init test
+		conn                     *net.UDPConn
+		sessions                 map[string]*UDPSession
+		chAccepts                chan *UDPSession
+		chDeadlinks              chan net.Addr
+		headerSize               int
+		die                      chan struct{}
+		rxbuf                    sync.Pool
+	}
+
+	packet struct {
+		from *net.UDPAddr
+		data []byte
+	}
+)
+
+// monitor incoming data for all connections of server
+func (l *Listener) monitor() {
+	chPacket := make(chan packet, txQueueLimit)
+	go l.receiver(chPacket)
+	ticker := time.NewTicker(10 * time.Millisecond)
+	defer ticker.Stop()
+	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()
+				s, ok := l.sessions[addr]
+				if !ok { // new session
+					var conv uint32
+					convValid := false
+					if l.fec != 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 {
+						if s := newUDPSession(conv, l.dataShards, l.parityShards, l, l.conn, from, l.block); s != nil {
+							s.kcpInput(data)
+							l.sessions[addr] = s
+							l.chAccepts <- s
+						} else {
+							log.Println("cannot create session")
+						}
+					}
+				} else {
+					s.kcpInput(data)
+				}
+			}
+
+			xorBytes(raw, raw, raw)
+			l.rxbuf.Put(raw)
+		case deadlink := <-l.chDeadlinks:
+			delete(l.sessions, deadlink.String())
+		case <-l.die:
+			return
+		case <-ticker.C:
+			now := time.Now()
+			for _, s := range l.sessions {
+				select {
+				case s.chTicker <- now:
+				default:
+				}
+			}
+		}
+	}
+}
+
+func (l *Listener) receiver(ch chan packet) {
+	for {
+		data := l.rxbuf.Get().([]byte)[:mtuLimit]
+		if n, from, err := l.conn.ReadFromUDP(data); err == nil && n >= l.headerSize+IKCP_OVERHEAD {
+			ch <- packet{from, data[:n]}
+		} else if err != nil {
+			return
+		} else {
+			atomic.AddUint64(&DefaultSnmp.InErrs, 1)
+		}
+	}
+}
+
+// Accept implements the Accept method in the Listener interface; it waits for the next call and returns a generic Conn.
+func (l *Listener) Accept() (*UDPSession, error) {
+	select {
+	case c := <-l.chAccepts:
+		return c, nil
+	case <-l.die:
+		return nil, errors.New("listener stopped")
+	}
+}
+
+// Close stops listening on the UDP address. Already Accepted connections are not closed.
+func (l *Listener) Close() error {
+	if err := l.conn.Close(); err == nil {
+		close(l.die)
+		return nil
+	} else {
+		return err
+	}
+}
+
+// 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) (*Listener, error) {
+	return ListenWithOptions(laddr, nil, 10, 3)
+}
+
+// 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, err
+	}
+	conn, err := net.ListenUDP("udp", udpaddr)
+	if err != nil {
+		return nil, err
+	}
+	conn.SetReadBuffer(soBuffer)
+	conn.SetWriteBuffer(soBuffer)
+
+	l := new(Listener)
+	l.conn = conn
+	l.sessions = make(map[string]*UDPSession)
+	l.chAccepts = make(chan *UDPSession, 1024)
+	l.chDeadlinks = make(chan net.Addr, 1024)
+	l.die = make(chan struct{})
+	l.dataShards = dataShards
+	l.parityShards = parityShards
+	l.block = block
+	l.fec = newFEC(rxFecLimit, dataShards, parityShards)
+	l.rxbuf.New = func() interface{} {
+		return make([]byte, mtuLimit)
+	}
+
+	// caculate header size
+	if l.block != nil {
+		l.headerSize += cryptHeaderSize
+	}
+	if l.fec != 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) (*UDPSession, error) {
+	return DialWithOptions(raddr, nil, 10, 3)
+}
+
+// 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, err
+	}
+	for {
+		port := basePort + rng.Int()%(maxPort-basePort)
+		if udpconn, err := net.ListenUDP("udp", &net.UDPAddr{Port: port}); err == nil {
+			udpconn.SetReadBuffer(soBuffer)
+			udpconn.SetWriteBuffer(soBuffer)
+			return newUDPSession(rng.Uint32(), dataShards, parityShards, nil, udpconn, udpaddr, block), nil
+		}
+	}
+}
+
+func currentMs() uint32 {
+	return uint32(time.Now().UnixNano() / int64(time.Millisecond))
+}

sess_test.go

@@ -0,0 +1,216 @@
+package kcp
+
+import (
+	"crypto/sha1"
+	"fmt"
+	"sync"
+	"testing"
+	"time"
+
+	"golang.org/x/crypto/pbkdf2"
+)
+
+const port = "127.0.0.1:9999"
+const salt = "kcptest"
+
+var key = []byte("testkey")
+var fec = 4
+
+func DialTest() (*UDPSession, error) {
+	pass := pbkdf2.Key(key, []byte(salt), 4096, 32, sha1.New)
+	block, _ := NewNoneBlockCrypt(pass)
+	//block, _ := NewSimpleXORBlockCrypt(pass)
+	//block, _ := NewTEABlockCrypt(pass[:16])
+	//block, _ := NewAESBlockCrypt(pass)
+	return DialWithOptions(port, block, 10, 3)
+}
+
+func ListenTest() (*Listener, error) {
+	pass := pbkdf2.Key(key, []byte(salt), 4096, 32, sha1.New)
+	block, _ := NewNoneBlockCrypt(pass)
+	//block, _ := NewSimpleXORBlockCrypt(pass)
+	//block, _ := NewTEABlockCrypt(pass[:16])
+	//block, _ := NewAESBlockCrypt(pass)
+	return ListenWithOptions(port, block, 10, 3)
+}
+
+func server() {
+	l, err := ListenTest()
+	if err != nil {
+		panic(err)
+	}
+	for {
+		s, err := l.Accept()
+		if err != nil {
+			panic(err)
+		}
+
+		go handle_client(s)
+	}
+}
+
+func init() {
+	go server()
+}
+
+func handle_client(conn *UDPSession) {
+	conn.SetWindowSize(1024, 1024)
+	conn.SetNoDelay(1, 20, 2, 1)
+	fmt.Println("new client", conn.RemoteAddr())
+	buf := make([]byte, 65536)
+	count := 0
+	for {
+		n, err := conn.Read(buf)
+		if err != nil {
+			panic(err)
+		}
+		count++
+		conn.Write(buf[:n])
+	}
+}
+
+func TestSendRecv(t *testing.T) {
+	var wg sync.WaitGroup
+	const par = 1
+	wg.Add(par)
+	for i := 0; i < par; i++ {
+		go client(&wg)
+	}
+	wg.Wait()
+}
+
+func client(wg *sync.WaitGroup) {
+	cli, err := DialTest()
+	if err != nil {
+		panic(err)
+	}
+	cli.SetNoDelay(1, 20, 2, 1)
+	const N = 100
+	buf := make([]byte, 10)
+	for i := 0; i < N; i++ {
+		msg := fmt.Sprintf("hello%v", i)
+		fmt.Println("sent:", msg)
+		cli.Write([]byte(msg))
+		if n, err := cli.Read(buf); err == nil {
+			fmt.Println("recv:", string(buf[:n]))
+		} else {
+			panic(err)
+		}
+	}
+	cli.Close()
+	wg.Done()
+}
+
+func TestBigPacket(t *testing.T) {
+	var wg sync.WaitGroup
+	wg.Add(1)
+	go client2(&wg)
+	wg.Wait()
+}
+
+func client2(wg *sync.WaitGroup) {
+	cli, err := DialTest()
+	if err != nil {
+		panic(err)
+	}
+	cli.SetNoDelay(1, 20, 2, 1)
+	const N = 10
+	buf := make([]byte, 1024*512)
+	msg := make([]byte, 1024*512)
+	for i := 0; i < N; i++ {
+		cli.Write(msg)
+	}
+	println("total written:", len(msg)*N)
+
+	nrecv := 0
+	cli.SetReadDeadline(time.Now().Add(3 * time.Second))
+	for {
+		n, err := cli.Read(buf)
+		if err != nil {
+			break
+		} else {
+			nrecv += n
+			if nrecv == len(msg)*N {
+				break
+			}
+		}
+	}
+
+	println("total recv:", nrecv)
+	cli.Close()
+	wg.Done()
+}
+
+func TestSpeed(t *testing.T) {
+	var wg sync.WaitGroup
+	wg.Add(1)
+	go client3(&wg)
+	wg.Wait()
+}
+
+func client3(wg *sync.WaitGroup) {
+	cli, err := DialTest()
+	if err != nil {
+		panic(err)
+	}
+	cli.SetNoDelay(1, 20, 2, 1)
+	start := time.Now()
+
+	go func() {
+		buf := make([]byte, 1024*1024)
+		nrecv := 0
+		for {
+			n, err := cli.Read(buf)
+			if err != nil {
+				fmt.Println(err)
+				break
+			} else {
+				nrecv += n
+				if nrecv == 4096*4096 {
+					break
+				}
+			}
+		}
+		println("total recv:", nrecv)
+		cli.Close()
+		fmt.Println("time for 16MB rtt with encryption", time.Now().Sub(start))
+		fmt.Printf("%+v\n", DefaultSnmp.Copy())
+		wg.Done()
+	}()
+	msg := make([]byte, 4096)
+	cli.SetWindowSize(1024, 1024)
+	for i := 0; i < 4096; i++ {
+		cli.Write(msg)
+	}
+}
+
+func TestParallel(t *testing.T) {
+	par := 200
+	var wg sync.WaitGroup
+	wg.Add(par)
+	fmt.Println("testing parallel", par, "connections")
+	for i := 0; i < par; i++ {
+		go client4(&wg)
+	}
+	wg.Wait()
+}
+
+func client4(wg *sync.WaitGroup) {
+	cli, err := DialTest()
+	if err != nil {
+		panic(err)
+	}
+	const N = 100
+	cli.SetNoDelay(1, 20, 2, 1)
+	buf := make([]byte, 10)
+	for i := 0; i < N; i++ {
+		msg := fmt.Sprintf("hello%v", i)
+		cli.Write([]byte(msg))
+		if _, err := cli.Read(buf); err != nil {
+			break
+		}
+		<-time.After(10 * time.Millisecond)
+	}
+	cli.Close()
+	wg.Done()
+}

snmp.go

@@ -0,0 +1,62 @@
+package kcp
+
+import "sync/atomic"
+
+// Snmp defines network statistics indicator
+type Snmp struct {
+	BytesSent        uint64 // payload bytes sent
+	BytesReceived    uint64
+	MaxConn          uint64
+	ActiveOpens      uint64
+	PassiveOpens     uint64
+	CurrEstab        uint64
+	InErrs           uint64
+	InCsumErrors     uint64 // checksum errors
+	InSegs           uint64
+	OutSegs          uint64
+	OutBytes         uint64 // udp bytes sent
+	RetransSegs      uint64
+	FastRetransSegs  uint64
+	EarlyRetransSegs uint64
+	LostSegs         uint64
+	RepeatSegs       uint64
+	FECRecovered     uint64
+	FECErrs          uint64
+	FECSegs          uint64 // fec segments received
+}
+
+func newSnmp() *Snmp {
+	return new(Snmp)
+}
+
+// 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.InSegs = atomic.LoadUint64(&s.InSegs)
+	d.OutSegs = atomic.LoadUint64(&s.OutSegs)
+	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.FECSegs = atomic.LoadUint64(&s.FECSegs)
+	d.FECErrs = atomic.LoadUint64(&s.FECErrs)
+	d.FECRecovered = atomic.LoadUint64(&s.FECRecovered)
+	return d
+}
+
+// DefaultSnmp is the global KCP connection statistics collector
+var DefaultSnmp *Snmp
+
+func init() {
+	DefaultSnmp = newSnmp()
+}

xor.go

@@ -0,0 +1,105 @@
+// Copyright 2013 The Go Authors. All rights reserved.
+// Use of this source code is governed by a BSD-style
+// license that can be found in the LICENSE file.
+
+package kcp
+
+import (
+	"runtime"
+	"unsafe"
+)
+
+const wordSize = int(unsafe.Sizeof(uintptr(0)))
+const supportsUnaligned = runtime.GOARCH == "386" || runtime.GOARCH == "amd64" || runtime.GOARCH == "ppc64" || runtime.GOARCH == "ppc64le" || runtime.GOARCH == "s390x"
+
+// fastXORBytes xors in bulk. It only works on architectures that
+// support unaligned read/writes.
+func fastXORBytes(dst, a, b []byte) int {
+	n := len(a)
+	if len(b) < n {
+		n = len(b)
+	}
+
+	w := n / wordSize
+	if w > 0 {
+		wordBytes := w * wordSize
+		fastXORWords(dst[:wordBytes], a[:wordBytes], b[:wordBytes])
+	}
+
+	for i := (n - n%wordSize); i < n; i++ {
+		dst[i] = a[i] ^ b[i]
+	}
+
+	return n
+}
+
+func safeXORBytes(dst, a, b []byte) int {
+	n := len(a)
+	if len(b) < n {
+		n = len(b)
+	}
+	ex := n % 8
+	for i := 0; i < ex; i++ {
+		dst[i] = a[i] ^ b[i]
+	}
+
+	for i := ex; i < n; i += 8 {
+		dst[i] = a[i] ^ b[i]
+		dst[i+1] = a[i+1] ^ b[i+1]
+		dst[i+2] = a[i+2] ^ b[i+2]
+		dst[i+3] = a[i+3] ^ b[i+3]
+
+		dst[i+4] = a[i+4] ^ b[i+4]
+		dst[i+5] = a[i+5] ^ b[i+5]
+		dst[i+6] = a[i+6] ^ b[i+6]
+		dst[i+7] = a[i+7] ^ b[i+7]
+	}
+	return n
+}
+
+// xorBytes xors the bytes in a and b. The destination is assumed to have enough
+// space. Returns the number of bytes xor'd.
+func xorBytes(dst, a, b []byte) int {
+	if supportsUnaligned {
+		return fastXORBytes(dst, a, b)
+	} else {
+		// TODO(hanwen): if (dst, a, b) have common alignment
+		// we could still try fastXORBytes. It is not clear
+		// how often this happens, and it's only worth it if
+		// the block encryption itself is hardware
+		// accelerated.
+		return safeXORBytes(dst, a, b)
+	}
+}
+
+// fastXORWords XORs multiples of 4 or 8 bytes (depending on architecture.)
+// The arguments are assumed to be of equal length.
+func fastXORWords(dst, a, b []byte) {
+	dw := *(*[]uintptr)(unsafe.Pointer(&dst))
+	aw := *(*[]uintptr)(unsafe.Pointer(&a))
+	bw := *(*[]uintptr)(unsafe.Pointer(&b))
+	n := len(b) / wordSize
+	ex := n % 8
+	for i := 0; i < ex; i++ {
+		dw[i] = aw[i] ^ bw[i]
+	}
+
+	for i := ex; i < n; i += 8 {
+		dw[i] = aw[i] ^ bw[i]
+		dw[i+1] = aw[i+1] ^ bw[i+1]
+		dw[i+2] = aw[i+2] ^ bw[i+2]
+		dw[i+3] = aw[i+3] ^ bw[i+3]
+		dw[i+4] = aw[i+4] ^ bw[i+4]
+		dw[i+5] = aw[i+5] ^ bw[i+5]
+		dw[i+6] = aw[i+6] ^ bw[i+6]
+		dw[i+7] = aw[i+7] ^ bw[i+7]
+	}
+}
+
+func xorWords(dst, a, b []byte) {
+	if supportsUnaligned {
+		fastXORWords(dst, a, b)
+	} else {
+		safeXORBytes(dst, a, b)
+	}
+}

xor_test.go

@@ -0,0 +1,28 @@
+// Copyright 2013 The Go Authors. All rights reserved.
+// Use of this source code is governed by a BSD-style
+// license that can be found in the LICENSE file.
+
+package kcp
+
+import (
+	"bytes"
+	"testing"
+)
+
+func TestXOR(t *testing.T) {
+	for alignP := 0; alignP < 2; alignP++ {
+		for alignQ := 0; alignQ < 2; alignQ++ {
+			for alignD := 0; alignD < 2; alignD++ {
+				p := make([]byte, 1024)[alignP:]
+				q := make([]byte, 1024)[alignQ:]
+				d1 := make([]byte, 1024+alignD)[alignD:]
+				d2 := make([]byte, 1024+alignD)[alignD:]
+				xorBytes(d1, p, q)
+				safeXORBytes(d2, p, q)
+				if bytes.Compare(d1, d2) != 0 {
+					t.Error("not equal")
+				}
+			}
+		}
+	}
+}