diff --git a/server/jsonrpc_blockchain.go b/server/jsonrpc_blockchain.go index 6b0f9ef..de5c457 100644 --- a/server/jsonrpc_blockchain.go +++ b/server/jsonrpc_blockchain.go @@ -65,6 +65,13 @@ func min[Ord constraints.Ordered](x, y Ord) Ord { return y } +func max[Ord constraints.Ordered](x, y Ord) Ord { + if x > y { + return x + } + return y +} + type BlockHeaderElectrum struct { Version uint32 `json:"version"` PrevBlockHash string `json:"prev_block_hash"` diff --git a/server/session.go b/server/session.go index 0ae1b11..9b9fbb7 100644 --- a/server/session.go +++ b/server/session.go @@ -119,6 +119,7 @@ type sessionManager struct { sessionsWait sync.WaitGroup sessionsMax int sessionTimeout time.Duration + manageTicker *time.Ticker db *db.ReadOnlyDBColumnFamily chain *chaincfg.Params // headerSubs are sessions subscribed via 'blockchain.headers.subscribe' @@ -132,6 +133,7 @@ func newSessionManager(db *db.ReadOnlyDBColumnFamily, chain *chaincfg.Params, se sessions: make(sessionMap), sessionsMax: sessionsMax, sessionTimeout: time.Duration(sessionTimeout) * time.Second, + manageTicker: time.NewTicker(time.Duration(max(5, sessionTimeout/20)) * time.Second), db: db, chain: chain, headerSubs: make(sessionMap), @@ -156,17 +158,18 @@ func (sm *sessionManager) stop() { } func (sm *sessionManager) manage() { - sm.sessionsMut.Lock() - for _, sess := range sm.sessions { - if time.Since(sess.lastRecv) > sm.sessionTimeout { - sm.removeSessionLocked(sess) - log.Infof("session %v timed out", sess.addr.String()) + for { + sm.sessionsMut.Lock() + for _, sess := range sm.sessions { + if time.Since(sess.lastRecv) > sm.sessionTimeout { + sm.removeSessionLocked(sess) + log.Infof("session %v timed out", sess.addr.String()) + } } + sm.sessionsMut.Unlock() + // Wait for next management clock tick. + <-sm.manageTicker.C } - sm.sessionsMut.Unlock() - - dur, _ := time.ParseDuration("10s") - time.AfterFunc(dur, func() { sm.manage() }) } func (sm *sessionManager) addSession(conn net.Conn) *session {