optimize updater to a per-session timer for multicore

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

updater.go

@@ -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()
-	}
-}