remove multiconn.LastUsage + fix peer gc

peer: gc active sub conn

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/config.go

@@ -5,21 +5,3 @@ import "errors"
 var (
 	ErrUnableToConnect = errors.New("unable to connect")
 )
-
-type ConfigGetter interface {
-	GetNet() Config
-}
-
-type Config struct {
-	Server ServerConfig `yaml:"server"`
-	Stream StreamConfig `yaml:"stream"`
-}
-
-type ServerConfig struct {
-	ListenAddrs []string `yaml:"listenAddrs"`
-}
-
-type StreamConfig struct {
-	TimeoutMilliseconds int `yaml:"timeoutMilliseconds"`
-	MaxMsgSizeMb        int `yaml:"maxMsgSizeMb"`
-}

net/peer/peer.go

@@ -6,6 +6,7 @@ import (
 	"github.com/anyproto/any-sync/app/logger"
 	"github.com/anyproto/any-sync/app/ocache"
 	"github.com/anyproto/any-sync/net/connutil"
+	"github.com/anyproto/any-sync/net/rpc"
 	"github.com/anyproto/any-sync/net/secureservice/handshake"
 	"github.com/anyproto/any-sync/net/secureservice/handshake/handshakeproto"
 	"github.com/anyproto/any-sync/net/transport"
@@ -14,6 +15,9 @@ import (
 	"net"
 	"storj.io/drpc"
 	"storj.io/drpc/drpcconn"
+	"storj.io/drpc/drpcmanager"
+	"storj.io/drpc/drpcstream"
+	"storj.io/drpc/drpcwire"
 	"sync"
 	"time"
 )
@@ -22,6 +26,7 @@ var log = logger.NewNamed("common.net.peer")
 
 type connCtrl interface {
 	ServeConn(ctx context.Context, conn net.Conn) (err error)
+	DrpcConfig() rpc.Config
 }
 
 func NewPeer(mc transport.MultiConn, ctrl connCtrl) (p Peer, err error) {
@@ -30,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
@@ -69,6 +75,8 @@ type peer struct {
 
 	mu sync.Mutex
 
+	created time.Time
+
 	transport.MultiConn
 }
 
@@ -90,6 +98,12 @@ func (p *peer) AcquireDrpcConn(ctx context.Context) (drpc.Conn, error) {
 	idx := len(p.inactive) - 1
 	res := p.inactive[idx]
 	p.inactive = p.inactive[:idx]
+	select {
+	case <-res.Closed():
+		p.mu.Unlock()
+		return p.AcquireDrpcConn(ctx)
+	default:
+	}
 	p.active[res] = struct{}{}
 	p.mu.Unlock()
 	return res, nil
@@ -127,8 +141,14 @@ func (p *peer) openDrpcConn(ctx context.Context) (dconn *subConn, err error) {
 		return nil, err
 	}
 	tconn := connutil.NewLastUsageConn(conn)
+	bufSize := p.ctrl.DrpcConfig().Stream.MaxMsgSizeMb * (1 << 20)
 	return &subConn{
-		Conn:          drpcconn.New(tconn),
+		Conn: drpcconn.NewWithOptions(conn, drpcconn.Options{
+			Manager: drpcmanager.Options{
+				Reader: drpcwire.ReaderOptions{MaximumBufferSize: bufSize},
+				Stream: drpcstream.Options{MaximumBufferSize: bufSize},
+			},
+		}),
 		LastUsageConn: tconn,
 	}, nil
 }
@@ -173,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)
@@ -211,9 +231,17 @@ func (p *peer) gc(ttl time.Duration) {
 		select {
 		case <-act.Closed():
 			delete(p.active, act)
+			continue
 		default:
 		}
+		if act.LastUsage().Before(minLastUsage) {
+			log.Warn("close active connection because no activity", zap.String("peerId", p.id), zap.String("addr", p.Addr()))
+			_ = act.Close()
+			delete(p.active, act)
+			continue
+		}
 	}
+	return len(p.active) + len(p.inactive)
 }
 
 func (p *peer) Close() (err error) {

net/peer/peer_test.go

@@ -2,6 +2,7 @@ package peer
 
 import (
 	"context"
+	"github.com/anyproto/any-sync/net/rpc"
 	"github.com/anyproto/any-sync/net/secureservice/handshake"
 	"github.com/anyproto/any-sync/net/secureservice/handshake/handshakeproto"
 	"github.com/anyproto/any-sync/net/transport/mock_transport"
@@ -63,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)
@@ -75,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)
@@ -84,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()
@@ -104,6 +101,7 @@ func TestPeer_TryClose(t *testing.T) {
 		}()
 		defer out2.Close()
 		fx.mc.EXPECT().Open(gomock.Any()).Return(in2, nil)
+		fx.mc.EXPECT().Addr().Return("")
 		dc2, err := fx.AcquireDrpcConn(ctx)
 		require.NoError(t, err)
 		_ = dc2.Close()
@@ -117,6 +115,18 @@ func TestPeer_TryClose(t *testing.T) {
 		fx.mc.EXPECT().Open(gomock.Any()).Return(in3, nil)
 		dc3, err := fx.AcquireDrpcConn(ctx)
 		require.NoError(t, err)
+
+		// make another active, should be removed by ttl
+		in4, out4 := net.Pipe()
+		go func() {
+			handshake.IncomingProtoHandshake(ctx, out4, defaultProtoChecker)
+		}()
+		defer out4.Close()
+		fx.mc.EXPECT().Open(gomock.Any()).Return(in4, nil)
+		dc4, err := fx.AcquireDrpcConn(ctx)
+		require.NoError(t, err)
+		defer dc4.Close()
+
 		fx.ReleaseDrpcConn(dc3)
 		_ = dc3.Close()
 		fx.ReleaseDrpcConn(dc)
@@ -176,6 +186,10 @@ type testCtrl struct {
 	closeCh   chan struct{}
 }
 
+func (t *testCtrl) DrpcConfig() rpc.Config {
+	return rpc.Config{Stream: rpc.StreamConfig{MaxMsgSizeMb: 10}}
+}
+
 func (t *testCtrl) ServeConn(ctx context.Context, conn net.Conn) (err error) {
 	t.serveConn <- conn
 	<-t.closeCh

net/rpc/debugserver/config.go

@@ -0,0 +1,9 @@
+package debugserver
+
+type configGetter interface {
+	GetDebugServer() Config
+}
+
+type Config struct {
+	ListenAddr string `yaml:"listenAddr"`
+}

net/rpc/debugserver/debugserver.go

@@ -0,0 +1,70 @@
+package debugserver
+
+import (
+	"context"
+	"github.com/anyproto/any-sync/app"
+	"github.com/anyproto/any-sync/net/rpc"
+	"net"
+	"storj.io/drpc"
+	"storj.io/drpc/drpcmanager"
+	"storj.io/drpc/drpcmux"
+	"storj.io/drpc/drpcserver"
+	"storj.io/drpc/drpcstream"
+	"storj.io/drpc/drpcwire"
+)
+
+const CName = "net.rpc.debugserver"
+
+func New() DebugServer {
+	return &debugServer{}
+}
+
+type DebugServer interface {
+	app.ComponentRunnable
+	drpc.Mux
+}
+
+type debugServer struct {
+	drpcServer *drpcserver.Server
+	*drpcmux.Mux
+	drpcConf     rpc.Config
+	config       Config
+	runCtx       context.Context
+	runCtxCancel context.CancelFunc
+}
+
+func (d *debugServer) Init(a *app.App) (err error) {
+	d.drpcConf = a.MustComponent("config").(rpc.ConfigGetter).GetDrpc()
+	d.config = a.MustComponent("config").(configGetter).GetDebugServer()
+	d.Mux = drpcmux.New()
+	bufSize := d.drpcConf.Stream.MaxMsgSizeMb * (1 << 20)
+	d.drpcServer = drpcserver.NewWithOptions(d, drpcserver.Options{Manager: drpcmanager.Options{
+		Reader: drpcwire.ReaderOptions{MaximumBufferSize: bufSize},
+		Stream: drpcstream.Options{MaximumBufferSize: bufSize},
+	}})
+	return nil
+}
+
+func (d *debugServer) Name() (name string) {
+	return CName
+}
+
+func (d *debugServer) Run(ctx context.Context) (err error) {
+	if d.config.ListenAddr == "" {
+		return
+	}
+	lis, err := net.Listen("tcp", d.config.ListenAddr)
+	if err != nil {
+		return
+	}
+	d.runCtx, d.runCtxCancel = context.WithCancel(context.Background())
+	go d.drpcServer.Serve(d.runCtx, lis)
+	return
+}
+
+func (d *debugServer) Close(ctx context.Context) (err error) {
+	if d.runCtx != nil {
+		d.runCtxCancel()
+	}
+	return nil
+}

net/rpc/drpcconfig.go

@@ -0,0 +1,13 @@
+package rpc
+
+type ConfigGetter interface {
+	GetDrpc() Config
+}
+
+type Config struct {
+	Stream StreamConfig `yaml:"stream"`
+}
+
+type StreamConfig struct {
+	MaxMsgSizeMb int `yaml:"maxMsgSizeMb"`
+}

net/rpc/rpctest/pool.go

@@ -0,0 +1,76 @@
+package rpctest
+
+import (
+	"context"
+	"github.com/anyproto/any-sync/app"
+	"github.com/anyproto/any-sync/net"
+	"github.com/anyproto/any-sync/net/peer"
+	"github.com/anyproto/any-sync/net/pool"
+	"sync"
+)
+
+func NewTestPool() *TestPool {
+	return &TestPool{peers: map[string]peer.Peer{}}
+}
+
+type TestPool struct {
+	peers map[string]peer.Peer
+	mu    sync.Mutex
+	ts    *TestServer
+}
+
+func (t *TestPool) Init(a *app.App) (err error) {
+	return nil
+}
+
+func (t *TestPool) Name() (name string) {
+	return pool.CName
+}
+
+func (t *TestPool) WithServer(ts *TestServer) *TestPool {
+	t.mu.Lock()
+	defer t.mu.Unlock()
+	t.ts = ts
+	return t
+}
+
+func (t *TestPool) Run(ctx context.Context) (err error) {
+	return nil
+}
+
+func (t *TestPool) Close(ctx context.Context) (err error) {
+	return nil
+}
+
+func (t *TestPool) Get(ctx context.Context, id string) (peer.Peer, error) {
+	t.mu.Lock()
+	defer t.mu.Unlock()
+	if p, ok := t.peers[id]; ok {
+		return p, nil
+	}
+	if t.ts == nil {
+		return nil, net.ErrUnableToConnect
+	}
+	return t.ts.Dial(id)
+}
+
+func (t *TestPool) GetOneOf(ctx context.Context, peerIds []string) (peer.Peer, error) {
+	t.mu.Lock()
+	defer t.mu.Unlock()
+	for _, peerId := range peerIds {
+		if p, ok := t.peers[peerId]; ok {
+			return p, nil
+		}
+	}
+	if t.ts == nil || len(peerIds) == 0 {
+		return nil, net.ErrUnableToConnect
+	}
+	return t.ts.Dial(peerIds[0])
+}
+
+func (t *TestPool) AddPeer(ctx context.Context, p peer.Peer) (err error) {
+	t.mu.Lock()
+	defer t.mu.Unlock()
+	t.peers[p.Id()] = p
+	return nil
+}

net/rpc/rpctest/server.go

@@ -3,12 +3,25 @@ package rpctest
 import (
 	"context"
 	"github.com/anyproto/any-sync/app"
+	"github.com/anyproto/any-sync/net/peer"
+	"github.com/anyproto/any-sync/net/rpc"
 	"github.com/anyproto/any-sync/net/rpc/server"
 	"net"
 	"storj.io/drpc/drpcmux"
 	"storj.io/drpc/drpcserver"
 )
 
+type mockCtrl struct {
+}
+
+func (m mockCtrl) ServeConn(ctx context.Context, conn net.Conn) (err error) {
+	return nil
+}
+
+func (m mockCtrl) DrpcConfig() rpc.Config {
+	return rpc.Config{}
+}
+
 func NewTestServer() *TestServer {
 	ts := &TestServer{
 		Mux: drpcmux.New(),
@@ -22,22 +35,37 @@ type TestServer struct {
 	*drpcserver.Server
 }
 
-func (ts *TestServer) Init(a *app.App) (err error) {
+func (s *TestServer) Init(a *app.App) (err error) {
 	return nil
 }
 
-func (ts *TestServer) Name() (name string) {
+func (s *TestServer) Name() (name string) {
 	return server.CName
 }
 
-func (ts *TestServer) Run(ctx context.Context) (err error) {
+func (s *TestServer) Run(ctx context.Context) (err error) {
 	return nil
 }
 
-func (ts *TestServer) Close(ctx context.Context) (err error) {
+func (s *TestServer) Close(ctx context.Context) (err error) {
 	return nil
 }
 
 func (s *TestServer) ServeConn(ctx context.Context, conn net.Conn) (err error) {
 	return s.Server.ServeOne(ctx, conn)
 }
+
+func (s *TestServer) DrpcConfig() rpc.Config {
+	return rpc.Config{Stream: rpc.StreamConfig{MaxMsgSizeMb: 10}}
+}
+
+func (s *TestServer) Dial(peerId string) (clientPeer peer.Peer, err error) {
+	mcS, mcC := MultiConnPair(peerId+"server", peerId)
+	// NewPeer runs the accept loop
+	_, err = peer.NewPeer(mcS, s)
+	if err != nil {
+		return
+	}
+	// and we ourselves don't call server methods on accept
+	return peer.NewPeer(mcC, mockCtrl{})
+}

net/rpc/server/drpcserver.go

@@ -5,13 +5,14 @@ import (
 	"github.com/anyproto/any-sync/app"
 	"github.com/anyproto/any-sync/app/logger"
 	"github.com/anyproto/any-sync/metric"
-	anyNet "github.com/anyproto/any-sync/net"
+	"github.com/anyproto/any-sync/net/rpc"
 	"go.uber.org/zap"
 	"net"
 	"storj.io/drpc"
 	"storj.io/drpc/drpcmanager"
 	"storj.io/drpc/drpcmux"
 	"storj.io/drpc/drpcserver"
+	"storj.io/drpc/drpcstream"
 	"storj.io/drpc/drpcwire"
 )
 
@@ -25,6 +26,7 @@ func New() DRPCServer {
 
 type DRPCServer interface {
 	ServeConn(ctx context.Context, conn net.Conn) (err error)
+	DrpcConfig() rpc.Config
 	app.Component
 	drpc.Mux
 }
@@ -32,7 +34,7 @@ type DRPCServer interface {
 type drpcServer struct {
 	drpcServer *drpcserver.Server
 	*drpcmux.Mux
-	config anyNet.Config
+	config rpc.Config
 	metric metric.Metric
 }
 
@@ -43,7 +45,7 @@ func (s *drpcServer) Name() (name string) {
 }
 
 func (s *drpcServer) Init(a *app.App) (err error) {
-	s.config = a.MustComponent("config").(anyNet.ConfigGetter).GetNet()
+	s.config = a.MustComponent("config").(rpc.ConfigGetter).GetDrpc()
 	s.metric, _ = a.Component(metric.CName).(metric.Metric)
 	s.Mux = drpcmux.New()
 
@@ -52,8 +54,10 @@ func (s *drpcServer) Init(a *app.App) (err error) {
 	if s.metric != nil {
 		handler = s.metric.WrapDRPCHandler(s)
 	}
+	bufSize := s.config.Stream.MaxMsgSizeMb * (1 << 20)
 	s.drpcServer = drpcserver.NewWithOptions(handler, drpcserver.Options{Manager: drpcmanager.Options{
-		Reader: drpcwire.ReaderOptions{MaximumBufferSize: s.config.Stream.MaxMsgSizeMb * (1 << 20)},
+		Reader: drpcwire.ReaderOptions{MaximumBufferSize: bufSize},
+		Stream: drpcstream.Options{MaximumBufferSize: bufSize},
 	}})
 	return
 }
@@ -63,3 +67,7 @@ func (s *drpcServer) ServeConn(ctx context.Context, conn net.Conn) (err error) {
 	l.Debug("drpc serve peer")
 	return s.drpcServer.ServeOne(ctx, conn)
 }
+
+func (s *drpcServer) DrpcConfig() rpc.Config {
+	return s.config
+}

net/secureservice/secureservice.go

@@ -3,6 +3,9 @@ package secureservice
 import (
 	"context"
 	"crypto/tls"
+	"io"
+	"net"
+
 	commonaccount "github.com/anyproto/any-sync/accountservice"
 	"github.com/anyproto/any-sync/app"
 	"github.com/anyproto/any-sync/app/logger"
@@ -13,14 +16,18 @@ import (
 	"github.com/libp2p/go-libp2p/core/crypto"
 	libp2ptls "github.com/libp2p/go-libp2p/p2p/security/tls"
 	"go.uber.org/zap"
-	"io"
-	"net"
 )
 
 const CName = "common.net.secure"
 
 var log = logger.NewNamed(CName)
 
+const (
+	// ProtoVersion 0 - first any-sync version with raw tcp connections
+	// ProtoVersion 1 - version with yamux over tcp and quic
+	ProtoVersion = 1
+)
+
 func New() SecureService {
 	return &secureService{}
 }
@@ -46,6 +53,9 @@ type secureService struct {
 }
 
 func (s *secureService) Init(a *app.App) (err error) {
+	if s.protoVersion == 0 {
+		s.protoVersion = ProtoVersion
+	}
 	account := a.MustComponent(commonaccount.CName).(commonaccount.Service)
 	peerKey, err := account.Account().PeerKey.Raw()
 	if err != nil {

net/secureservice/secureservice_test.go

@@ -57,7 +57,7 @@ func TestHandshake(t *testing.T) {
 
 func TestHandshakeIncompatibleVersion(t *testing.T) {
 	nc := testnodeconf.GenNodeConfig(2)
-	fxS := newFixture(t, nc, nc.GetAccountService(0), 0)
+	fxS := newFixture(t, nc, nc.GetAccountService(0), 1)
 	defer fxS.Finish(t)
 	sc, cc := net.Pipe()
 
@@ -72,7 +72,7 @@ func TestHandshakeIncompatibleVersion(t *testing.T) {
 		ar.ctx, ar.err = fxS.SecureInbound(ctx, sc)
 		resCh <- ar
 	}()
-	fxC := newFixture(t, nc, nc.GetAccountService(1), 1)
+	fxC := newFixture(t, nc, nc.GetAccountService(1), 2)
 	defer fxC.Finish(t)
 	_, err := fxC.SecureOutbound(ctx, cc)
 	require.Equal(t, handshake.ErrIncompatibleVersion, err)

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/yamux.go

@@ -11,6 +11,7 @@ import (
 	"github.com/hashicorp/yamux"
 	"go.uber.org/zap"
 	"net"
+	"sync"
 	"time"
 )
 
@@ -25,6 +26,7 @@ func New() Yamux {
 // Yamux implements transport.Transport with tcp+yamux
 type Yamux interface {
 	transport.Transport
+	AddListener(lis net.Listener)
 	app.ComponentRunnable
 }
 
@@ -37,15 +39,31 @@ type yamuxTransport struct {
 	listCtx       context.Context
 	listCtxCancel context.CancelFunc
 	yamuxConf     *yamux.Config
+	mu            sync.Mutex
 }
 
 func (y *yamuxTransport) Init(a *app.App) (err error) {
 	y.secure = a.MustComponent(secureservice.CName).(secureservice.SecureService)
 	y.conf = a.MustComponent("config").(configGetter).GetYamux()
+	if y.conf.DialTimeoutSec <= 0 {
+		y.conf.DialTimeoutSec = 10
+	}
+	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())
 	return
 }
 
@@ -57,6 +75,8 @@ func (y *yamuxTransport) Run(ctx context.Context) (err error) {
 	if y.accepter == nil {
 		return fmt.Errorf("can't run service without accepter")
 	}
+	y.mu.Lock()
+	defer y.mu.Unlock()
 	for _, listAddr := range y.conf.ListenAddrs {
 		list, err := net.Listen("tcp", listAddr)
 		if err != nil {
@@ -64,7 +84,6 @@ func (y *yamuxTransport) Run(ctx context.Context) (err error) {
 		}
 		y.listeners = append(y.listeners, list)
 	}
-	y.listCtx, y.listCtxCancel = context.WithCancel(context.Background())
 	for _, list := range y.listeners {
 		go y.acceptLoop(y.listCtx, list)
 	}
@@ -75,6 +94,13 @@ func (y *yamuxTransport) SetAccepter(accepter transport.Accepter) {
 	y.accepter = accepter
 }
 
+func (y *yamuxTransport) AddListener(lis net.Listener) {
+	y.mu.Lock()
+	defer y.mu.Unlock()
+	y.listeners = append(y.listeners, lis)
+	go y.acceptLoop(y.listCtx, lis)
+}
+
 func (y *yamuxTransport) Dial(ctx context.Context, addr string) (mc transport.MultiConn, err error) {
 	dialTimeout := time.Duration(y.conf.DialTimeoutSec) * time.Second
 	conn, err := net.DialTimeout("tcp", addr, dialTimeout)
@@ -88,7 +114,7 @@ func (y *yamuxTransport) Dial(ctx context.Context, addr string) (mc transport.Mu
 		_ = conn.Close()
 		return nil, err
 	}
-	luc := connutil.NewLastUsageConn(conn)
+	luc := connutil.NewLastUsageConn(connutil.NewTimeout(conn, time.Duration(y.conf.WriteTimeoutSec)*time.Second))
 	sess, err := yamux.Client(luc, y.yamuxConf)
 	if err != nil {
 		return
@@ -134,7 +160,7 @@ func (y *yamuxTransport) accept(conn net.Conn) {
 		log.Warn("incoming connection handshake error", zap.Error(err))
 		return
 	}
-	luc := connutil.NewLastUsageConn(conn)
+	luc := connutil.NewLastUsageConn(connutil.NewTimeout(conn, time.Duration(y.conf.WriteTimeoutSec)*time.Second))
 	sess, err := yamux.Server(luc, y.yamuxConf)
 	if err != nil {
 		log.Warn("incoming connection yamux session error", zap.Error(err))