fix typo

Component can provide optional fields for the start/stop logger

app/app.go

@@ -4,14 +4,15 @@ import (
 	"context"
 	"errors"
 	"fmt"
-	"github.com/anyproto/any-sync/app/logger"
-	"go.uber.org/zap"
 	"os"
 	"runtime"
 	"runtime/debug"
 	"strings"
 	"sync"
 	"time"
+
+	"github.com/anyproto/any-sync/app/logger"
+	"go.uber.org/zap"
 )
 
 var (
@@ -48,6 +49,12 @@ type ComponentRunnable interface {
 	Close(ctx context.Context) (err error)
 }
 
+type ComponentLogFieldsGetter interface {
+	// GetLogFields returns additional useful fields for logs to debug long app start/stop duration or something else in the future
+	// You don't need to provide the component name in the field's Key, because it will be added automatically
+	GetLogFields() []zap.Field
+}
+
 type ComponentStatable interface {
 	StateChange(state int)
 }
@@ -219,7 +226,7 @@ func (app *App) Start(ctx context.Context) (err error) {
 		case <-done:
 			return
 		case <-time.After(StartWarningAfter):
-			l := statLogger(app.stopStat, log).With(zap.String("in_progress", currentComponentStarting))
+			l := app.statLogger(app.startStat, log).With(zap.String("in_progress", currentComponentStarting))
 			l.Warn("components start in progress")
 		}
 	}()
@@ -254,7 +261,8 @@ func (app *App) Start(ctx context.Context) (err error) {
 	}
 
 	close(done)
-	l := statLogger(app.stopStat, log)
+	l := app.statLogger(app.startStat, log)
+
 	if app.startStat.SpentMsTotal > StartWarningAfter.Milliseconds() {
 		l.Warn("all components started")
 	}
@@ -282,13 +290,22 @@ func stackAllGoroutines() []byte {
 	}
 }
 
-func statLogger(stat Stat, ctxLogger logger.CtxLogger) logger.CtxLogger {
+func (app *App) statLogger(stat Stat, ctxLogger logger.CtxLogger) logger.CtxLogger {
 	l := ctxLogger
 	for k, v := range stat.SpentMsPerComp {
 		l = l.With(zap.Int64(k, v))
 	}
 	l = l.With(zap.Int64("total", stat.SpentMsTotal))
 
+	for _, s := range app.components {
+		// get optional log fields from components
+		if compLog, ok := s.(ComponentLogFieldsGetter); ok {
+			for _, val := range compLog.GetLogFields() {
+				val.Key = s.Name() + "_" + val.Key
+				l = l.With(val)
+			}
+		}
+	}
 	return l
 }
 
@@ -307,7 +324,7 @@ func (app *App) Close(ctx context.Context) error {
 		case <-done:
 			return
 		case <-time.After(StopWarningAfter):
-			statLogger(app.stopStat, log).
+			app.statLogger(app.stopStat, log).
 				With(zap.String("in_progress", currentComponentStopping)).
 				Warn("components close in progress")
 		}
@@ -341,7 +358,7 @@ func (app *App) Close(ctx context.Context) error {
 		return errors.New(strings.Join(errs, "\n"))
 	}
 
-	l := statLogger(app.stopStat, log)
+	l := app.statLogger(app.stopStat, log)
 	if app.stopStat.SpentMsTotal > StopWarningAfter.Milliseconds() {
 		l.Warn("all components have been closed")
 	}

commonspace/object/tree/objecttree/rawloader.go

@@ -2,11 +2,10 @@ package objecttree
 
 import (
 	"context"
-	"time"
-
 	"github.com/anyproto/any-sync/commonspace/object/tree/treechangeproto"
 	"github.com/anyproto/any-sync/commonspace/object/tree/treestorage"
 	"github.com/anyproto/any-sync/util/slice"
+	"time"
 )
 
 type rawChangeLoader struct {
@@ -23,7 +22,6 @@ type rawCacheEntry struct {
 	change    *Change
 	rawChange *treechangeproto.RawTreeChangeWithId
 	position  int
-	removed   bool
 }
 
 func newStorageLoader(treeStorage treestorage.TreeStorage, changeBuilder ChangeBuilder) *rawChangeLoader {
@@ -128,6 +126,7 @@ func (r *rawChangeLoader) loadFromStorage(commonSnapshot string, heads, breakpoi
 		if err != nil {
 			continue
 		}
+		entry.position = -1
 		r.cache[b] = entry
 		existingBreakpoints = append(existingBreakpoints, b)
 	}
@@ -136,7 +135,8 @@ func (r *rawChangeLoader) loadFromStorage(commonSnapshot string, heads, breakpoi
 	dfs := func(
 		commonSnapshot string,
 		heads []string,
-		shouldVisit func(entry rawCacheEntry, mapExists bool) bool,
+		startCounter int,
+		shouldVisit func(counter int, mapExists bool) bool,
 		visit func(entry rawCacheEntry) rawCacheEntry) bool {
 
 		// resetting stack
@@ -150,7 +150,7 @@ func (r *rawChangeLoader) loadFromStorage(commonSnapshot string, heads, breakpoi
 			r.idStack = r.idStack[:len(r.idStack)-1]
 
 			entry, exists := r.cache[id]
-			if !shouldVisit(entry, exists) {
+			if !shouldVisit(entry.position, exists) {
 				continue
 			}
 			if id == commonSnapshot {
@@ -159,6 +159,7 @@ func (r *rawChangeLoader) loadFromStorage(commonSnapshot string, heads, breakpoi
 			}
 			if !exists {
 				entry, err = r.loadEntry(id)
+				entry.position = -1
 				if err != nil {
 					continue
 				}
@@ -173,7 +174,7 @@ func (r *rawChangeLoader) loadFromStorage(commonSnapshot string, heads, breakpoi
 					break
 				}
 				prevEntry, exists := r.cache[prev]
-				if !shouldVisit(prevEntry, exists) {
+				if !shouldVisit(prevEntry.position, exists) {
 					continue
 				}
 				r.idStack = append(r.idStack, prev)
@@ -186,8 +187,8 @@ func (r *rawChangeLoader) loadFromStorage(commonSnapshot string, heads, breakpoi
 	r.idStack = append(r.idStack, heads...)
 	var buffer []*treechangeproto.RawTreeChangeWithId
 
-	rootVisited := dfs(commonSnapshot, heads,
-		func(_ rawCacheEntry, mapExists bool) bool {
+	rootVisited := dfs(commonSnapshot, heads, 0,
+		func(counter int, mapExists bool) bool {
 			return !mapExists
 		},
 		func(entry rawCacheEntry) rawCacheEntry {
@@ -212,13 +213,11 @@ func (r *rawChangeLoader) loadFromStorage(commonSnapshot string, heads, breakpoi
 	}
 
 	// marking all visited as nil
-	dfs(commonSnapshot, existingBreakpoints,
-		func(entry rawCacheEntry, mapExists bool) bool {
-			// only going through already loaded changes
-			return mapExists && !entry.removed
+	dfs(commonSnapshot, existingBreakpoints, len(buffer),
+		func(counter int, mapExists bool) bool {
+			return !mapExists || counter < len(buffer)
 		},
 		func(entry rawCacheEntry) rawCacheEntry {
-			entry.removed = true
 			if entry.position != -1 {
 				buffer[entry.position] = nil
 			}
@@ -249,7 +248,6 @@ func (r *rawChangeLoader) loadEntry(id string) (entry rawCacheEntry, err error)
 	entry = rawCacheEntry{
 		change:    change,
 		rawChange: rawChange,
-		position:  -1,
 	}
 	return
 }

net/peer/limiter.go

@@ -1,18 +0,0 @@
-package peer
-
-import (
-	"time"
-)
-
-type limiter struct {
-	startThreshold int
-	slowDownStep   time.Duration
-}
-
-func (l limiter) wait(count int) <-chan time.Time {
-	if count > l.startThreshold {
-		wait := l.slowDownStep * time.Duration(count-l.startThreshold)
-		return time.After(wait)
-	}
-	return nil
-}

net/peer/peer.go

@@ -19,7 +19,6 @@ import (
 	"storj.io/drpc/drpcstream"
 	"storj.io/drpc/drpcwire"
 	"sync"
-	"sync/atomic"
 	"time"
 )
 
@@ -36,15 +35,8 @@ func NewPeer(mc transport.MultiConn, ctrl connCtrl) (p Peer, err error) {
 		active:    map[*subConn]struct{}{},
 		MultiConn: mc,
 		ctrl:      ctrl,
-		limiter: limiter{
-			// start throttling after 10 sub conns
-			startThreshold: 10,
-			slowDownStep:   time.Millisecond * 100,
-		},
-		subConnRelease: make(chan drpc.Conn),
-		created:        time.Now(),
+		created:   time.Now(),
 	}
-	pr.acceptCtx, pr.acceptCtxCancel = context.WithCancel(context.Background())
 	if pr.id, err = CtxPeerId(ctx); err != nil {
 		return
 	}
@@ -78,22 +70,13 @@ type peer struct {
 	ctrl connCtrl
 
 	// drpc conn pool
-	// outgoing
-	inactive         []*subConn
-	active           map[*subConn]struct{}
-	subConnRelease   chan drpc.Conn
-	openingWaitCount atomic.Int32
-
-	incomingCount atomic.Int32
-	acceptCtx     context.Context
-
-	acceptCtxCancel context.CancelFunc
-
-	limiter limiter
+	inactive []*subConn
+	active   map[*subConn]struct{}
 
 	mu sync.Mutex
 
 	created time.Time
+
 	transport.MultiConn
 }
 
@@ -104,20 +87,7 @@ func (p *peer) Id() string {
 func (p *peer) AcquireDrpcConn(ctx context.Context) (drpc.Conn, error) {
 	p.mu.Lock()
 	if len(p.inactive) == 0 {
-		wait := p.limiter.wait(len(p.active) + int(p.openingWaitCount.Load()))
 		p.mu.Unlock()
-		if wait != nil {
-			p.openingWaitCount.Add(1)
-			defer p.openingWaitCount.Add(-1)
-			// throttle new connection opening
-			select {
-			case <-ctx.Done():
-				return nil, ctx.Err()
-			case dconn := <-p.subConnRelease:
-				return dconn, nil
-			case <-wait:
-			}
-		}
 		dconn, err := p.openDrpcConn(ctx)
 		if err != nil {
 			return nil, err
@@ -140,21 +110,6 @@ func (p *peer) AcquireDrpcConn(ctx context.Context) (drpc.Conn, error) {
 }
 
 func (p *peer) ReleaseDrpcConn(conn drpc.Conn) {
-	// do nothing if it's closed connection
-	select {
-	case <-conn.Closed():
-		return
-	default:
-	}
-
-	// try to send this connection to acquire if anyone is waiting for it
-	select {
-	case p.subConnRelease <- conn:
-		return
-	default:
-	}
-
-	// return to pool
 	p.mu.Lock()
 	defer p.mu.Unlock()
 	sc, ok := conn.(*subConn)
@@ -207,21 +162,12 @@ func (p *peer) acceptLoop() {
 		}
 	}()
 	for {
-		if wait := p.limiter.wait(int(p.incomingCount.Load())); wait != nil {
-			select {
-			case <-wait:
-			case <-p.acceptCtx.Done():
-				return
-			}
-		}
 		conn, err := p.Accept()
 		if err != nil {
 			exitErr = err
 			return
 		}
 		go func() {
-			p.incomingCount.Add(1)
-			defer p.incomingCount.Add(-1)
 			serveErr := p.serve(conn)
 			if serveErr != io.EOF && serveErr != transport.ErrConnClosed {
 				log.InfoCtx(p.Context(), "serve connection error", zap.Error(serveErr))

net/peer/peer_test.go

@@ -12,8 +12,6 @@ import (
 	"io"
 	"net"
 	_ "net/http/pprof"
-	"storj.io/drpc"
-	"storj.io/drpc/drpcconn"
 	"testing"
 	"time"
 )
@@ -21,86 +19,32 @@ import (
 var ctx = context.Background()
 
 func TestPeer_AcquireDrpcConn(t *testing.T) {
-	t.Run("generic", func(t *testing.T) {
-		fx := newFixture(t, "p1")
-		defer fx.finish()
-		in, out := net.Pipe()
-		go func() {
-			handshake.IncomingProtoHandshake(ctx, out, defaultProtoChecker)
-		}()
-		defer out.Close()
-		fx.mc.EXPECT().Open(gomock.Any()).Return(in, nil)
-		dc, err := fx.AcquireDrpcConn(ctx)
-		require.NoError(t, err)
-		assert.NotEmpty(t, dc)
-		defer dc.Close()
-
-		assert.Len(t, fx.active, 1)
-		assert.Len(t, fx.inactive, 0)
-
-		fx.ReleaseDrpcConn(dc)
-
-		assert.Len(t, fx.active, 0)
-		assert.Len(t, fx.inactive, 1)
-
-		dc, err = fx.AcquireDrpcConn(ctx)
-		require.NoError(t, err)
-		assert.NotEmpty(t, dc)
-		assert.Len(t, fx.active, 1)
-		assert.Len(t, fx.inactive, 0)
-	})
-	t.Run("closed sub conn", func(t *testing.T) {
-		fx := newFixture(t, "p1")
-		defer fx.finish()
-
-		closedIn, _ := net.Pipe()
-		dc := drpcconn.New(closedIn)
-		fx.ReleaseDrpcConn(&subConn{Conn: dc})
-		dc.Close()
-
-		in, out := net.Pipe()
-		go func() {
-			handshake.IncomingProtoHandshake(ctx, out, defaultProtoChecker)
-		}()
-		defer out.Close()
-		fx.mc.EXPECT().Open(gomock.Any()).Return(in, nil)
-		_, err := fx.AcquireDrpcConn(ctx)
-		require.NoError(t, err)
-	})
-}
-
-func TestPeer_DrpcConn_OpenThrottling(t *testing.T) {
 	fx := newFixture(t, "p1")
 	defer fx.finish()
-
-	acquire := func() (func(), drpc.Conn, error) {
-		in, out := net.Pipe()
-		go func() {
-			_, err := handshake.IncomingProtoHandshake(ctx, out, defaultProtoChecker)
-			require.NoError(t, err)
-		}()
-
-		fx.mc.EXPECT().Open(gomock.Any()).Return(in, nil)
-		dconn, err := fx.AcquireDrpcConn(ctx)
-		return func() { out.Close() }, dconn, err
-	}
-
-	var conCount = fx.limiter.startThreshold + 3
-	var conns []drpc.Conn
-	for i := 0; i < conCount; i++ {
-		cc, dc, err := acquire()
-		require.NoError(t, err)
-		defer cc()
-		conns = append(conns, dc)
-	}
-
+	in, out := net.Pipe()
 	go func() {
-		time.Sleep(fx.limiter.slowDownStep)
-		fx.ReleaseDrpcConn(conns[0])
-		conns = conns[1:]
+		handshake.IncomingProtoHandshake(ctx, out, defaultProtoChecker)
 	}()
-	_, err := fx.AcquireDrpcConn(ctx)
+	defer out.Close()
+	fx.mc.EXPECT().Open(gomock.Any()).Return(in, nil)
+	dc, err := fx.AcquireDrpcConn(ctx)
 	require.NoError(t, err)
+	assert.NotEmpty(t, dc)
+	defer dc.Close()
+
+	assert.Len(t, fx.active, 1)
+	assert.Len(t, fx.inactive, 0)
+
+	fx.ReleaseDrpcConn(dc)
+
+	assert.Len(t, fx.active, 0)
+	assert.Len(t, fx.inactive, 1)
+
+	dc, err = fx.AcquireDrpcConn(ctx)
+	require.NoError(t, err)
+	assert.NotEmpty(t, dc)
+	assert.Len(t, fx.active, 1)
+	assert.Len(t, fx.inactive, 0)
 }
 
 func TestPeerAccept(t *testing.T) {
@@ -119,26 +63,6 @@ func TestPeerAccept(t *testing.T) {
 	assert.NoError(t, <-outHandshakeCh)
 }
 
-func TestPeer_DrpcConn_AcceptThrottling(t *testing.T) {
-	fx := newFixture(t, "p1")
-	defer fx.finish()
-
-	var conCount = fx.limiter.startThreshold + 3
-	for i := 0; i < conCount; i++ {
-		in, out := net.Pipe()
-		defer out.Close()
-
-		var outHandshakeCh = make(chan error)
-		go func() {
-			outHandshakeCh <- handshake.OutgoingProtoHandshake(ctx, out, handshakeproto.ProtoType_DRPC)
-		}()
-		fx.acceptCh <- acceptedConn{conn: in}
-		cn := <-fx.testCtrl.serveConn
-		assert.Equal(t, in, cn)
-		assert.NoError(t, <-outHandshakeCh)
-	}
-}
-
 func TestPeer_TryClose(t *testing.T) {
 	t.Run("not close in first minute", func(t *testing.T) {
 		fx := newFixture(t, "p1")

net/pool/poolservice.go

@@ -49,8 +49,8 @@ func (p *poolService) Init(a *app.App) (err error) {
 			return p.dialer.Dial(ctx, id)
 		},
 		ocache.WithLogger(log.Sugar()),
-		ocache.WithGCPeriod(time.Minute/2),
-		ocache.WithTTL(time.Minute),
+		ocache.WithGCPeriod(time.Minute),
+		ocache.WithTTL(time.Minute*5),
 		ocache.WithPrometheus(p.metricReg, "netpool", "outgoing"),
 	)
 	p.pool.incoming = ocache.New(
@@ -58,8 +58,8 @@ func (p *poolService) Init(a *app.App) (err error) {
 			return nil, ocache.ErrNotExists
 		},
 		ocache.WithLogger(log.Sugar()),
-		ocache.WithGCPeriod(time.Minute/2),
-		ocache.WithTTL(time.Minute),
+		ocache.WithGCPeriod(time.Minute),
+		ocache.WithTTL(time.Minute*5),
 		ocache.WithPrometheus(p.metricReg, "netpool", "incoming"),
 	)
 	return nil

net/rpc/rpctest/multiconntest/multiconntest.go

@@ -1,29 +0,0 @@
-package multiconntest
-
-import (
-	"context"
-	"github.com/anyproto/any-sync/net/connutil"
-	"github.com/anyproto/any-sync/net/transport"
-	yamux2 "github.com/anyproto/any-sync/net/transport/yamux"
-	"github.com/hashicorp/yamux"
-	"net"
-)
-
-func MultiConnPair(peerServCtx, peerClientCtx context.Context) (serv, client transport.MultiConn) {
-	sc, cc := net.Pipe()
-	var servConn = make(chan transport.MultiConn, 1)
-	go func() {
-		sess, err := yamux.Server(sc, yamux.DefaultConfig())
-		if err != nil {
-			panic(err)
-		}
-		servConn <- yamux2.NewMultiConn(peerServCtx, connutil.NewLastUsageConn(sc), "", sess)
-	}()
-	sess, err := yamux.Client(cc, yamux.DefaultConfig())
-	if err != nil {
-		panic(err)
-	}
-	client = yamux2.NewMultiConn(peerClientCtx, connutil.NewLastUsageConn(cc), "", sess)
-	serv = <-servConn
-	return
-}

net/rpc/rpctest/peer.go

@@ -2,11 +2,29 @@ package rpctest
 
 import (
 	"context"
+	"github.com/anyproto/any-sync/net/connutil"
 	"github.com/anyproto/any-sync/net/peer"
-	"github.com/anyproto/any-sync/net/rpc/rpctest/multiconntest"
 	"github.com/anyproto/any-sync/net/transport"
+	yamux2 "github.com/anyproto/any-sync/net/transport/yamux"
+	"github.com/hashicorp/yamux"
+	"net"
 )
 
 func MultiConnPair(peerIdServ, peerIdClient string) (serv, client transport.MultiConn) {
-	return multiconntest.MultiConnPair(peer.CtxWithPeerId(context.Background(), peerIdServ), peer.CtxWithPeerId(context.Background(), peerIdClient))
+	sc, cc := net.Pipe()
+	var servConn = make(chan transport.MultiConn, 1)
+	go func() {
+		sess, err := yamux.Server(sc, yamux.DefaultConfig())
+		if err != nil {
+			panic(err)
+		}
+		servConn <- yamux2.NewMultiConn(peer.CtxWithPeerId(context.Background(), peerIdServ), connutil.NewLastUsageConn(sc), "", sess)
+	}()
+	sess, err := yamux.Client(cc, yamux.DefaultConfig())
+	if err != nil {
+		panic(err)
+	}
+	client = yamux2.NewMultiConn(peer.CtxWithPeerId(context.Background(), peerIdClient), connutil.NewLastUsageConn(cc), "", sess)
+	serv = <-servConn
+	return
 }

net/transport/yamux/conn.go

@@ -6,7 +6,6 @@ import (
 	"github.com/anyproto/any-sync/net/peer"
 	"github.com/anyproto/any-sync/net/transport"
 	"github.com/hashicorp/yamux"
-	"io"
 	"net"
 	"time"
 )
@@ -49,7 +48,7 @@ func (y *yamuxConn) Addr() string {
 
 func (y *yamuxConn) Accept() (conn net.Conn, err error) {
 	if conn, err = y.Session.Accept(); err != nil {
-		if err == yamux.ErrSessionShutdown || err == io.EOF {
+		if err == yamux.ErrSessionShutdown {
 			err = transport.ErrConnClosed
 		}
 		return

net/transport/yamux/yamux_test.go

@@ -30,12 +30,8 @@ func TestYamuxTransport_Dial(t *testing.T) {
 
 	mcC, err := fxC.Dial(ctx, fxS.addr)
 	require.NoError(t, err)
-	var mcS transport.MultiConn
-	select {
-	case mcS = <-fxS.accepter.mcs:
-	case <-time.After(time.Second * 5):
-		require.True(t, false, "timeout")
-	}
+	require.Len(t, fxS.accepter.mcs, 1)
+	mcS := <-fxS.accepter.mcs
 
 	var (
 		sData     string
@@ -73,11 +69,11 @@ func TestYamuxTransport_Dial(t *testing.T) {
 // no deadline - 69100 rps
 // common write deadline - 66700 rps
 // subconn write deadline - 67100 rps
-func TestWriteBenchReuse(t *testing.T) {
+func TestWriteBench(t *testing.T) {
 	t.Skip()
 	var (
 		numSubConn = 10
-		numWrites  = 10000
+		numWrites  = 100000
 	)
 
 	fxS := newFixture(t)
@@ -128,63 +124,6 @@ func TestWriteBenchReuse(t *testing.T) {
 	t.Logf("%.2f req per sec", float64(numWrites*numSubConn)/dur.Seconds())
 }
 
-func TestWriteBenchNew(t *testing.T) {
-	t.Skip()
-	var (
-		numSubConn = 10
-		numWrites  = 10000
-	)
-
-	fxS := newFixture(t)
-	defer fxS.finish(t)
-	fxC := newFixture(t)
-	defer fxC.finish(t)
-
-	mcC, err := fxC.Dial(ctx, fxS.addr)
-	require.NoError(t, err)
-	mcS := <-fxS.accepter.mcs
-
-	go func() {
-		for i := 0; i < numSubConn; i++ {
-			require.NoError(t, err)
-			go func() {
-				var b = make([]byte, 1024)
-				for {
-					conn, _ := mcS.Accept()
-					n, _ := conn.Read(b)
-					if n > 0 {
-						conn.Write(b[:n])
-					} else {
-						_ = conn.Close()
-						break
-					}
-					conn.Close()
-				}
-			}()
-		}
-	}()
-
-	var wg sync.WaitGroup
-	wg.Add(numSubConn)
-	st := time.Now()
-	for i := 0; i < numSubConn; i++ {
-		go func() {
-			defer wg.Done()
-			for j := 0; j < numWrites; j++ {
-				sc, err := mcC.Open(ctx)
-				require.NoError(t, err)
-				var b = []byte("some data some data some data some data some data some data some data some data some data")
-				sc.Write(b)
-				sc.Read(b)
-				sc.Close()
-			}
-		}()
-	}
-	wg.Wait()
-	dur := time.Since(st)
-	t.Logf("%.2f req per sec", float64(numWrites*numSubConn)/dur.Seconds())
-}
-
 type fixture struct {
 	*yamuxTransport
 	a            *app.App