From d002baf70a62b817f7585a02f6ae2ea28b2adbab Mon Sep 17 00:00:00 2001 From: xtaci Date: Thu, 26 Sep 2019 10:53:49 +0800 Subject: [PATCH] optimize updater to a per-session timer for multicore --- sess.go | 47 +++++++++++++----------- updater.go | 104 ----------------------------------------------------- 2 files changed, 26 insertions(+), 125 deletions(-) delete mode 100644 updater.go diff --git a/sess.go b/sess.go index e97d6a0..5fa4459 100644 --- a/sess.go +++ b/sess.go @@ -62,11 +62,10 @@ func init() { type ( // UDPSession defines a KCP session implemented by UDP UDPSession struct { - updaterIdx int // record slice index in updater - conn net.PacketConn // the underlying packet connection - kcp *KCP // KCP ARQ protocol - l *Listener // pointing to the Listener object if it's been accepted by a Listener - block BlockCrypt // block encryption object + conn net.PacketConn // the underlying packet connection + kcp *KCP // KCP ARQ protocol + l *Listener // pointing to the Listener object if it's been accepted by a Listener + block BlockCrypt // block encryption object // kcp receiving is based on packets // recvbuf turns packets into stream @@ -175,10 +174,6 @@ func newUDPSession(conv uint32, dataShards, parityShards int, l *Listener, conn }) sess.kcp.ReserveBytes(sess.headerSize) - // register current session to the global updater, - // which call sess.update() periodically. - updater.addSession(sess) - if sess.l == nil { // it's a client connection go sess.readLoop() atomic.AddUint64(&DefaultSnmp.ActiveOpens, 1) @@ -186,6 +181,9 @@ func newUDPSession(conv uint32, dataShards, parityShards int, l *Listener, conn atomic.AddUint64(&DefaultSnmp.PassiveOpens, 1) } + // start per-session updater + go sess.updater() + currestab := atomic.AddUint64(&DefaultSnmp.CurrEstab, 1) maxconn := atomic.LoadUint64(&DefaultSnmp.MaxConn) if currestab > maxconn { @@ -349,8 +347,6 @@ func (s *UDPSession) Close() error { }) if once { - // remove from updater - updater.removeSession(s) atomic.AddUint64(&DefaultSnmp.CurrEstab, ^uint64(0)) if s.l != nil { // belongs to listener @@ -566,17 +562,26 @@ func (s *UDPSession) output(buf []byte) { } } -// kcp update, returns interval for next calling -func (s *UDPSession) update() (interval time.Duration) { - s.mu.Lock() - interval = time.Duration(s.kcp.flush(false)) * time.Millisecond - waitsnd := s.kcp.WaitSnd() - if waitsnd < int(s.kcp.snd_wnd) && waitsnd < int(s.kcp.rmt_wnd) { - s.notifyWriteEvent() +// sess updater to trigger protocol +func (s *UDPSession) updater() { + timer := time.NewTimer(0) + for { + select { + case <-timer.C: + s.mu.Lock() + interval := time.Duration(s.kcp.flush(false)) * time.Millisecond + waitsnd := s.kcp.WaitSnd() + if waitsnd < int(s.kcp.snd_wnd) && waitsnd < int(s.kcp.rmt_wnd) { + s.notifyWriteEvent() + } + s.uncork() + s.mu.Unlock() + timer.Reset(interval) + case <-s.die: + timer.Stop() + return + } } - s.uncork() - s.mu.Unlock() - return } // GetConv gets conversation id of a session diff --git a/updater.go b/updater.go deleted file mode 100644 index b36d1b9..0000000 --- a/updater.go +++ /dev/null @@ -1,104 +0,0 @@ -package kcp - -import ( - "container/heap" - "sync" - "time" -) - -var updater updateHeap - -func init() { - updater.init() - go updater.updateTask() -} - -// entry contains a session update info -type entry struct { - ts time.Time - s *UDPSession -} - -// a global heap managed kcp.flush() caller -type updateHeap struct { - entries []entry - mu sync.Mutex - chWakeUp chan struct{} -} - -func (h *updateHeap) Len() int { return len(h.entries) } -func (h *updateHeap) Less(i, j int) bool { return h.entries[i].ts.Before(h.entries[j].ts) } -func (h *updateHeap) Swap(i, j int) { - h.entries[i], h.entries[j] = h.entries[j], h.entries[i] - h.entries[i].s.updaterIdx = i - h.entries[j].s.updaterIdx = j -} - -func (h *updateHeap) Push(x interface{}) { - h.entries = append(h.entries, x.(entry)) - n := len(h.entries) - h.entries[n-1].s.updaterIdx = n - 1 -} - -func (h *updateHeap) Pop() interface{} { - n := len(h.entries) - x := h.entries[n-1] - h.entries[n-1].s.updaterIdx = -1 - h.entries[n-1] = entry{} // manual set nil for GC - h.entries = h.entries[0 : n-1] - return x -} - -func (h *updateHeap) init() { - h.chWakeUp = make(chan struct{}, 1) -} - -func (h *updateHeap) addSession(s *UDPSession) { - h.mu.Lock() - heap.Push(h, entry{time.Now(), s}) - h.mu.Unlock() - h.wakeup() -} - -func (h *updateHeap) removeSession(s *UDPSession) { - h.mu.Lock() - if s.updaterIdx != -1 { - heap.Remove(h, s.updaterIdx) - } - h.mu.Unlock() -} - -func (h *updateHeap) wakeup() { - select { - case h.chWakeUp <- struct{}{}: - default: - } -} - -func (h *updateHeap) updateTask() { - timer := time.NewTimer(0) - for { - select { - case <-timer.C: - case <-h.chWakeUp: - } - - h.mu.Lock() - hlen := h.Len() - for i := 0; i < hlen; i++ { - entry := &h.entries[0] - if !time.Now().Before(entry.ts) { - interval := entry.s.update() - entry.ts = time.Now().Add(interval) - heap.Fix(h, 0) - } else { - break - } - } - - if hlen > 0 { - timer.Reset(h.entries[0].ts.Sub(time.Now())) - } - h.mu.Unlock() - } -}