benjamyn/any-sync
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Fix waiters in streampool

Browse Source
This commit is contained in:
mcrakhman 2022-12-27 19:59:27 +01:00 committed by Mikhail Iudin
parent 270a6bdcee
commit 021342bfe9
No known key found for this signature in database
GPG Key ID: FAAAA8BAABDFF1C0
1 changed files with 38 additions and 47 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

85
common/commonspace/objectsync/streampool.go
View File

@ -1,12 +1,12 @@
package objectsync package syncservice
import ( import (
"context" "context"
"errors" "errors"
"fmt" "fmt"
"github.com/anytypeio/go-anytype-infrastructure-experiments/common/app/ocache"
"github.com/anytypeio/go-anytype-infrastructure-experiments/common/commonspace/spacesyncproto" "github.com/anytypeio/go-anytype-infrastructure-experiments/common/commonspace/spacesyncproto"
"github.com/anytypeio/go-anytype-infrastructure-experiments/common/net/peer" "github.com/anytypeio/go-anytype-infrastructure-experiments/common/net/peer"
"github.com/anytypeio/go-anytype-infrastructure-experiments/common/pkg/ocache"
"go.uber.org/zap" "go.uber.org/zap"
"sync" "sync"
"sync/atomic" "sync/atomic"
@ -16,7 +16,7 @@ import (
var ErrEmptyPeer = errors.New("don't have such a peer") var ErrEmptyPeer = errors.New("don't have such a peer")
var ErrStreamClosed = errors.New("stream is already closed") var ErrStreamClosed = errors.New("stream is already closed")
var maxStreamReaders = 10 var maxSimultaneousOperationsPerStream = 10
var syncWaitPeriod = 2 * time.Second var syncWaitPeriod = 2 * time.Second
var ErrSyncTimeout = errors.New("too long wait on sync receive") var ErrSyncTimeout = errors.New("too long wait on sync receive")
@ -24,8 +24,8 @@ var ErrSyncTimeout = errors.New("too long wait on sync receive")
// StreamPool can be made generic to work with different streams // StreamPool can be made generic to work with different streams
type StreamPool interface { type StreamPool interface {
ocache.ObjectLastUsage ocache.ObjectLastUsage
AddAndReadStreamSync(stream spacesyncproto.ObjectSyncStream) (err error) AddAndReadStreamSync(stream spacesyncproto.SpaceStream) (err error)
AddAndReadStreamAsync(stream spacesyncproto.ObjectSyncStream) (err error) AddAndReadStreamAsync(stream spacesyncproto.SpaceStream) (err error)
SendSync(peerId string, message *spacesyncproto.ObjectSyncMessage) (reply *spacesyncproto.ObjectSyncMessage, err error) SendSync(peerId string, message *spacesyncproto.ObjectSyncMessage) (reply *spacesyncproto.ObjectSyncMessage, err error)
SendAsync(peers []string, message *spacesyncproto.ObjectSyncMessage) (err error) SendAsync(peers []string, message *spacesyncproto.ObjectSyncMessage) (err error)
@ -43,7 +43,7 @@ type responseWaiter struct {
type streamPool struct { type streamPool struct {
sync.Mutex sync.Mutex
peerStreams map[string]spacesyncproto.ObjectSyncStream peerStreams map[string]spacesyncproto.SpaceStream
messageHandler MessageHandler messageHandler MessageHandler
wg *sync.WaitGroup wg *sync.WaitGroup
waiters map[string]responseWaiter waiters map[string]responseWaiter
@ -54,7 +54,7 @@ type streamPool struct {
func newStreamPool(messageHandler MessageHandler) StreamPool { func newStreamPool(messageHandler MessageHandler) StreamPool {
s := &streamPool{ s := &streamPool{
peerStreams: make(map[string]spacesyncproto.ObjectSyncStream), peerStreams: make(map[string]spacesyncproto.SpaceStream),
messageHandler: messageHandler, messageHandler: messageHandler,
waiters: make(map[string]responseWaiter), waiters: make(map[string]responseWaiter),
wg: &sync.WaitGroup{}, wg: &sync.WaitGroup{},
@ -110,7 +110,7 @@ func (s *streamPool) SendSync(
func (s *streamPool) SendAsync(peers []string, message *spacesyncproto.ObjectSyncMessage) (err error) { func (s *streamPool) SendAsync(peers []string, message *spacesyncproto.ObjectSyncMessage) (err error) {
s.lastUsage.Store(time.Now().Unix()) s.lastUsage.Store(time.Now().Unix())
getStreams := func() (streams []spacesyncproto.ObjectSyncStream) { getStreams := func() (streams []spacesyncproto.SpaceStream) {
for _, pId := range peers { for _, pId := range peers {
stream, err := s.getOrDeleteStream(pId) stream, err := s.getOrDeleteStream(pId)
if err != nil { if err != nil {
@ -139,7 +139,7 @@ func (s *streamPool) SendAsync(peers []string, message *spacesyncproto.ObjectSyn
return err return err
} }
func (s *streamPool) getOrDeleteStream(id string) (stream spacesyncproto.ObjectSyncStream, err error) { func (s *streamPool) getOrDeleteStream(id string) (stream spacesyncproto.SpaceStream, err error) {
stream, exists := s.peerStreams[id] stream, exists := s.peerStreams[id]
if !exists { if !exists {
err = ErrEmptyPeer err = ErrEmptyPeer
@ -156,7 +156,7 @@ func (s *streamPool) getOrDeleteStream(id string) (stream spacesyncproto.ObjectS
return return
} }
func (s *streamPool) getAllStreams() (streams []spacesyncproto.ObjectSyncStream) { func (s *streamPool) getAllStreams() (streams []spacesyncproto.SpaceStream) {
s.Lock() s.Lock()
defer s.Unlock() defer s.Unlock()
Loop: Loop:
@ -188,7 +188,7 @@ func (s *streamPool) BroadcastAsync(message *spacesyncproto.ObjectSyncMessage) (
return nil return nil
} }
func (s *streamPool) AddAndReadStreamAsync(stream spacesyncproto.ObjectSyncStream) (err error) { func (s *streamPool) AddAndReadStreamAsync(stream spacesyncproto.SpaceStream) (err error) {
peerId, err := s.addStream(stream) peerId, err := s.addStream(stream)
if err != nil { if err != nil {
return return
@ -197,7 +197,7 @@ func (s *streamPool) AddAndReadStreamAsync(stream spacesyncproto.ObjectSyncStrea
return return
} }
func (s *streamPool) AddAndReadStreamSync(stream spacesyncproto.ObjectSyncStream) (err error) { func (s *streamPool) AddAndReadStreamSync(stream spacesyncproto.SpaceStream) (err error) {
peerId, err := s.addStream(stream) peerId, err := s.addStream(stream)
if err != nil { if err != nil {
return return
@ -205,7 +205,7 @@ func (s *streamPool) AddAndReadStreamSync(stream spacesyncproto.ObjectSyncStream
return s.readPeerLoop(peerId, stream) return s.readPeerLoop(peerId, stream)
} }
func (s *streamPool) addStream(stream spacesyncproto.ObjectSyncStream) (peerId string, err error) { func (s *streamPool) addStream(stream spacesyncproto.SpaceStream) (peerId string, err error) {
s.Lock() s.Lock()
peerId, err = peer.CtxPeerId(stream.Context()) peerId, err = peer.CtxPeerId(stream.Context())
if err != nil { if err != nil {
@ -245,74 +245,65 @@ func (s *streamPool) Close() (err error) {
return nil return nil
} }
func (s *streamPool) readPeerLoop(peerId string, stream spacesyncproto.ObjectSyncStream) (err error) { func (s *streamPool) readPeerLoop(peerId string, stream spacesyncproto.SpaceStream) (err error) {
var ( log := log.With(zap.String("peerId", peerId))
log = log.With(zap.String("peerId", peerId))
queue = NewActionQueue(maxStreamReaders, 100)
)
queue.Run()
defer func() { defer func() {
log.Debug("stopped reading stream from peer") log.Debug("stopped reading stream from peer")
s.removePeer(peerId, stream) s.removePeer(peerId, stream)
queue.Close()
s.wg.Done() s.wg.Done()
}() }()
log.Debug("started reading stream from peer") log.Debug("started reading stream from peer")
stopWaiter := func(msg *spacesyncproto.ObjectSyncMessage) bool { limiter := make(chan struct{}, maxSimultaneousOperationsPerStream)
s.waitersMx.Lock() for i := 0; i < maxSimultaneousOperationsPerStream; i++ {
waiter, exists := s.waiters[msg.ReplyId] limiter <- struct{}{}
if exists {
delete(s.waiters, msg.ReplyId)
s.waitersMx.Unlock()
waiter.ch <- msg
return true
}
s.waitersMx.Unlock()
return false
} }
process := func(msg *spacesyncproto.ObjectSyncMessage) error { process := func(msg *spacesyncproto.ObjectSyncMessage) {
log := log.With(zap.String("replyId", msg.ReplyId), zap.String("object id", msg.ObjectId)) log := log.With(zap.String("replyId", msg.ReplyId), zap.String("object id", msg.ObjectId))
log.Debug("getting message with reply id") log.Debug("getting message with reply id")
err = s.messageHandler(stream.Context(), peerId, msg) err = s.messageHandler(stream.Context(), peerId, msg)
if err != nil { if err != nil {
log.With(zap.Error(err)).Debug("message handling failed") log.With(zap.Error(err)).Debug("message handling failed")
} }
return nil
} }
for { for {
select {
case <-stream.Context().Done():
return
default:
}
var msg *spacesyncproto.ObjectSyncMessage var msg *spacesyncproto.ObjectSyncMessage
msg, err = stream.Recv() msg, err = stream.Recv()
s.lastUsage.Store(time.Now().Unix()) s.lastUsage.Store(time.Now().Unix())
if err != nil { if err != nil {
stream.Close()
return return
} }
if msg.ReplyId != "" { if msg.ReplyId != "" {
// then we can send it directly to waiters without adding to queue or starting a reader s.waitersMx.Lock()
if stopWaiter(msg) { waiter, exists := s.waiters[msg.ReplyId]
if exists {
delete(s.waiters, msg.ReplyId)
s.waitersMx.Unlock()
waiter.ch <- msg
continue continue
} }
s.waitersMx.Unlock()
log.With(zap.String("replyId", msg.ReplyId)).Debug("reply id does not exist") log.With(zap.String("replyId", msg.ReplyId)).Debug("reply id does not exist")
} }
queue.Send(func() error { select {
return process(msg) case <-limiter:
}) case <-stream.Context().Done():
return
}
go func(msg *spacesyncproto.ObjectSyncMessage) {
process(msg)
limiter <- struct{}{}
}(msg)
} }
} }
func (s *streamPool) removePeer(peerId string, stream spacesyncproto.ObjectSyncStream) (err error) { func (s *streamPool) removePeer(peerId string, stream spacesyncproto.SpaceStream) (err error) {
s.Lock() s.Lock()
defer s.Unlock() defer s.Unlock()
mapStream, ok := s.peerStreams[peerId] mapStream, ok := s.peerStreams[peerId]