Merge remote-tracking branch 'origin/master' into feat/post-worker
This commit is contained in:
Łukasz Magiera
@@ -15,6 +15,7 @@ import (
|
||||
logging "github.com/ipfs/go-log/v2"
|
||||
ci "github.com/libp2p/go-libp2p-core/crypto"
|
||||
"github.com/libp2p/go-libp2p-core/host"
|
||||
"github.com/libp2p/go-libp2p-core/network"
|
||||
"github.com/libp2p/go-libp2p-core/peer"
|
||||
"github.com/libp2p/go-libp2p-core/peerstore"
|
||||
"github.com/libp2p/go-libp2p-core/routing"
|
||||
@@ -68,6 +69,7 @@ var (
|
||||
BandwidthReporterKey = special{11} // Libp2p option
|
||||
ConnGaterKey = special{12} // libp2p option
|
||||
DAGStoreKey = special{13} // constructor returns multiple values
|
||||
ResourceManagerKey = special{14} // Libp2p option
|
||||
)
|
||||
|
||||
type invoke int
|
||||
@@ -215,6 +217,10 @@ var LibP2P = Options(
|
||||
Override(ConnectionManagerKey, lp2p.ConnectionManager(50, 200, 20*time.Second, nil)),
|
||||
Override(new(*conngater.BasicConnectionGater), lp2p.ConnGater),
|
||||
Override(ConnGaterKey, lp2p.ConnGaterOption),
|
||||
|
||||
// Services (resource management)
|
||||
Override(new(network.ResourceManager), lp2p.ResourceManager),
|
||||
Override(ResourceManagerKey, lp2p.ResourceManagerOption),
|
||||
)
|
||||
|
||||
func IsType(t repo.RepoType) func(s *Settings) bool {
|
||||
|
||||
@@ -25,12 +25,13 @@ import (
|
||||
type NetAPI struct {
|
||||
fx.In
|
||||
|
||||
RawHost lp2p.RawHost
|
||||
Host host.Host
|
||||
Router lp2p.BaseIpfsRouting
|
||||
ConnGater *conngater.BasicConnectionGater
|
||||
Reporter metrics.Reporter
|
||||
Sk *dtypes.ScoreKeeper
|
||||
RawHost lp2p.RawHost
|
||||
Host host.Host
|
||||
Router lp2p.BaseIpfsRouting
|
||||
ConnGater *conngater.BasicConnectionGater
|
||||
ResourceManager network.ResourceManager
|
||||
Reporter metrics.Reporter
|
||||
Sk *dtypes.ScoreKeeper
|
||||
}
|
||||
|
||||
func (a *NetAPI) ID(context.Context) (peer.ID, error) {
|
||||
|
||||
@@ -0,0 +1,271 @@
|
||||
package net
|
||||
|
||||
import (
|
||||
"context"
|
||||
"strings"
|
||||
|
||||
"golang.org/x/xerrors"
|
||||
|
||||
"github.com/libp2p/go-libp2p-core/network"
|
||||
"github.com/libp2p/go-libp2p-core/peer"
|
||||
"github.com/libp2p/go-libp2p-core/protocol"
|
||||
rcmgr "github.com/libp2p/go-libp2p-resource-manager"
|
||||
|
||||
"github.com/filecoin-project/lotus/api"
|
||||
)
|
||||
|
||||
func (a *NetAPI) NetStat(ctx context.Context, scope string) (result api.NetStat, err error) {
|
||||
switch {
|
||||
case scope == "all":
|
||||
rapi, ok := a.ResourceManager.(rcmgr.ResourceManagerState)
|
||||
if !ok {
|
||||
return result, xerrors.Errorf("rexource manager does not support ResourceManagerState API")
|
||||
}
|
||||
|
||||
stat := rapi.Stat()
|
||||
result.System = &stat.System
|
||||
result.Transient = &stat.Transient
|
||||
if len(stat.Services) > 0 {
|
||||
result.Services = stat.Services
|
||||
}
|
||||
if len(stat.Protocols) > 0 {
|
||||
result.Protocols = make(map[string]network.ScopeStat, len(stat.Protocols))
|
||||
for proto, stat := range stat.Protocols {
|
||||
result.Protocols[string(proto)] = stat
|
||||
}
|
||||
}
|
||||
if len(stat.Peers) > 0 {
|
||||
result.Peers = make(map[string]network.ScopeStat, len(stat.Peers))
|
||||
for p, stat := range stat.Peers {
|
||||
result.Peers[p.Pretty()] = stat
|
||||
}
|
||||
}
|
||||
|
||||
return result, nil
|
||||
|
||||
case scope == "system":
|
||||
err = a.ResourceManager.ViewSystem(func(s network.ResourceScope) error {
|
||||
stat := s.Stat()
|
||||
result.System = &stat
|
||||
return nil
|
||||
})
|
||||
return result, err
|
||||
|
||||
case scope == "transient":
|
||||
err = a.ResourceManager.ViewTransient(func(s network.ResourceScope) error {
|
||||
stat := s.Stat()
|
||||
result.Transient = &stat
|
||||
return nil
|
||||
})
|
||||
return result, err
|
||||
|
||||
case strings.HasPrefix(scope, "svc:"):
|
||||
svc := scope[4:]
|
||||
err = a.ResourceManager.ViewService(svc, func(s network.ServiceScope) error {
|
||||
stat := s.Stat()
|
||||
result.Services = map[string]network.ScopeStat{
|
||||
svc: stat,
|
||||
}
|
||||
return nil
|
||||
})
|
||||
return result, err
|
||||
|
||||
case strings.HasPrefix(scope, "proto:"):
|
||||
proto := scope[6:]
|
||||
err = a.ResourceManager.ViewProtocol(protocol.ID(proto), func(s network.ProtocolScope) error {
|
||||
stat := s.Stat()
|
||||
result.Protocols = map[string]network.ScopeStat{
|
||||
proto: stat,
|
||||
}
|
||||
return nil
|
||||
})
|
||||
return result, err
|
||||
|
||||
case strings.HasPrefix(scope, "peer:"):
|
||||
p := scope[5:]
|
||||
pid, err := peer.IDFromString(p)
|
||||
if err != nil {
|
||||
return result, xerrors.Errorf("invalid peer ID: %s: %w", p, err)
|
||||
}
|
||||
err = a.ResourceManager.ViewPeer(pid, func(s network.PeerScope) error {
|
||||
stat := s.Stat()
|
||||
result.Peers = map[string]network.ScopeStat{
|
||||
p: stat,
|
||||
}
|
||||
return nil
|
||||
})
|
||||
return result, err
|
||||
|
||||
default:
|
||||
return result, xerrors.Errorf("invalid scope %s", scope)
|
||||
}
|
||||
}
|
||||
|
||||
func (a *NetAPI) NetLimit(ctx context.Context, scope string) (result api.NetLimit, err error) {
|
||||
getLimit := func(s network.ResourceScope) error {
|
||||
limiter, ok := s.(rcmgr.ResourceScopeLimiter)
|
||||
if !ok {
|
||||
return xerrors.Errorf("resource scope doesn't implement ResourceScopeLimiter interface")
|
||||
}
|
||||
|
||||
limit := limiter.Limit()
|
||||
switch l := limit.(type) {
|
||||
case *rcmgr.StaticLimit:
|
||||
result.Memory = l.Memory
|
||||
result.Streams = l.BaseLimit.Streams
|
||||
result.StreamsInbound = l.BaseLimit.StreamsInbound
|
||||
result.StreamsOutbound = l.BaseLimit.StreamsOutbound
|
||||
result.Conns = l.BaseLimit.Conns
|
||||
result.ConnsInbound = l.BaseLimit.ConnsInbound
|
||||
result.ConnsOutbound = l.BaseLimit.ConnsOutbound
|
||||
result.FD = l.BaseLimit.FD
|
||||
|
||||
case *rcmgr.DynamicLimit:
|
||||
result.Dynamic = true
|
||||
result.MemoryFraction = l.MemoryLimit.MemoryFraction
|
||||
result.MinMemory = l.MemoryLimit.MinMemory
|
||||
result.MaxMemory = l.MemoryLimit.MaxMemory
|
||||
result.Streams = l.BaseLimit.Streams
|
||||
result.StreamsInbound = l.BaseLimit.StreamsInbound
|
||||
result.StreamsOutbound = l.BaseLimit.StreamsOutbound
|
||||
result.Conns = l.BaseLimit.Conns
|
||||
result.ConnsInbound = l.BaseLimit.ConnsInbound
|
||||
result.ConnsOutbound = l.BaseLimit.ConnsOutbound
|
||||
result.FD = l.BaseLimit.FD
|
||||
|
||||
default:
|
||||
return xerrors.Errorf("unknown limit type %T", limit)
|
||||
}
|
||||
|
||||
return nil
|
||||
}
|
||||
|
||||
switch {
|
||||
case scope == "system":
|
||||
err = a.ResourceManager.ViewSystem(func(s network.ResourceScope) error {
|
||||
return getLimit(s)
|
||||
})
|
||||
return result, err
|
||||
|
||||
case scope == "transient":
|
||||
err = a.ResourceManager.ViewTransient(func(s network.ResourceScope) error {
|
||||
return getLimit(s)
|
||||
})
|
||||
return result, err
|
||||
|
||||
case strings.HasPrefix(scope, "svc:"):
|
||||
svc := scope[4:]
|
||||
err = a.ResourceManager.ViewService(svc, func(s network.ServiceScope) error {
|
||||
return getLimit(s)
|
||||
})
|
||||
return result, err
|
||||
|
||||
case strings.HasPrefix(scope, "proto:"):
|
||||
proto := scope[6:]
|
||||
err = a.ResourceManager.ViewProtocol(protocol.ID(proto), func(s network.ProtocolScope) error {
|
||||
return getLimit(s)
|
||||
})
|
||||
return result, err
|
||||
|
||||
case strings.HasPrefix(scope, "peer:"):
|
||||
p := scope[5:]
|
||||
pid, err := peer.IDFromString(p)
|
||||
if err != nil {
|
||||
return result, xerrors.Errorf("invalid peer ID: %s: %w", p, err)
|
||||
}
|
||||
err = a.ResourceManager.ViewPeer(pid, func(s network.PeerScope) error {
|
||||
return getLimit(s)
|
||||
})
|
||||
return result, err
|
||||
|
||||
default:
|
||||
return result, xerrors.Errorf("invalid scope %s", scope)
|
||||
}
|
||||
}
|
||||
|
||||
func (a *NetAPI) NetSetLimit(ctx context.Context, scope string, limit api.NetLimit) error {
|
||||
setLimit := func(s network.ResourceScope) error {
|
||||
limiter, ok := s.(rcmgr.ResourceScopeLimiter)
|
||||
if !ok {
|
||||
return xerrors.Errorf("resource scope doesn't implement ResourceScopeLimiter interface")
|
||||
}
|
||||
|
||||
var newLimit rcmgr.Limit
|
||||
if limit.Dynamic {
|
||||
newLimit = &rcmgr.DynamicLimit{
|
||||
MemoryLimit: rcmgr.MemoryLimit{
|
||||
MemoryFraction: limit.MemoryFraction,
|
||||
MinMemory: limit.MinMemory,
|
||||
MaxMemory: limit.MaxMemory,
|
||||
},
|
||||
BaseLimit: rcmgr.BaseLimit{
|
||||
Streams: limit.Streams,
|
||||
StreamsInbound: limit.StreamsInbound,
|
||||
StreamsOutbound: limit.StreamsOutbound,
|
||||
Conns: limit.Conns,
|
||||
ConnsInbound: limit.ConnsInbound,
|
||||
ConnsOutbound: limit.ConnsOutbound,
|
||||
FD: limit.FD,
|
||||
},
|
||||
}
|
||||
} else {
|
||||
newLimit = &rcmgr.StaticLimit{
|
||||
Memory: limit.Memory,
|
||||
BaseLimit: rcmgr.BaseLimit{
|
||||
Streams: limit.Streams,
|
||||
StreamsInbound: limit.StreamsInbound,
|
||||
StreamsOutbound: limit.StreamsOutbound,
|
||||
Conns: limit.Conns,
|
||||
ConnsInbound: limit.ConnsInbound,
|
||||
ConnsOutbound: limit.ConnsOutbound,
|
||||
FD: limit.FD,
|
||||
},
|
||||
}
|
||||
}
|
||||
|
||||
limiter.SetLimit(newLimit)
|
||||
return nil
|
||||
}
|
||||
|
||||
switch {
|
||||
case scope == "system":
|
||||
err := a.ResourceManager.ViewSystem(func(s network.ResourceScope) error {
|
||||
return setLimit(s)
|
||||
})
|
||||
return err
|
||||
|
||||
case scope == "transient":
|
||||
err := a.ResourceManager.ViewTransient(func(s network.ResourceScope) error {
|
||||
return setLimit(s)
|
||||
})
|
||||
return err
|
||||
|
||||
case strings.HasPrefix(scope, "svc:"):
|
||||
svc := scope[4:]
|
||||
err := a.ResourceManager.ViewService(svc, func(s network.ServiceScope) error {
|
||||
return setLimit(s)
|
||||
})
|
||||
return err
|
||||
|
||||
case strings.HasPrefix(scope, "proto:"):
|
||||
proto := scope[6:]
|
||||
err := a.ResourceManager.ViewProtocol(protocol.ID(proto), func(s network.ProtocolScope) error {
|
||||
return setLimit(s)
|
||||
})
|
||||
return err
|
||||
|
||||
case strings.HasPrefix(scope, "peer:"):
|
||||
p := scope[5:]
|
||||
pid, err := peer.IDFromString(p)
|
||||
if err != nil {
|
||||
return xerrors.Errorf("invalid peer ID: %s: %w", p, err)
|
||||
}
|
||||
err = a.ResourceManager.ViewPeer(pid, func(s network.PeerScope) error {
|
||||
return setLimit(s)
|
||||
})
|
||||
return err
|
||||
|
||||
default:
|
||||
return xerrors.Errorf("invalid scope %s", scope)
|
||||
}
|
||||
}
|
||||
@@ -114,11 +114,7 @@ func NewClientGraphsyncDataTransfer(lc fx.Lifecycle, h host.Host, gs dtypes.Grap
|
||||
net := dtnet.NewFromLibp2pHost(h, dtRetryParams)
|
||||
|
||||
dtDs := namespace.Wrap(ds, datastore.NewKey("/datatransfer/client/transfers"))
|
||||
transport := dtgstransport.NewTransport(h.ID(), gs, net)
|
||||
err := os.MkdirAll(filepath.Join(r.Path(), "data-transfer"), 0755) //nolint: gosec
|
||||
if err != nil && !os.IsExist(err) {
|
||||
return nil, err
|
||||
}
|
||||
transport := dtgstransport.NewTransport(h.ID(), gs)
|
||||
|
||||
// data-transfer push / pull channel restart configuration:
|
||||
dtRestartConfig := dtimpl.ChannelRestartConfig(channelmonitor.Config{
|
||||
@@ -138,7 +134,7 @@ func NewClientGraphsyncDataTransfer(lc fx.Lifecycle, h host.Host, gs dtypes.Grap
|
||||
// After trying to restart 3 times, give up and fail the transfer
|
||||
MaxConsecutiveRestarts: 3,
|
||||
})
|
||||
dt, err := dtimpl.NewDataTransfer(dtDs, filepath.Join(r.Path(), "data-transfer"), net, transport, dtRestartConfig)
|
||||
dt, err := dtimpl.NewDataTransfer(dtDs, net, transport, dtRestartConfig)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
@@ -10,10 +10,10 @@ import (
|
||||
|
||||
logging "github.com/ipfs/go-log/v2"
|
||||
"github.com/libp2p/go-libp2p"
|
||||
connmgr "github.com/libp2p/go-libp2p-connmgr"
|
||||
"github.com/libp2p/go-libp2p-core/crypto"
|
||||
"github.com/libp2p/go-libp2p-core/peer"
|
||||
"github.com/libp2p/go-libp2p-core/peerstore"
|
||||
"github.com/libp2p/go-libp2p/p2p/net/connmgr"
|
||||
"go.uber.org/fx"
|
||||
)
|
||||
|
||||
|
||||
@@ -0,0 +1,72 @@
|
||||
package lp2p
|
||||
|
||||
import (
|
||||
"context"
|
||||
"errors"
|
||||
"fmt"
|
||||
"os"
|
||||
"path/filepath"
|
||||
|
||||
"go.uber.org/fx"
|
||||
|
||||
"github.com/libp2p/go-libp2p"
|
||||
"github.com/libp2p/go-libp2p-core/network"
|
||||
rcmgr "github.com/libp2p/go-libp2p-resource-manager"
|
||||
|
||||
"github.com/filecoin-project/lotus/node/repo"
|
||||
)
|
||||
|
||||
func ResourceManager(lc fx.Lifecycle, repo repo.LockedRepo) (network.ResourceManager, error) {
|
||||
var limiter *rcmgr.BasicLimiter
|
||||
var opts []rcmgr.Option
|
||||
|
||||
repoPath := repo.Path()
|
||||
|
||||
// create limiter -- parse $repo/limits.json if exists
|
||||
limitsFile := filepath.Join(repoPath, "limits.json")
|
||||
limitsIn, err := os.Open(limitsFile)
|
||||
switch {
|
||||
case err == nil:
|
||||
defer limitsIn.Close() //nolint:errcheck
|
||||
limiter, err = rcmgr.NewDefaultLimiterFromJSON(limitsIn)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("error parsing limit file: %w", err)
|
||||
}
|
||||
|
||||
case errors.Is(err, os.ErrNotExist):
|
||||
limiter = rcmgr.NewDefaultLimiter()
|
||||
|
||||
default:
|
||||
return nil, err
|
||||
}
|
||||
|
||||
// TODO: also set appropriate default limits for lotus protocols
|
||||
libp2p.SetDefaultServiceLimits(limiter)
|
||||
|
||||
if os.Getenv("LOTUS_DEBUG_RCMGR") != "" {
|
||||
debugPath := filepath.Join(repoPath, "debug")
|
||||
if err := os.MkdirAll(debugPath, 0755); err != nil {
|
||||
return nil, fmt.Errorf("error creating debug directory: %w", err)
|
||||
}
|
||||
traceFile := filepath.Join(debugPath, "rcmgr.json.gz")
|
||||
opts = append(opts, rcmgr.WithTrace(traceFile))
|
||||
}
|
||||
|
||||
mgr, err := rcmgr.NewResourceManager(limiter, opts...)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("error creating resource manager: %w", err)
|
||||
}
|
||||
|
||||
lc.Append(fx.Hook{
|
||||
OnStop: func(_ context.Context) error {
|
||||
return mgr.Close()
|
||||
}})
|
||||
|
||||
return mgr, nil
|
||||
}
|
||||
|
||||
func ResourceManagerOption(mgr network.ResourceManager) Libp2pOpts {
|
||||
return Libp2pOpts{
|
||||
Opts: []libp2p.Option{libp2p.ResourceManager(mgr)},
|
||||
}
|
||||
}
|
||||
@@ -341,19 +341,15 @@ func NewProviderTransferNetwork(h host.Host) dtypes.ProviderTransferNetwork {
|
||||
}
|
||||
|
||||
// NewProviderTransport sets up a data transfer transport over graphsync
|
||||
func NewProviderTransport(h host.Host, gs dtypes.StagingGraphsync, net dtypes.ProviderTransferNetwork) dtypes.ProviderTransport {
|
||||
return dtgstransport.NewTransport(h.ID(), gs, net)
|
||||
func NewProviderTransport(h host.Host, gs dtypes.StagingGraphsync) dtypes.ProviderTransport {
|
||||
return dtgstransport.NewTransport(h.ID(), gs)
|
||||
}
|
||||
|
||||
// NewProviderDataTransfer returns a data transfer manager
|
||||
func NewProviderDataTransfer(lc fx.Lifecycle, net dtypes.ProviderTransferNetwork, transport dtypes.ProviderTransport, ds dtypes.MetadataDS, r repo.LockedRepo) (dtypes.ProviderDataTransfer, error) {
|
||||
dtDs := namespace.Wrap(ds, datastore.NewKey("/datatransfer/provider/transfers"))
|
||||
err := os.MkdirAll(filepath.Join(r.Path(), "data-transfer"), 0755) //nolint: gosec
|
||||
if err != nil && !os.IsExist(err) {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
dt, err := dtimpl.NewDataTransfer(dtDs, filepath.Join(r.Path(), "data-transfer"), net, transport)
|
||||
dt, err := dtimpl.NewDataTransfer(dtDs, net, transport)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user