From cedc62374ba6758d1b4d2b2f8efcde286a1cad08 Mon Sep 17 00:00:00 2001 From: tavenli Date: Tue, 1 Dec 2020 21:32:35 +0800 Subject: [PATCH] - --- forward-core/Common/KcpPipe.go | 1285 ++++++++++++++++++++++++++++++++ 1 file changed, 1285 insertions(+) create mode 100644 forward-core/Common/KcpPipe.go diff --git a/forward-core/Common/KcpPipe.go b/forward-core/Common/KcpPipe.go new file mode 100644 index 0000000..e8a1b1f --- /dev/null +++ b/forward-core/Common/KcpPipe.go @@ -0,0 +1,1285 @@ +package Common + +import ( + "encoding/binary" + "errors" + "github.com/astaxie/beego/logs" + "github.com/vzex/zappy" + "math/rand" + "net" + "sync" + "forward-core/ikcp" + "github.com/klauspost/reedsolomon" + "time" +) + +const WriteBufferSize = 5000 //udp writer will add some data for checksum or encrypt +const ReadBufferSize = WriteBufferSize * 2 //so reader must be larger + +const dataLimit = WriteBufferSize + +const mainV = 0 +const subV = 1 + +func init() { + +} + +func NewKCP(addr string, setting *KcpSetting) (*UDPListener, error) { + udpAddr, err := net.ResolveUDPAddr("udp", addr) + if err != nil { + return nil, err + } + udpSocket, _err := net.ListenUDP("udp", udpAddr) + if _err != nil { + return nil, _err + } + + listener := &UDPListener{connChan: make(chan *UDPMakeSession), quitChan: make(chan struct{}), sock: udpSocket, readBuffer: make([]byte, ReadBufferSize*2), sessions: make(map[string]*UDPMakeSession), setting: setting} + go listener.loop() + return listener, nil +} + + +func DefaultKcpSetting() *KcpSetting { + return &KcpSetting{Nodelay: 1, Interval: 10, Resend: 2, Nc: 1, Sndwnd: 1024, Rcvwnd: 1024, Mtu: 1400} +} + +func DialKcpTimeout(addr string, timeout int) (*UDPMakeSession, error) { + return DialKcpTimeoutWithSetting(addr, timeout, DefaultKcpSetting(), 0, 0, false, false) +} + +func DialKcpTimeoutWithSetting(addr string, timeout int, setting *KcpSetting, ds, ps int, comp, confuse bool) (*UDPMakeSession, error) { + bReset := false + if timeout < 5 { + bReset = true + timeout = 5 + } else if timeout > 255 { + bReset = true + timeout = 255 + } + if bReset { + logs.Debug("timeout should in [5, 255], force reset timeout to", timeout) + } + udpAddr, err := net.ResolveUDPAddr("udp", addr) + if err != nil { + return nil, err + } + sock, _err := net.ListenUDP("udp", &net.UDPAddr{}) + if _err != nil { + logs.Debug("dial addr fail", _err.Error()) + return nil, _err + } + session := &UDPMakeSession{readBuffer: make([]byte, ReadBufferSize*2), do: make(chan Action), do2: make(chan Action), quitChan: make(chan struct{}), recvChan: make(chan KcpCache), processBuffer: make([]byte, ReadBufferSize), closeChan: make(chan struct{}), encodeBuffer: make([]byte, 7), checkCanWrite: make(chan (chan struct{})), xor: setting.Xor} + session.remote = udpAddr + session.sock = sock + session.status = "firstsyn" + session.timeout = int64(timeout) + if confuse { + session.confuseSeed = rand.Intn(int(WriteBufferSize/2)) + 10 + } + if ds != 0 && ps != 0 { + er := session.SetFec(ds, ps) + if er != nil { + logs.Debug("set fec error:", er.Error()) + } else { + logs.Debug("set fec ok:", ds, ps) + } + } + if comp { + session.compressCache = make([]byte, zappy.MaxEncodedLen(ReadBufferSize)) + } + _timeout := int(timeout / 2) + if _timeout < 5 { + timeout = 5 + } + arg := int(int32(timeout) + int32(mainV<<24) + int32(subV<<16)) + arg2 := int16((ds << 8) | ps) + if comp { + arg2 = arg2 | (1 << 7) + } else { + arg2 = arg2 & ^(1 << 7) + } + info := makeEncode(session.encodeBuffer, FirstSYN, arg, arg2, session.xor) + code := session.doAndWait(func() { + sock.WriteToUDP(info, udpAddr) + }, _timeout, func(status byte, arg int32, arg2 int16) int { + if status == ResetAck { + _mainV, _subV := int(byte(arg>>24)), int(byte(arg>>16)) + logs.Debug("pipe version not eq,%d.%d=>%d.%d", mainV, subV, _mainV, _subV) + return 1 + } + if status != FirstACK { + logs.Debug("recv not FirstACK", status) + return -1 + } else { + session.status = "firstack" + session.id = int(arg) + return 0 + } + }) + if code != 0 { + logs.Debug("handshakefail with code", code) + return nil, errors.New("handshake fail,1") + } + code = session.doAndWait(func() { + if session.confuseSeed > 0 { + sock.WriteToUDP(makeEncode(session.encodeBuffer, SndSYN, session.id, 1, session.xor), udpAddr) + } else { + sock.WriteToUDP(makeEncode(session.encodeBuffer, SndSYN, session.id, 0, session.xor), udpAddr) + } + }, _timeout, func(status byte, arg int32, arg2 int16) int { + if status != SndACK { + return -1 + } else if session.id != int(arg) { + return 2 + } else { + session.status = "ok" + return 0 + } + }) + if code != 0 { + return nil, errors.New("handshake fail,2") + } + session.kcp = ikcp.Ikcp_create(uint32(session.id), session) + session.kcp.Output = udp_output + ikcp.Ikcp_wndsize(session.kcp, setting.Sndwnd, setting.Rcvwnd) + ikcp.Ikcp_nodelay(session.kcp, setting.Nodelay, setting.Interval, setting.Resend, setting.Nc) + ikcp.Ikcp_setmtu(session.kcp, setting.Mtu) + go session.loop() + return session, nil +} + +type UDPListener struct { + connChan chan *UDPMakeSession + quitChan chan struct{} + sock *net.UDPConn + readBuffer []byte + sessions map[string]*UDPMakeSession + sessionsLock sync.RWMutex + setting *KcpSetting +} + + +type UDPMakeSession struct { + id int + idstr string + status string + overTime int64 + quitChan chan struct{} + closeChan chan struct{} + recvChan chan KcpCache + handShakeChan chan string + handShakeChanQuit chan struct{} + sock *net.UDPConn + remote *net.UDPAddr + kcp *ikcp.Ikcpcb + do chan Action + do2 chan Action + checkCanWrite chan (chan struct{}) + listener *UDPListener + closed bool + + readBuffer []byte + processBuffer []byte + encodeBuffer []byte + timeout int64 + + xor string + + compressCache []byte + fecDataShards int + fecParityShards int + fecW *reedsolomon.Encoder + fecR *reedsolomon.Encoder + fecRCacheTbl map[uint]*fecInfo + fecWCacheTbl *fecInfo + fecWriteId uint //uint16 + fecSendC uint + fecSendL int + fecRecvId uint + + confuseSeed int +} + +type KcpSetting struct { + Nodelay int32 + Interval int32 //not for set + Resend int32 + Nc int32 + + Sndwnd int32 + Rcvwnd int32 + + Mtu int32 + + Xor string +} + + +type KcpCache struct { + b []byte + l int + c chan int +} + + +type Action struct { + t string + args []interface{} +} + +type fecInfo struct { + bytes [][]byte + overTime int64 +} + +func iclock() int32 { + return int32((time.Now().UnixNano() / 1000000) & 0xffffffff) +} + +func udp_output(buf []byte, _len int32, kcp *ikcp.Ikcpcb, user interface{}) int32 { + c := user.(*UDPMakeSession) + //logs.Debug("send udp", _len, c.remote.String()) + c.output(buf[:_len]) + return 0 +} + + +func (l *UDPListener) Accept() (net.Conn, error) { + var c *UDPMakeSession + var err error + select { + case c = <-l.connChan: + case <-l.quitChan: + } + if c == nil { + err = errors.New("listener quit") + } + return net.Conn(c), err +} + +func (l *UDPListener) Dump() { + l.sessionsLock.RLock() + defer l.sessionsLock.RUnlock() + for addr, session := range l.sessions { + logs.Debug("listener", addr, session.status) + } +} + +func (l *UDPListener) inner_loop() { + sock := l.sock + for { + n, from, err := sock.ReadFromUDP(l.readBuffer) + if err == nil { + //logs.Debug("recv", n, from) + addr := from.String() + l.sessionsLock.RLock() + session, bHave := l.sessions[addr] + l.sessionsLock.RUnlock() + if bHave { + if session.status == "ok" { + if session.remote.String() == from.String() && (n >= int(ikcp.IKCP_OVERHEAD) || session.compressCache != nil) { + __xor(l.readBuffer, session.xor) + var buf []byte + if n <= 7 || session.compressCache == nil { + buf = make([]byte, n) + copy(buf, l.readBuffer[:n]) + } else { + _b, _er := zappy.Decode(nil, l.readBuffer[:n]) + if _er != nil { + logs.Debug("decompress fail", _er.Error()) + //go session.Close() + //don't close pipe, just drop this data + continue + } + buf = _b + //logs.Debug("decompress", n, len(_b)) + } + session.DoAction2("input", buf, len(buf)) + } + continue + } else { + session.serverDo(string(l.readBuffer[:n])) + } + } else { + status, _, fec := makeDecode(l.readBuffer[:n], l.setting.Xor) + if status != FirstSYN { + go sock.WriteToUDP([]byte("0"), from) + logs.Debug("invalid package,reset", from, status) + continue + } + sessionId := GetId("udp") + session = &UDPMakeSession{status: "init", overTime: time.Now().Unix() + 10, remote: from, sock: sock, recvChan: make(chan KcpCache), quitChan: make(chan struct{}), readBuffer: make([]byte, ReadBufferSize*2), processBuffer: make([]byte, ReadBufferSize), timeout: 30, do: make(chan Action), do2: make(chan Action), id: sessionId, handShakeChan: make(chan string), handShakeChanQuit: make(chan struct{}), listener: l, closeChan: make(chan struct{}), encodeBuffer: make([]byte, 7), checkCanWrite: make(chan (chan struct{})), xor: l.setting.Xor} + if fec & ^(1<<7) != 0 { + er := session.SetFec((int(fec)>>8)&0xff, int(fec&^(1<<7)&0xff)) + if er != nil { + logs.Debug("set fec error:", fec, er.Error()) + } else { + logs.Debug("set fec ok:", (int(fec)>>8)&0xff, int(fec&^(1<<7)&0xff)) + } + } + if int(fec)&(1<<7) != 0 { + session.compressCache = make([]byte, zappy.MaxEncodedLen(ReadBufferSize)) + } + l.sessionsLock.Lock() + l.sessions[addr] = session + l.sessionsLock.Unlock() + session.serverInit(l, l.setting) + session.serverDo(string(l.readBuffer[:n])) + } + //logs.Debug("debug out.........") + } else { + e, ok := err.(net.Error) + if !ok || !e.Timeout() { + logs.Debug("recv error", err.Error(), from) + l.remove(from.String()) + //time.Sleep(time.Second) + break + } + } + } +} + +func (l *UDPListener) remove(addr string) { + logs.Debug("listener remove", addr) + l.sessionsLock.Lock() + defer l.sessionsLock.Unlock() + session, bHave := l.sessions[addr] + if bHave { + RmId("udp", session.id) + } + delete(l.sessions, addr) +} + +func (l *UDPListener) Close() error { + if l.sock != nil { + l.sock.Close() + l.sock = nil + } else { + return nil + } + close(l.quitChan) + return nil +} + +func (l *UDPListener) Addr() net.Addr { + return nil +} + +func (l *UDPListener) loop() { + l.inner_loop() +} + + + +const ( + Reset byte = 0 + FirstSYN byte = 6 + FirstACK byte = 1 + SndSYN byte = 2 + SndACK byte = 2 + Data byte = 4 + Ping byte = 5 + Close byte = 7 + CloseBack byte = 8 + ResetAck byte = 9 + + Fake byte = 50 +) + +func __xor(s []byte, xor string) []byte { + if len(xor) == 0 { + return s + } + encodingData := []byte(xor) + encodingLen := len(encodingData) + n := len(s) + if n == 0 { + return s + } + for i := 0; i < n; i++ { + s[i] = s[i] ^ encodingData[i%encodingLen] + } + return s +} + +func _xor(s []byte, xor string) []byte { + if len(xor) == 0 { + return s + } + encodingData := []byte(xor) + encodingLen := len(encodingData) + n := len(s) + if n == 0 { + return s + } + r := make([]byte, n) + for i := 0; i < n; i++ { + r[i] = s[i] ^ encodingData[i%encodingLen] + } + return r +} + +func makeEncode(buf []byte, status byte, arg int, arg2 int16, xor string) []byte { + buf[0] = status + binary.LittleEndian.PutUint32(buf[1:], uint32(arg)) + binary.LittleEndian.PutUint16(buf[5:], uint16(arg2)) + return _xor(buf, xor) +} + +func makeDecode(data []byte, xor string) (status byte, arg int32, arg2 int16) { + if len(data) < 7 { + return Reset, 0, 0 + } + data = _xor(data, xor) + status = data[0] + arg = int32(binary.LittleEndian.Uint32(data[1:5])) + arg2 = int16(binary.LittleEndian.Uint16(data[5:7])) + return +} + + + + +func (session *UDPMakeSession) SetFec(DataShards, ParityShards int) (er error) { + session.fecDataShards = DataShards + session.fecParityShards = ParityShards + var fec reedsolomon.Encoder + fec, er = reedsolomon.New(DataShards, ParityShards) + if er != nil { + return + } + session.fecR = &fec + fec, er = reedsolomon.New(DataShards, ParityShards) + if er == nil { + session.fecRCacheTbl = make(map[uint]*fecInfo) + session.fecWCacheTbl = nil + session.fecW = &fec + } else { + session.fecR = nil + } + return +} + +func (session *UDPMakeSession) doAndWait(f func(), sec int, readf func(status byte, arg int32, arg2 int16) int) (code int) { + t := time.NewTicker(50 * time.Millisecond) + currT := time.Now().Unix() + f() +out: + for { + select { + case <-t.C: + if time.Now().Unix()-currT >= int64(sec) { + logs.Debug("session timeout") + code = -1 + break out + } + session.sock.SetReadDeadline(time.Now().Add(2 * time.Second)) + n, from, err := session.sock.ReadFromUDP(session.readBuffer) + if err != nil { + e, ok := err.(net.Error) + if !ok || !e.Timeout() { + logs.Debug("recv error", err.Error(), from) + code = -2 + break out + } + } else { + code = readf(makeDecode(session.readBuffer[:n], session.xor)) + if code >= 0 { + break out + } + } + go f() + } + } + t.Stop() + if code > 0 { + logs.Debug("handshake fail,got code", code) + } + return +} + +func (session *UDPMakeSession) writeTo(b []byte) { + if session.compressCache != nil && len(b) > 7 { + enc, er := zappy.Encode(session.compressCache, b) + if er != nil { + logs.Debug("compress error", er.Error()) + go session.Close() + return + } + //logs.Debug("compress", len(b), len(enc)) + session.sock.WriteTo(__xor(enc, session.xor), session.remote) + } else { + session.sock.WriteTo(__xor(b, session.xor), session.remote) + } +} + +func (session *UDPMakeSession) output(b []byte) { + if session.fecW == nil { + session.writeTo(b) + } else { + id := session.fecWriteId + session.fecSendC++ + + info := session.fecWCacheTbl + if info == nil { + info = &fecInfo{make([][]byte, session.fecDataShards+session.fecParityShards), time.Now().Unix() + 15} + session.fecWCacheTbl = info + } + _b := make([]byte, len(b)+7) + _len := len(b) + _b[0] = byte(_len & 0xff) + _b[1] = byte((_len >> 8) & 0xff) + _b[2] = byte(id & 0xff) + _b[3] = byte((id >> 8) & 0xff) + _b[4] = byte((id >> 16) & 0xff) + _b[5] = byte((id >> 32) & 0xff) + _b[6] = byte(session.fecSendC - 1) + copy(_b[7:], b) + info.bytes[session.fecSendC-1] = _b + if session.fecSendL < len(_b) { + session.fecSendL = len(_b) + } + session.writeTo(_b) + if session.fecSendC >= uint(session.fecDataShards) { + for i := 0; i < session.fecDataShards; i++ { + if session.fecSendL > len(info.bytes[i]) { + __b := make([]byte, session.fecSendL) + copy(__b, info.bytes[i]) + info.bytes[i] = __b + } + } + for i := 0; i < session.fecParityShards; i++ { + info.bytes[i+session.fecDataShards] = make([]byte, session.fecSendL) + } + er := (*session.fecW).Encode(info.bytes) + if er != nil { + //logs.Debug("wocao encode err", er.Error()) + go session.Close() + return + } + for i := session.fecDataShards; i < session.fecDataShards+session.fecParityShards; i++ { + //if rand.Intn(100) >= 15 { + _info := info.bytes[i] + session.writeTo(_info) + //_len := int(_info[0]) | (int(_info[1]) << 8) + //logs.Debug("output udp id", id, i, _len, len(_info)) + //} else { + // logs.Debug("drop output udp id", id, i, _len, len(_info)) + //} + } + session.fecWCacheTbl = nil + session.fecSendC = 0 + session.fecSendL = 0 + session.fecWriteId++ + //logs.Debug("flush id", id) + } + //logs.Debug("output sn", c.fecWriteId, c.fecSendC, _len) + } +} + +func (session *UDPMakeSession) serverDo(s string) { + go func() { + //logs.Debug("prepare handshake", session.remote) + select { + case <-session.handShakeChanQuit: + case session.handShakeChan <- s: + } + }() +} +func (session *UDPMakeSession) serverInit(l *UDPListener, setting *KcpSetting) { + go func() { + c := time.NewTicker(50 * time.Millisecond) + defer func() { + close(session.handShakeChanQuit) + c.Stop() + if session.status != "ok" { + session.Close() + } + }() + overTime := time.Now().Unix() + session.timeout + for { + select { + case s := <-session.handShakeChan: + //logs.Debug("process handshake", session.remote) + status, arg, arg2 := makeDecode([]byte(s), session.xor) + switch session.status { + case "init": + if status != FirstSYN { + logs.Debug("status != FirstSYN, reset", session.remote, status) + session.sock.WriteToUDP(makeEncode(session.encodeBuffer, Reset, 0, 0, session.xor), session.remote) + return + } + _mainV, _subV := int(byte(arg>>24)), int(byte(arg>>16)) + if _mainV != mainV || _subV != subV { + session.sock.WriteToUDP(makeEncode(session.encodeBuffer, ResetAck, (mainV<<24)+(subV<<16), 0, session.xor), session.remote) + logs.Debug("pipe version not eq,kickout,%d.%d=>%d.%d", mainV, subV, _mainV, _subV) + return + } + session.status = "firstack" + session.timeout = int64(arg & 0xff) + session.sock.WriteToUDP(makeEncode(session.encodeBuffer, FirstACK, session.id, 0, session.xor), session.remote) + overTime = time.Now().Unix() + session.timeout + case "firstack": + if status != SndSYN { + logs.Debug("status != SndSYN, nothing", session.remote, status) + /* + if status == FirstSYN { + session.sock.WriteToUDP(makeEncode(session.encodeBuffer, FirstACK, session.id), session.remote, session.xor) + }*/ + return + } + if arg2 > 0 { + session.confuseSeed = rand.Intn(int(WriteBufferSize/2)) + 10 + logs.Debug("confuse!") + } + session.status = "ok" + session.kcp = ikcp.Ikcp_create(uint32(session.id), session) + session.kcp.Output = udp_output + ikcp.Ikcp_wndsize(session.kcp, setting.Sndwnd, setting.Rcvwnd) + ikcp.Ikcp_nodelay(session.kcp, setting.Nodelay, setting.Interval, setting.Resend, setting.Nc) + ikcp.Ikcp_setmtu(session.kcp, setting.Mtu) + go session.loop() + go func() { + select { + case l.connChan <- session: + case <-l.quitChan: + } + }() + session.sock.WriteToUDP(makeEncode(session.encodeBuffer, SndACK, session.id, 0, session.xor), session.remote) + overTime = time.Now().Unix() + session.timeout + } + case <-c.C: + if time.Now().Unix() > overTime { + return + } + switch session.status { + case "firstack": + session.sock.WriteToUDP(makeEncode(session.encodeBuffer, FirstACK, session.id, 0, session.xor), session.remote) + case "ok": + buf := make([]byte, 7) + session.sock.WriteToUDP(makeEncode(buf, SndACK, session.id, 0, session.xor), session.remote) + } + } + } + }() +} + +func (session *UDPMakeSession) loop() { + curr := time.Now().Unix() + session.overTime = curr + session.timeout + ping := make(chan struct{}) + pingC := 0 + callUpdate := false + updateC := make(chan struct{}) + if session.listener == nil { + go func() { + tmp := session.readBuffer + t := time.Time{} + session.sock.SetReadDeadline(t) + for { + n, from, err := session.sock.ReadFromUDP(tmp) + if err != nil { + e, ok := err.(net.Error) + if !ok || !e.Timeout() { + break + } + } + if session.remote.String() == from.String() { + //logs.Debug("===", n, len(session.compressCache)) + if n >= int(ikcp.IKCP_OVERHEAD) || n <= 7 || session.compressCache != nil { + __xor(session.readBuffer[:n], session.xor) + var buf []byte + if n <= 7 || session.compressCache == nil { + buf = make([]byte, n) + copy(buf, session.readBuffer[:n]) + } else { + _b, _er := zappy.Decode(nil, session.readBuffer[:n]) + if _er != nil { + logs.Debug("decompress fail", _er.Error()) + //go session.Close() + //don't close pipe, just drop data + continue + } + buf = _b + //logs.Debug("decompress", n, len(_b)) + } + session.DoAction2("input", buf, len(buf)) + } + } + } + }() + } + updateF := func(n time.Duration) { + if !callUpdate { + callUpdate = true + time.AfterFunc(n*time.Millisecond, func() { + select { + case updateC <- struct{}{}: + case <-session.quitChan: + } + }) + } + } + + fastCheck := false + waitList := [](chan struct{}){} + recoverChan := make(chan struct{}) + + var waitRecvCache *KcpCache + var forceWait int64 = 0 + go func() { + out: + for { + select { + //session.wait.Done() + case <-ping: + updateF(50) + pingC++ + if pingC >= 4 { + pingC = 0 + if ikcp.Ikcp_waitsnd(session.kcp) <= dataLimit/2 { + go session.DoWrite(makeEncode(session.encodeBuffer, Ping, 0, 0, session.xor)) + } + + if session.fecR != nil { + curr := time.Now().Unix() + for id, info := range session.fecRCacheTbl { + if curr >= info.overTime { + delete(session.fecRCacheTbl, id) + if session.fecRecvId <= id { + session.fecRecvId = id + 1 + } + //logs.Debug("timeout after del", id, len(c.fecRCacheTbl)) + } + } + } + if forceWait > 0 { + if time.Now().Unix() > forceWait && ikcp.Ikcp_waitsnd(session.kcp) <= dataLimit/2 { + forceWait = 0 + go func() { + select { + case <-session.quitChan: + case recoverChan <- struct{}{}: + } + }() + } + } + } + if time.Now().Unix() > session.overTime { + logs.Debug("overtime close", session.LocalAddr().String(), session.RemoteAddr().String()) + go session.Close() + } else { + time.AfterFunc(300*time.Millisecond, func() { + select { + case ping <- struct{}{}: + case <-session.quitChan: + } + }) + } + case <-recoverChan: + fastCheck = false + for _, r := range waitList { + //logs.Debug("recover writing data") + select { + case r <- struct{}{}: + case <-session.quitChan: + } + } + waitList = [](chan struct{}){} + case c := <-session.checkCanWrite: + if session.confuseSeed > 0 { + if WriteBufferSize/2-rand.Intn(session.confuseSeed) < 500 || rand.Intn(100) < 5 { + //make a sleep + forceWait = time.Now().Add(time.Millisecond * time.Duration(rand.Intn(1000))).Unix() + } + } + if ikcp.Ikcp_waitsnd(session.kcp) > dataLimit { + //logs.Debug("wait for data limit") + waitList = append(waitList, c) + if !fastCheck { + fastCheck = true + var f func() + f = func() { + n := ikcp.Ikcp_waitsnd(session.kcp) + //logs.Debug("fast check!", n, len(waitList)) + if n <= dataLimit/2 { + select { + case <-session.quitChan: + logs.Debug("recover writing data quit") + case recoverChan <- struct{}{}: + } + } else { + updateF(20) + time.AfterFunc(40*time.Millisecond, f) + } + } + time.AfterFunc(20*time.Millisecond, f) + } + } else if forceWait > 0 && time.Now().Unix() < forceWait { + waitList = append(waitList, c) + } else { + select { + case c <- struct{}{}: + case <-session.quitChan: + } + } + case ca := <-session.recvChan: + tmp := session.processBuffer + for { + hr := ikcp.Ikcp_recv(session.kcp, tmp, ReadBufferSize) + if hr > 0 { + status := tmp[0] + if status == Data { + n := 1 + l := hr + if session.confuseSeed > 0 { + n = 3 + l = (int32(tmp[1]) | (int32(tmp[2]) << 8)) + int32(n) + } + //logs.Debug("try recv", hr, n, l) + copy(ca.b, tmp[n:int(l)]) + select { + case ca.c <- int(int(l) - n): + case <-session.quitChan: + } + break + } else if status == Fake { + } else { + session.DoAction("recv", status) + } + } else { + waitRecvCache = &ca + break + } + } + case action := <-session.do2: + switch action.t { + case "input": + session.overTime = time.Now().Unix() + session.timeout + args := action.args + s := args[0].([]byte) + n := args[1].(int) + + if session.fecR != nil { + if len(s) <= 7 { + if n < 7 { + logs.Debug("recv reset") + go session._Close(false) + break + } else if n == 7 { + status, _, _ := makeDecode(s, session.xor) + if status == Reset || status == ResetAck { + logs.Debug("recv reset2", status) + go session._Close(false) + } + break + } + break + } + id := uint(int(s[2]) | (int(s[3]) << 8) | (int(s[4]) << 16) | (int(s[5]) << 24)) + var seq uint = uint(s[6]) + _len := int(s[0]) | (int(s[1]) << 8) + + //binary.Read(head[:4], binary.LittleEndian, &id) + if id < session.fecRecvId { + //logs.Debug("drop id for noneed", id, seq) + break + } + if seq < uint(session.fecDataShards) { + ikcp.Ikcp_input(session.kcp, s[7:], _len) + //logs.Debug("direct input udp", id, seq, _len) + } + if seq >= uint(session.fecDataShards+session.fecParityShards) { + logs.Debug("-ds and -ps must be equal on both sides") + go session.Close() + break + } + + tbl, have := session.fecRCacheTbl[id] + if !have { + tbl = &fecInfo{make([][]byte, session.fecDataShards+session.fecParityShards), time.Now().Unix() + 3} + session.fecRCacheTbl[id] = tbl + } + //logs.Debug("got", id, seq, n, _len) + if tbl.bytes[seq] != nil { + //dup, drop + break + } else { + tbl.bytes[seq] = s + } + count := 0 + reaL := 0 + for _, v := range tbl.bytes { + if v != nil { + count++ + if reaL < len(v) { + reaL = len(v) + } + } + } + if count >= session.fecDataShards { + markTbl := make(map[int]bool, len(tbl.bytes)) + for _seq, _b := range tbl.bytes { + if _b != nil { + markTbl[_seq] = true + } + } + bNeedRebuild := false + for i, v := range tbl.bytes { + if v != nil { + if i >= session.fecDataShards { + bNeedRebuild = true + } + if len(v) < reaL { + _b := make([]byte, reaL) + copy(_b, v) + tbl.bytes[i] = _b + } + } + } + + if bNeedRebuild { + er := (*session.fecR).Reconstruct(tbl.bytes) + if er != nil { + //logs.Debug("2Reconstruct fail, close pipe", count, session.fecDataShards, session.fecParityShards, er.Error()) + //break //broken data, may be should be closed, now just keep input + } else { + //logs.Debug("Reconstruct ok, input", id) + for i := 0; i < session.fecDataShards; i++ { + if _, have := markTbl[i]; !have { + _len := int(tbl.bytes[i][0]) | (int(tbl.bytes[i][1]) << 8) + ikcp.Ikcp_input(session.kcp, tbl.bytes[i][7:], int(_len)) + //logs.Debug("fec input for mark ok", i, id, _len) + } + } + } + } + delete(session.fecRCacheTbl, id) + //logs.Debug("after del", id, len(c.fecRCacheTbl)) + if session.fecRecvId <= id { + session.fecRecvId = id + 1 + } + } + } else { + if n < 7 { + logs.Debug("recv reset") + go session._Close(false) + break + } else if n == 7 { + status, _, _ := makeDecode(s, session.xor) + if status == Reset || status == ResetAck { + logs.Debug("recv reset2", status) + go session._Close(false) + } + break + } + ikcp.Ikcp_input(session.kcp, s, n) + } + + if waitRecvCache != nil { + ca := *waitRecvCache + tmp := session.processBuffer + for { + hr := ikcp.Ikcp_recv(session.kcp, tmp, ReadBufferSize) + if hr > 0 { + status := tmp[0] + if status == Data { + n := 1 + l := hr + if session.confuseSeed > 0 { + n = 3 + l = (int32(tmp[1]) | (int32(tmp[2]) << 8)) + int32(n) + } + //logs.Debug("try recv", n, l) + waitRecvCache = nil + copy(ca.b, tmp[n:int(l)]) + select { + case ca.c <- (int(l) - n): + case <-session.quitChan: + } + break + } else if status == Fake { + } else { + session.DoAction("recv", status) + } + } else { + break + } + } + } + updateF(10) + case "write": + b := action.args[0].([]byte) + ikcp.Ikcp_send(session.kcp, b, len(b)) + updateF(10) + } + case <-session.quitChan: + break out + case <-updateC: + now := uint32(iclock()) + ikcp.Ikcp_update(session.kcp, now) + callUpdate = false + } + } + }() + select { + case ping <- struct{}{}: + case <-session.quitChan: + } +out: + for { + select { + case action := <-session.do: + switch action.t { + case "quit": + if session.closed { + break + } + session._Close(true) + case "closebegin": + //A tell B to close and wait + time.AfterFunc(time.Millisecond*500, func() { + //logs.Debug("close over, step3", session.LocalAddr().String(), session.RemoteAddr().String()) + go session.DoAction("closeover") + }) + buf := make([]byte, 7) + go session.DoWrite(makeEncode(buf, Close, 0, 0, session.xor)) + case "closeover": + //A call timeover + close(session.closeChan) + break out + case "closeend": + //B call close + session._Close(false) + break out + case "recv": + status := (action.args[0]).(byte) + switch status { + case CloseBack: + //A call from B + //logs.Debug("recv back close, step2", session.LocalAddr().String(), session.RemoteAddr().String()) + go session.DoAction("closeover") + case Close: + if session.closed { + break + } + if session.status != "ok" { + session._Close(false) + } else { + //logs.Debug("recv remote close, step1", session.LocalAddr().String(), session.RemoteAddr().String()) + buf := make([]byte, 7) + go session.DoWrite(makeEncode(buf, CloseBack, 0, 0, session.xor)) + time.AfterFunc(time.Millisecond*500, func() { + //logs.Debug("close remote over, step4", session.LocalAddr().String(), session.RemoteAddr().String()) + if session.closed { + return + } + go session.DoAction("closeend") + }) + } + case Reset: + logs.Debug("recv reset") + go session._Close(false) + case Ping: + default: + if session.status != "ok" { + session.Close() + } + } + } + } + } +} + +func (session *UDPMakeSession) _Close(bFirstCall bool) { + if session.closed { + return + } + session.closed = true + //session.wait.Wait() + go func() { + //logs.Debug("pipe begin close", session.LocalAddr().String(), session.RemoteAddr().String()) + if bFirstCall { + go session.DoAction("closebegin", session.LocalAddr().String()) + <-session.closeChan + } else { + close(session.closeChan) + } + //logs.Debug("pipe end close", session.id) + close(session.quitChan) + if session.listener != nil { + session.listener.remove(session.remote.String()) + } else { + if session.sock != nil { + session.sock.Close() + } + } + }() +} + +func (session *UDPMakeSession) Close() error { + if session.closed { + return nil + } + if session.status != "ok" { + session._Close(false) + return nil + } + go session.DoAction("quit") + <-session.closeChan + return nil +} + +func (session *UDPMakeSession) DoAction2(action string, args ...interface{}) { + //session.wait.Add(1) + //logs.Debug(action, len(args)) + select { + case session.do2 <- Action{t: action, args: args}: + case <-session.quitChan: + //session.wait.Done() + } +} +func (session *UDPMakeSession) DoAction(action string, args ...interface{}) { + //session.wait.Add(1) + //logs.Debug(action, len(args)) + select { + case session.do <- Action{t: action, args: args}: + case <-session.quitChan: + //session.wait.Done() + } +} + +func (session *UDPMakeSession) processInput(s string, n int) { +} + +func (session *UDPMakeSession) LocalAddr() net.Addr { + return session.sock.LocalAddr() +} + +func (session *UDPMakeSession) RemoteAddr() net.Addr { + return net.Addr(session.remote) +} + +func (session *UDPMakeSession) SetDeadline(t time.Time) error { + return session.sock.SetDeadline(t) +} + +func (session *UDPMakeSession) SetReadDeadline(t time.Time) error { + return session.sock.SetReadDeadline(t) +} + +func (session *UDPMakeSession) SetWriteDeadline(t time.Time) error { + return session.sock.SetWriteDeadline(t) +} + +func (session *UDPMakeSession) DoWrite(s []byte) bool { + wc := make(chan struct{}) + select { + case session.checkCanWrite <- wc: + select { + case <-wc: + case <-session.quitChan: + return false + } + session.DoAction2("write", s) + return true + case <-session.quitChan: + return false + } +} + +func (session *UDPMakeSession) Write(b []byte) (n int, err error) { + sendL := len(b) + if sendL == 0 || session.status != "ok" { + return 0, nil + } + var data []byte + if session.confuseSeed <= 0 { + data = make([]byte, sendL+1) + data[0] = Data + copy(data[1:], b) + } else { + remain := WriteBufferSize - sendL - 3 + if remain > 0 && (WriteBufferSize/2-rand.Intn(session.confuseSeed) < 500 || rand.Intn(100) < 5) { + if remain > 1000 { + remain = 1000 + } + remain = rand.Intn(remain / 2) + } else { + remain = 0 + } + data = make([]byte, sendL+remain+3) + data[0] = Data + data[1] = byte(sendL & 0xff) + data[2] = byte((sendL >> 8) & 0xff) + copy(data[3:], b) + //logs.Debug("try send", len(data), sendL) + //copy(data[3+sendL:], b) + } + ok := session.DoWrite(data) + if !ok { + return 0, errors.New("closed") + } + if session.confuseSeed > 0 { + if rand.Intn(WriteBufferSize/2) > session.confuseSeed/2 { + n := rand.Intn(session.confuseSeed) + if n > 10 { + d := make([]byte, n) + d[0] = Fake + session.DoWrite(d) + } + } + } + return sendL, err +} + +//udp read does not relay on the len(p), please make a big enough array to cache data +func (session *UDPMakeSession) Read(p []byte) (n int, err error) { + wc := KcpCache{p, 0, make(chan int)} + select { + case session.recvChan <- wc: + select { + case n = <-wc.c: + case <-session.quitChan: + n = -1 + } + case <-session.quitChan: + n = -1 + } + //logs.Debug("real recv", l, string(b[:l])) + if n == -1 { + return 0, errors.New("force quit for read error") + } else { + return n, nil + } +} + + + +type _reuseTbl struct { + tbl map[int]bool +} + +var currIdMap map[string]int +var reuseTbl map[string]*_reuseTbl + +func GetId(name string) int { + if currIdMap == nil { + currIdMap = make(map[string]int) + currIdMap[name] = 0 + } + i, _ := currIdMap[name] + i++ + if i >= 2147483647 { + i = 0 + } + currIdMap[name] = i + // println("gen new id", currIdMap[name]) + return i +} + +func RmId(name string, id int) { + return +} \ No newline at end of file