[#1170] pkg/morph: Add worker pool

Add worker pool to the listener to prevent blocking. It is used only for
notary notifications and new block events handling since it uses RPC
calls. That may lead to the deadlock state: neo-go cannot send RPC until
notification channel is read but notification channel cannot be read since
neo-go client cannot send RPC.

Signed-off-by: Pavel Karpy <carpawell@nspcc.ru>
This commit is contained in:
Pavel Karpy 2022-04-01 17:18:14 +03:00 committed by Alex Vanin
parent 4ed5b1ceef
commit 01e69f2f7a
3 changed files with 56 additions and 7 deletions

View file

@ -173,6 +173,14 @@ func waitNotaryDeposit(c *cfg, tx util.Uint256) error {
} }
func listenMorphNotifications(c *cfg) { func listenMorphNotifications(c *cfg) {
// listenerPoolCap is a capacity of a
// worker pool inside the listener. It
// is used to prevent blocking in neo-go:
// the client cannot make RPC requests if
// the notification channel is not being
// read by another goroutine.
const listenerPoolCap = 10
var ( var (
err error err error
subs subscriber.Subscriber subs subscriber.Subscriber
@ -194,6 +202,7 @@ func listenMorphNotifications(c *cfg) {
lis, err := event.NewListener(event.ListenerParams{ lis, err := event.NewListener(event.ListenerParams{
Logger: c.log, Logger: c.log,
Subscriber: subs, Subscriber: subs,
WorkerPoolCapacity: listenerPoolCap,
}) })
fatalOnErr(err) fatalOnErr(err)

View file

@ -921,6 +921,14 @@ func New(ctx context.Context, log *zap.Logger, cfg *viper.Viper, errChan chan<-
} }
func createListener(ctx context.Context, cli *client.Client, p *chainParams) (event.Listener, error) { func createListener(ctx context.Context, cli *client.Client, p *chainParams) (event.Listener, error) {
// listenerPoolCap is a capacity of a
// worker pool inside the listener. It
// is used to prevent blocking in neo-go:
// the client cannot make RPC requests if
// the notification channel is not being
// read by another goroutine.
const listenerPoolCap = 10
var ( var (
sub subscriber.Subscriber sub subscriber.Subscriber
err error err error
@ -936,8 +944,9 @@ func createListener(ctx context.Context, cli *client.Client, p *chainParams) (ev
} }
listener, err := event.NewListener(event.ListenerParams{ listener, err := event.NewListener(event.ListenerParams{
Logger: p.log, Logger: p.log.With(zap.String("chain", p.name)),
Subscriber: sub, Subscriber: sub,
WorkerPoolCapacity: listenerPoolCap,
}) })
if err != nil { if err != nil {
return nil, err return nil, err

View file

@ -11,6 +11,7 @@ import (
"github.com/nspcc-dev/neo-go/pkg/util" "github.com/nspcc-dev/neo-go/pkg/util"
"github.com/nspcc-dev/neofs-node/pkg/morph/client" "github.com/nspcc-dev/neofs-node/pkg/morph/client"
"github.com/nspcc-dev/neofs-node/pkg/morph/subscriber" "github.com/nspcc-dev/neofs-node/pkg/morph/subscriber"
"github.com/panjf2000/ants/v2"
"go.uber.org/zap" "go.uber.org/zap"
) )
@ -85,6 +86,8 @@ type ListenerParams struct {
Logger *zap.Logger Logger *zap.Logger
Subscriber subscriber.Subscriber Subscriber subscriber.Subscriber
WorkerPoolCapacity int
} }
type listener struct { type listener struct {
@ -108,6 +111,8 @@ type listener struct {
subscriber subscriber.Subscriber subscriber subscriber.Subscriber
blockHandlers []BlockHandler blockHandlers []BlockHandler
pool *ants.Pool
} }
const newListenerFailMsg = "could not instantiate Listener" const newListenerFailMsg = "could not instantiate Listener"
@ -258,7 +263,12 @@ loop:
continue loop continue loop
} }
if err = l.pool.Submit(func() {
l.parseAndHandleNotary(notaryEvent) l.parseAndHandleNotary(notaryEvent)
}); err != nil {
l.log.Warn("listener worker pool drained",
zap.Int("capacity", l.pool.Cap()))
}
case b, ok := <-blockChan: case b, ok := <-blockChan:
if !ok { if !ok {
l.log.Warn("stop event listener by block channel") l.log.Warn("stop event listener by block channel")
@ -272,9 +282,14 @@ loop:
continue loop continue loop
} }
if err = l.pool.Submit(func() {
for i := range l.blockHandlers { for i := range l.blockHandlers {
l.blockHandlers[i](b) l.blockHandlers[i](b)
} }
}); err != nil {
l.log.Warn("listener worker pool drained",
zap.Int("capacity", l.pool.Cap()))
}
} }
} }
} }
@ -586,6 +601,11 @@ func (l *listener) RegisterBlockHandler(handler BlockHandler) {
// NewListener create the notification event listener instance and returns Listener interface. // NewListener create the notification event listener instance and returns Listener interface.
func NewListener(p ListenerParams) (Listener, error) { func NewListener(p ListenerParams) (Listener, error) {
// defaultPoolCap is a default worker
// pool capacity if it was not specified
// via params
const defaultPoolCap = 10
switch { switch {
case p.Logger == nil: case p.Logger == nil:
return nil, fmt.Errorf("%s: %w", newListenerFailMsg, errNilLogger) return nil, fmt.Errorf("%s: %w", newListenerFailMsg, errNilLogger)
@ -593,10 +613,21 @@ func NewListener(p ListenerParams) (Listener, error) {
return nil, fmt.Errorf("%s: %w", newListenerFailMsg, errNilSubscriber) return nil, fmt.Errorf("%s: %w", newListenerFailMsg, errNilSubscriber)
} }
poolCap := p.WorkerPoolCapacity
if poolCap == 0 {
poolCap = defaultPoolCap
}
pool, err := ants.NewPool(poolCap, ants.WithNonblocking(true))
if err != nil {
return nil, fmt.Errorf("could not init worker pool: %w", err)
}
return &listener{ return &listener{
notificationParsers: make(map[scriptHashWithType]NotificationParser), notificationParsers: make(map[scriptHashWithType]NotificationParser),
notificationHandlers: make(map[scriptHashWithType][]Handler), notificationHandlers: make(map[scriptHashWithType][]Handler),
log: p.Logger, log: p.Logger,
subscriber: p.Subscriber, subscriber: p.Subscriber,
pool: pool,
}, nil }, nil
} }