mirror/erigon-pulse
1
0
Fork 0
mirror of https://gitlab.com/pulsechaincom/erigon-pulse.git synced 2024-12-22 03:30:37 +00:00
Code Issues Actions Packages Projects Releases Wiki Activity

ws unfreeze (#7724)

Browse Source 
attempt to address next issue: 

> when I'm having a lot of websocket connections the node is freezing
and then it needs like 10 mins to sync. Then if I keep pushing requests
it falls out of sync all the time
This commit is contained in:
Alex Sharov 2023-06-14 16:25:05 +07:00 committed by GitHub
parent 8b312d54e8
commit 1b14785d79
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
2 changed files with 19 additions and 43 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

20
cmd/rpcdaemon/commands/eth_filters.go
View File

@ -145,12 +145,11 @@ func (api *APIImpl) NewHeads(ctx context.Context) (*rpc.Subscription, error) {
if h != nil {
err := notifier.Notify(rpcSub.ID, h)
if err != nil {
log.Warn("error while notifying subscription", "err", err)
return
log.Warn("[rpc] error while notifying subscription", "err", err)
}
}
if !ok {
log.Warn("new heads channel was closed")
log.Warn("[rpc] new heads channel was closed")
return
}
case <-rpcSub.Err():
@ -186,13 +185,12 @@ func (api *APIImpl) NewPendingTransactions(ctx context.Context) (*rpc.Subscripti
if t != nil {
err := notifier.Notify(rpcSub.ID, t.Hash())
if err != nil {
log.Warn("error while notifying subscription", "err", err)
return
log.Warn("[rpc] error while notifying subscription", "err", err)
}
}
}
if !ok {
log.Warn("new pending transactions channel was closed")
log.Warn("[rpc] new pending transactions channel was closed")
return
}
case <-rpcSub.Err():
@ -228,13 +226,12 @@ func (api *APIImpl) NewPendingTransactionsWithBody(ctx context.Context) (*rpc.Su
if t != nil {
err := notifier.Notify(rpcSub.ID, t)
if err != nil {
log.Warn("error while notifying subscription", "err", err)
return
log.Warn("[rpc] error while notifying subscription", "err", err)
}
}
}
if !ok {
log.Warn("new pending transactions channel was closed")
log.Warn("[rpc] new pending transactions channel was closed")
return
}
case <-rpcSub.Err():
@ -269,12 +266,11 @@ func (api *APIImpl) Logs(ctx context.Context, crit filters.FilterCriteria) (*rpc
if h != nil {
err := notifier.Notify(rpcSub.ID, h)
if err != nil {
log.Warn("error while notifying subscription", "err", err)
return
log.Warn("[rpc] error while notifying subscription", "err", err)
}
}
if !ok {
log.Warn("log channel was closed")
log.Warn("[rpc] log channel was closed")
return
}
case <-rpcSub.Err():

42
turbo/rpchelper/subscription.go
View File

@ -1,8 +1,8 @@
package rpchelper
import (
"context"
"sync"
"sync/atomic"
)
// a simple interface for subscriptions for rpc helper
@ -12,12 +12,8 @@ type Sub[T any] interface {
}
type chan_sub[T any] struct {
ch chan T
closed chan struct{}
ctx context.Context
cn context.CancelFunc
ch chan T
closed atomic.Bool
}
// buffered channel
@ -28,38 +24,23 @@ func newChanSub[T any](size int) *chan_sub[T] {
}
o := &chan_sub[T]{}
o.ch = make(chan T, size)
o.closed = make(chan struct{})
o.ctx, o.cn = context.WithCancel(context.Background())
return o
}
func (s *chan_sub[T]) Send(x T) {
if s.closed.Load() {
return
}
select {
// if the output buffer is empty, send
case s.ch <- x:
// if sub is canceled, dispose message
case <-s.ctx.Done():
// the sub is overloaded, dispose message
default:
default: // the sub is overloaded, dispose message
}
}
func (s *chan_sub[T]) Close() {
select {
case <-s.ctx.Done():
if swapped := s.closed.CompareAndSwap(false, true); !swapped {
return
default:
}
// its possible for multiple goroutines to get to this point, if Close is called twice at the same time
// close the context - allows any sends to exit
s.cn()
select {
case s.closed <- struct{}{}:
// but it is not possible for multiple goroutines to get to this point
// drain the channel
for range s.ch {
}
close(s.ch)
default:
}
close(s.ch)
}
func NewSyncMap[K comparable, T any]() *SyncMap[K, T] {
@ -117,8 +98,7 @@ func (m *SyncMap[K, T]) Range(fn func(k K, v T) error) error {
return nil
}
func (m *SyncMap[K, T]) Delete(k K) (T, bool) {
var t T
func (m *SyncMap[K, T]) Delete(k K) (t T, deleted bool) {
m.mu.Lock()
defer m.mu.Unlock()
val, ok := m.m[k]