remove multiconn.LastUsage + fix peer gc

Makefile

@@ -25,6 +25,7 @@ deps:
 	go mod download
 	go build -o deps storj.io/drpc/cmd/protoc-gen-go-drpc
 	go build -o deps github.com/gogo/protobuf/protoc-gen-gogofaster
+	go build -o deps github.com/golang/mock/mockgen
 
 test:
 	go test ./... --cover

net/connutil/timeout.go

@@ -21,13 +21,7 @@ func NewTimeout(conn net.Conn, timeout time.Duration) *TimeoutConn {
 	return &TimeoutConn{conn, timeout}
 }
 
-func (c *TimeoutConn) Write(p []byte) (n int, retErr error) {
-	log.Debug("start write", zap.Int("n", len(p)))
-	defer func() {
-		if retErr != nil {
-			log.Debug("conn write error", zap.Int("n", n), zap.Error(retErr))
-		}
-	}()
+func (c *TimeoutConn) Write(p []byte) (n int, err error) {
 	for {
 		if c.timeout != 0 {
 			if e := c.Conn.SetWriteDeadline(time.Now().Add(c.timeout)); e != nil {
@@ -53,7 +47,6 @@ func (c *TimeoutConn) Write(p []byte) (n int, retErr error) {
 			}
 			log.Debug("connection timed out", zap.String("remoteAddr", c.RemoteAddr().String()))
 		}
-		log.Debug("conn write", zap.Int("n", n))
 		return n, err
 	}
 }

net/peer/peer.go

@@ -35,6 +35,7 @@ func NewPeer(mc transport.MultiConn, ctrl connCtrl) (p Peer, err error) {
 		active:    map[*subConn]struct{}{},
 		MultiConn: mc,
 		ctrl:      ctrl,
+		created:   time.Now(),
 	}
 	if pr.id, err = CtxPeerId(ctx); err != nil {
 		return
@@ -74,6 +75,8 @@ type peer struct {
 
 	mu sync.Mutex
 
+	created time.Time
+
 	transport.MultiConn
 }
 
@@ -190,14 +193,14 @@ func (p *peer) serve(conn net.Conn) (err error) {
 }
 
 func (p *peer) TryClose(objectTTL time.Duration) (res bool, err error) {
-	p.gc(objectTTL)
-	if time.Now().Sub(p.LastUsage()) < objectTTL {
-		return false, nil
+	aliveCount := p.gc(objectTTL)
+	if aliveCount == 0 && p.created.Add(time.Minute).Before(time.Now()) {
+		return true, p.Close()
 	}
-	return true, p.Close()
+	return false, nil
 }
 
-func (p *peer) gc(ttl time.Duration) {
+func (p *peer) gc(ttl time.Duration) (aliveCount int) {
 	p.mu.Lock()
 	defer p.mu.Unlock()
 	minLastUsage := time.Now().Add(-ttl)
@@ -238,6 +241,7 @@ func (p *peer) gc(ttl time.Duration) {
 			continue
 		}
 	}
+	return len(p.active) + len(p.inactive)
 }
 
 func (p *peer) Close() (err error) {

net/peer/peer_test.go

@@ -64,11 +64,9 @@ func TestPeerAccept(t *testing.T) {
 }
 
 func TestPeer_TryClose(t *testing.T) {
-	t.Run("ttl", func(t *testing.T) {
+	t.Run("not close in first minute", func(t *testing.T) {
 		fx := newFixture(t, "p1")
 		defer fx.finish()
-		lu := time.Now()
-		fx.mc.EXPECT().LastUsage().Return(lu)
 		res, err := fx.TryClose(time.Second)
 		require.NoError(t, err)
 		assert.False(t, res)
@@ -76,8 +74,7 @@ func TestPeer_TryClose(t *testing.T) {
 	t.Run("close", func(t *testing.T) {
 		fx := newFixture(t, "p1")
 		defer fx.finish()
-		lu := time.Now().Add(-time.Second * 2)
-		fx.mc.EXPECT().LastUsage().Return(lu)
+		fx.peer.created = fx.peer.created.Add(-time.Minute * 2)
 		res, err := fx.TryClose(time.Second)
 		require.NoError(t, err)
 		assert.True(t, res)
@@ -85,8 +82,7 @@ func TestPeer_TryClose(t *testing.T) {
 	t.Run("gc", func(t *testing.T) {
 		fx := newFixture(t, "p1")
 		defer fx.finish()
-		now := time.Now()
-		fx.mc.EXPECT().LastUsage().Return(now.Add(time.Millisecond * 100))
+		//now := time.Now()
 
 		// make one inactive
 		in, out := net.Pipe()

net/transport/mock_transport/mock_transport.go

@@ -8,7 +8,6 @@ import (
 	context "context"
 	net "net"
 	reflect "reflect"
-	time "time"
 
 	transport "github.com/anyproto/any-sync/net/transport"
 	gomock "github.com/golang/mock/gomock"
@@ -158,20 +157,6 @@ func (mr *MockMultiConnMockRecorder) IsClosed() *gomock.Call {
 	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "IsClosed", reflect.TypeOf((*MockMultiConn)(nil).IsClosed))
 }
 
-// LastUsage mocks base method.
-func (m *MockMultiConn) LastUsage() time.Time {
-	m.ctrl.T.Helper()
-	ret := m.ctrl.Call(m, "LastUsage")
-	ret0, _ := ret[0].(time.Time)
-	return ret0
-}
-
-// LastUsage indicates an expected call of LastUsage.
-func (mr *MockMultiConnMockRecorder) LastUsage() *gomock.Call {
-	mr.mock.ctrl.T.Helper()
-	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "LastUsage", reflect.TypeOf((*MockMultiConn)(nil).LastUsage))
-}
-
 // Open mocks base method.
 func (m *MockMultiConn) Open(arg0 context.Context) (net.Conn, error) {
 	m.ctrl.T.Helper()

net/transport/transport.go

@@ -5,7 +5,6 @@ import (
 	"context"
 	"errors"
 	"net"
-	"time"
 )
 
 var (
@@ -29,8 +28,6 @@ type MultiConn interface {
 	Accept() (conn net.Conn, err error)
 	// Open opens new sub connection
 	Open(ctx context.Context) (conn net.Conn, err error)
-	// LastUsage returns the time of the last connection activity
-	LastUsage() time.Time
 	// Addr returns remote peer address
 	Addr() string
 	// IsClosed returns true when connection is closed

net/transport/yamux/config.go

@@ -5,7 +5,8 @@ type configGetter interface {
 }
 
 type Config struct {
-	ListenAddrs     []string `yaml:"listenAddrs"`
-	WriteTimeoutSec int      `yaml:"writeTimeoutSec"`
-	DialTimeoutSec  int      `yaml:"dialTimeoutSec"`
+	ListenAddrs        []string `yaml:"listenAddrs"`
+	WriteTimeoutSec    int      `yaml:"writeTimeoutSec"`
+	DialTimeoutSec     int      `yaml:"dialTimeoutSec"`
+	KeepAlivePeriodSec int      `yaml:"keepAlivePeriodSec"`
 }

net/transport/yamux/conn.go

@@ -29,7 +29,6 @@ func (y *yamuxConn) Open(ctx context.Context) (conn net.Conn, err error) {
 	if conn, err = y.Session.Open(); err != nil {
 		return
 	}
-	//conn = connutil.NewTimeout(conn, time.Second*10)
 	return
 }
 
@@ -52,6 +51,5 @@ func (y *yamuxConn) Accept() (conn net.Conn, err error) {
 		}
 		return
 	}
-	//conn = connutil.NewTimeout(conn, time.Second*10)
 	return
 }

net/transport/yamux/yamux.go

@@ -51,8 +51,16 @@ func (y *yamuxTransport) Init(a *app.App) (err error) {
 	if y.conf.WriteTimeoutSec <= 0 {
 		y.conf.WriteTimeoutSec = 10
 	}
+
 	y.yamuxConf = yamux.DefaultConfig()
-	y.yamuxConf.EnableKeepAlive = false
+	if y.conf.KeepAlivePeriodSec < 0 {
+		y.yamuxConf.EnableKeepAlive = false
+	} else {
+		y.yamuxConf.EnableKeepAlive = true
+		if y.conf.KeepAlivePeriodSec != 0 {
+			y.yamuxConf.KeepAliveInterval = time.Duration(y.conf.KeepAlivePeriodSec) * time.Second
+		}
+	}
 	y.yamuxConf.StreamOpenTimeout = time.Duration(y.conf.DialTimeoutSec) * time.Second
 	y.yamuxConf.ConnectionWriteTimeout = time.Duration(y.conf.WriteTimeoutSec) * time.Second
 	y.listCtx, y.listCtxCancel = context.WithCancel(context.Background())

testutil/nettest/net_test.go

@@ -1,114 +0,0 @@
-package nettest
-
-import (
-	"context"
-	"github.com/anyproto/any-sync/app"
-	"github.com/anyproto/any-sync/app/logger"
-	"github.com/anyproto/any-sync/coordinator/coordinatorclient"
-	"github.com/anyproto/any-sync/coordinator/coordinatorproto"
-	"github.com/anyproto/any-sync/coordinator/nodeconfsource"
-	"github.com/anyproto/any-sync/net/peerservice"
-	"github.com/anyproto/any-sync/net/pool"
-	"github.com/anyproto/any-sync/net/rpc"
-	"github.com/anyproto/any-sync/net/rpc/server"
-	"github.com/anyproto/any-sync/net/secureservice"
-	"github.com/anyproto/any-sync/net/transport/yamux"
-	"github.com/anyproto/any-sync/nodeconf"
-	"github.com/anyproto/any-sync/nodeconf/nodeconfstore"
-	"github.com/anyproto/any-sync/testutil/accounttest"
-	"github.com/stretchr/testify/require"
-	"go.uber.org/zap"
-	"net/http"
-	_ "net/http/pprof"
-	"testing"
-	"time"
-)
-
-var ctx = context.Background()
-
-func init() {
-	go http.ListenAndServe(":6061", nil)
-}
-
-func TestNet(t *testing.T) {
-	l := logger.NewNamed("test")
-	pl := pool.New()
-	a := new(app.App)
-	a.Register(&accounttest.AccountTestService{}).
-		Register(pl).
-		Register(peerservice.New()).
-		Register(yamux.New()).
-		Register(nodeconf.New()).
-		Register(nodeconfstore.New()).
-		Register(nodeconfsource.New()).
-		Register(server.New()).
-		Register(secureservice.New()).
-		Register(coordinatorclient.New()).
-		Register(&testConfig{})
-	require.NoError(t, a.Start(ctx))
-	defer a.Close(ctx)
-
-	p, err := pl.Get(ctx, "12D3KooWB4hmEo7YAdWzAaFpjyk4npkcwrPm2kRigsWu3MP9Xdmg")
-	require.NoError(t, err)
-
-	for i := 0; i < 1000000; i++ {
-		dc, err := p.AcquireDrpcConn(ctx)
-		require.NoError(t, err)
-		cl := coordinatorproto.NewDRPCCoordinatorClient(dc)
-		time.Sleep(time.Second)
-		res, err := cl.NetworkConfiguration(ctx, &coordinatorproto.NetworkConfigurationRequest{})
-		if err != nil {
-			l.Warn("req error", zap.Error(err))
-		} else {
-
-			l.Info("req success", zap.String("nid", res.NetworkId))
-		}
-		p.ReleaseDrpcConn(dc)
-	}
-
-}
-
-type testConfig struct {
-}
-
-func (t *testConfig) GetDrpc() rpc.Config {
-	return rpc.Config{
-		Stream: rpc.StreamConfig{
-			MaxMsgSizeMb: 1,
-		},
-	}
-}
-
-func (t *testConfig) GetNodeConfStorePath() string {
-	return "/tmp"
-}
-
-func (t *testConfig) GetNodeConf() nodeconf.Configuration {
-	return nodeconf.Configuration{
-		Id:        "",
-		NetworkId: "",
-		Nodes: []nodeconf.Node{
-			{
-				PeerId:    "12D3KooWB4hmEo7YAdWzAaFpjyk4npkcwrPm2kRigsWu3MP9Xdmg",
-				Addresses: []string{"ec2-3-65-13-48.eu-central-1.compute.amazonaws.com:443"},
-				Types:     []nodeconf.NodeType{nodeconf.NodeTypeTree},
-			},
-		},
-		CreationTime: time.Now(),
-	}
-}
-
-func (t *testConfig) GetYamux() yamux.Config {
-	return yamux.Config{
-		WriteTimeoutSec: 10,
-		DialTimeoutSec:  10,
-	}
-}
-
-func (t *testConfig) Init(a *app.App) (err error) {
-	return nil
-}
-
-func (t *testConfig) Name() (name string) {
-	return "config"
-}