cerc-io/lotus
11
0
Fork 0
Code Issues Pull Requests Packages Projects Releases 27 Wiki Activity

feat(deps): update to go-data-transfer v2, still wip

Browse Source
This commit is contained in:
hannahhoward
2023-03-07 11:44:29 -08:00
parent 64b9b532c0
commit 16dad6342c
25 changed files with 239 additions and 279 deletions
Download Patch File Download Diff File Expand all files Collapse all files
node/impl/client/client.go
+3 -3
View File
@@ -44,7 +44,7 @@ import (
"github.com/filecoin-project/go-address"
cborutil "github.com/filecoin-project/go-cbor-util"
"github.com/filecoin-project/go-commp-utils/writer"
datatransfer "github.com/filecoin-project/go-data-transfer"
datatransfer "github.com/filecoin-project/go-data-transfer/v2"
"github.com/filecoin-project/go-fil-markets/discovery"
rm "github.com/filecoin-project/go-fil-markets/retrievalmarket"
"github.com/filecoin-project/go-fil-markets/storagemarket"
@@ -1302,8 +1302,8 @@ func (a *API) ClientQueryAsk(ctx context.Context, p peer.ID, miner address.Addre
return nil, err
}
for _, s := range ps {
if strings.HasPrefix(s, dealProtoPrefix) {
res.DealProtocols = append(res.DealProtocols, s)
if strings.HasPrefix(string(s), dealProtoPrefix) {
res.DealProtocols = append(res.DealProtocols, string(s))
}
}
sort.Strings(res.DealProtocols)
node/impl/net/net.go
+6 -2
View File
@@ -91,8 +91,12 @@ func (a *NetAPI) NetPeerInfo(_ context.Context, p peer.ID) (*api.ExtendedPeerInf
protocols, err := a.Host.Peerstore().GetProtocols(p)
if err == nil {
sort.Strings(protocols)
info.Protocols = protocols
protocolStrings := make([]string, 0, len(protocols))
for _, protocol := range protocols {
protocolStrings = append(protocolStrings, string(protocol))
}
sort.Strings(protocolStrings)
info.Protocols = protocolStrings
}
if cm := a.Host.ConnManager().GetTagInfo(p); cm != nil {
node/impl/storminer.go
+2 -2
View File
@@ -25,8 +25,8 @@ import (
"github.com/filecoin-project/dagstore/shard"
"github.com/filecoin-project/go-address"
"github.com/filecoin-project/go-bitfield"
datatransfer "github.com/filecoin-project/go-data-transfer"
gst "github.com/filecoin-project/go-data-transfer/transport/graphsync"
datatransfer "github.com/filecoin-project/go-data-transfer/v2"
gst "github.com/filecoin-project/go-data-transfer/v2/transport/graphsync"
"github.com/filecoin-project/go-fil-markets/piecestore"
"github.com/filecoin-project/go-fil-markets/retrievalmarket"
"github.com/filecoin-project/go-fil-markets/storagemarket"
node/modules/client.go
+4 -14
View File
@@ -13,10 +13,10 @@ import (
"go.uber.org/fx"
"golang.org/x/xerrors"
"github.com/filecoin-project/go-data-transfer/channelmonitor"
dtimpl "github.com/filecoin-project/go-data-transfer/impl"
dtnet "github.com/filecoin-project/go-data-transfer/network"
dtgstransport "github.com/filecoin-project/go-data-transfer/transport/graphsync"
"github.com/filecoin-project/go-data-transfer/v2/channelmonitor"
dtimpl "github.com/filecoin-project/go-data-transfer/v2/impl"
dtnet "github.com/filecoin-project/go-data-transfer/v2/network"
dtgstransport "github.com/filecoin-project/go-data-transfer/v2/transport/graphsync"
"github.com/filecoin-project/go-fil-markets/discovery"
discoveryimpl "github.com/filecoin-project/go-fil-markets/discovery/impl"
"github.com/filecoin-project/go-fil-markets/retrievalmarket"
@@ -24,7 +24,6 @@ import (
rmnet "github.com/filecoin-project/go-fil-markets/retrievalmarket/network"
"github.com/filecoin-project/go-fil-markets/storagemarket"
storageimpl "github.com/filecoin-project/go-fil-markets/storagemarket/impl"
"github.com/filecoin-project/go-fil-markets/storagemarket/impl/requestvalidation"
smnet "github.com/filecoin-project/go-fil-markets/storagemarket/network"
"github.com/filecoin-project/go-state-types/abi"
@@ -95,15 +94,6 @@ func ClientBlockstore() dtypes.ClientBlockstore {
return blockstore.WrapIDStore(blockstore.FromDatastore(datastore.NewMapDatastore()))
}
// RegisterClientValidator is an initialization hook that registers the client
// request validator with the data transfer module as the validator for
// StorageDataTransferVoucher types
func RegisterClientValidator(crv dtypes.ClientRequestValidator, dtm dtypes.ClientDataTransfer) {
if err := dtm.RegisterVoucherType(&requestvalidation.StorageDataTransferVoucher{}, (*requestvalidation.UnifiedRequestValidator)(crv)); err != nil {
panic(err)
}
}
// NewClientGraphsyncDataTransfer returns a data transfer manager that just
// uses the clients's Client DAG service for transfers
func NewClientGraphsyncDataTransfer(lc fx.Lifecycle, h host.Host, gs dtypes.Graphsync, ds dtypes.MetadataDS, r repo.LockedRepo) (dtypes.ClientDataTransfer, error) {
node/modules/dtypes/storage.go
+2 -2
View File
@@ -6,8 +6,8 @@ import (
"github.com/ipfs/go-graphsync"
exchange "github.com/ipfs/go-ipfs-exchange-interface"
datatransfer "github.com/filecoin-project/go-data-transfer"
dtnet "github.com/filecoin-project/go-data-transfer/network"
datatransfer "github.com/filecoin-project/go-data-transfer/v2"
dtnet "github.com/filecoin-project/go-data-transfer/v2/network"
"github.com/filecoin-project/go-fil-markets/piecestore"
"github.com/filecoin-project/go-fil-markets/storagemarket/impl/requestvalidation"
"github.com/filecoin-project/go-statestore"
node/modules/lp2p/rcmgr.go
+30 -26
View File
@@ -14,10 +14,8 @@ import (
"github.com/libp2p/go-libp2p/core/peer"
"github.com/libp2p/go-libp2p/core/protocol"
rcmgr "github.com/libp2p/go-libp2p/p2p/host/resource-manager"
"github.com/libp2p/go-libp2p/p2p/host/resource-manager/obs"
rcmgrObs "github.com/libp2p/go-libp2p/p2p/host/resource-manager/obs"
"go.opencensus.io/stats"
"go.opencensus.io/stats/view"
"go.opencensus.io/tag"
"go.uber.org/fx"
@@ -32,7 +30,7 @@ func ResourceManager(connMgrHi uint) func(lc fx.Lifecycle, repo repo.LockedRepo)
if (isFullNode && envvar == "0") || // only set NullResourceManager if envvar is explicitly "0"
(!isFullNode && envvar != "1") { // set NullResourceManager *unless* envvar is explicitly "1"
log.Info("libp2p resource manager is disabled")
return network.NullResourceManager, nil
return &network.NullResourceManager{}, nil
}
log.Info("libp2p resource manager is enabled")
@@ -52,37 +50,41 @@ func ResourceManager(connMgrHi uint) func(lc fx.Lifecycle, repo repo.LockedRepo)
// For every extra 1GB of memory we have available, increase our limit by 1GiB
defaultLimits.SystemLimitIncrease.Memory = 1 << 30
defaultLimitConfig := defaultLimits.AutoScale()
if defaultLimitConfig.System.Memory > 4<<30 {
changes := rcmgr.PartialLimitConfig{}
if defaultLimitConfig.ToPartialLimitConfig().System.Memory > 4<<30 {
// Cap our memory limit
defaultLimitConfig.System.Memory = 4 << 30
changes.System.Memory = 4 << 30
}
maxconns := int(connMgrHi)
if 2*maxconns > defaultLimitConfig.System.ConnsInbound {
if rcmgr.LimitVal(2*maxconns) > defaultLimitConfig.ToPartialLimitConfig().System.ConnsInbound {
// adjust conns to 2x to allow for two conns per peer (TCP+QUIC)
defaultLimitConfig.System.ConnsInbound = logScale(2 * maxconns)
defaultLimitConfig.System.ConnsOutbound = logScale(2 * maxconns)
defaultLimitConfig.System.Conns = logScale(4 * maxconns)
changes.System.ConnsInbound = rcmgr.LimitVal(logScale(2 * maxconns))
changes.System.ConnsOutbound = rcmgr.LimitVal(logScale(2 * maxconns))
changes.System.Conns = rcmgr.LimitVal(logScale(4 * maxconns))
defaultLimitConfig.System.StreamsInbound = logScale(16 * maxconns)
defaultLimitConfig.System.StreamsOutbound = logScale(64 * maxconns)
defaultLimitConfig.System.Streams = logScale(64 * maxconns)
changes.System.StreamsInbound = rcmgr.LimitVal(logScale(16 * maxconns))
changes.System.StreamsOutbound = rcmgr.LimitVal(logScale(64 * maxconns))
changes.System.Streams = rcmgr.LimitVal(logScale(64 * maxconns))
if 2*maxconns > defaultLimitConfig.System.FD {
defaultLimitConfig.System.FD = logScale(2 * maxconns)
if rcmgr.LimitVal(2*maxconns) > defaultLimitConfig.ToPartialLimitConfig().System.FD {
changes.System.FD = rcmgr.LimitVal(logScale(2 * maxconns))
}
defaultLimitConfig.ServiceDefault.StreamsInbound = logScale(8 * maxconns)
defaultLimitConfig.ServiceDefault.StreamsOutbound = logScale(32 * maxconns)
defaultLimitConfig.ServiceDefault.Streams = logScale(32 * maxconns)
changes.ServiceDefault.StreamsInbound = rcmgr.LimitVal(logScale(8 * maxconns))
changes.ServiceDefault.StreamsOutbound = rcmgr.LimitVal(logScale(32 * maxconns))
changes.ServiceDefault.Streams = rcmgr.LimitVal(logScale(32 * maxconns))
defaultLimitConfig.ProtocolDefault.StreamsInbound = logScale(8 * maxconns)
defaultLimitConfig.ProtocolDefault.StreamsOutbound = logScale(32 * maxconns)
defaultLimitConfig.ProtocolDefault.Streams = logScale(32 * maxconns)
changes.ProtocolDefault.StreamsInbound = rcmgr.LimitVal(logScale(8 * maxconns))
changes.ProtocolDefault.StreamsOutbound = rcmgr.LimitVal(logScale(32 * maxconns))
changes.ProtocolDefault.Streams = rcmgr.LimitVal(logScale(32 * maxconns))
log.Info("adjusted default resource manager limits")
}
changedLimitConfig := changes.Build(defaultLimitConfig)
// initialize
var limiter rcmgr.Limiter
var opts []rcmgr.Option
@@ -95,13 +97,13 @@ func ResourceManager(connMgrHi uint) func(lc fx.Lifecycle, repo repo.LockedRepo)
switch {
case err == nil:
defer limitsIn.Close() //nolint:errcheck
limiter, err = rcmgr.NewLimiterFromJSON(limitsIn, defaultLimitConfig)
limiter, err = rcmgr.NewLimiterFromJSON(limitsIn, changedLimitConfig)
if err != nil {
return nil, fmt.Errorf("error parsing limit file: %w", err)
}
case errors.Is(err, os.ErrNotExist):
limiter = rcmgr.NewFixedLimiter(defaultLimitConfig)
limiter = rcmgr.NewFixedLimiter(changedLimitConfig)
default:
return nil, err
@@ -111,10 +113,12 @@ func ResourceManager(connMgrHi uint) func(lc fx.Lifecycle, repo repo.LockedRepo)
if err != nil {
return nil, fmt.Errorf("error creating resource manager stats reporter: %w", err)
}
err = view.Register(obs.DefaultViews...)
if err != nil {
return nil, fmt.Errorf("error registering rcmgr metrics: %w", err)
}
// TODO: Libp2p converted to native prometheus metrics -- how do we register them
//err = view.Register(obs.DefaultViews...)
//if err != nil {
// return nil, fmt.Errorf("error registering rcmgr metrics: %w", err)
//}
// Metrics
opts = append(opts, rcmgr.WithMetrics(rcmgrMetrics{}), rcmgr.WithTraceReporter(str))
node/modules/services.go
+2 -1
View File
@@ -12,6 +12,7 @@ import (
"github.com/libp2p/go-libp2p/core/event"
"github.com/libp2p/go-libp2p/core/host"
"github.com/libp2p/go-libp2p/core/peer"
"github.com/libp2p/go-libp2p/core/protocol"
"github.com/libp2p/go-libp2p/p2p/host/eventbus"
"go.uber.org/fx"
"golang.org/x/xerrors"
@@ -84,7 +85,7 @@ func RunHello(mctx helpers.MetricsCtx, lc fx.Lifecycle, h host.Host, svc *hello.
return nil
}
func protosContains(protos []string, search string) bool {
func protosContains(protos []protocol.ID, search protocol.ID) bool {
for _, p := range protos {
if p == search {
return true
node/modules/storageminer.go
+3 -3
View File
@@ -25,9 +25,9 @@ import (
"golang.org/x/xerrors"
"github.com/filecoin-project/go-address"
dtimpl "github.com/filecoin-project/go-data-transfer/impl"
dtnet "github.com/filecoin-project/go-data-transfer/network"
dtgstransport "github.com/filecoin-project/go-data-transfer/transport/graphsync"
dtimpl "github.com/filecoin-project/go-data-transfer/v2/impl"
dtnet "github.com/filecoin-project/go-data-transfer/v2/network"
dtgstransport "github.com/filecoin-project/go-data-transfer/v2/transport/graphsync"
piecefilestore "github.com/filecoin-project/go-fil-markets/filestore"
piecestoreimpl "github.com/filecoin-project/go-fil-markets/piecestore/impl"
"github.com/filecoin-project/go-fil-markets/retrievalmarket"