Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

feat: node warmup time for pull/pushsync protocols #2050

Merged
merged 1 commit into from
Jun 11, 2021
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
3 changes: 3 additions & 0 deletions cmd/bee/cmd/cmd.go
Original file line number Diff line number Diff line change
Expand Up @@ -11,6 +11,7 @@ import (
"os"
"path/filepath"
"strings"
"time"

"github.com/ethersphere/bee/pkg/logging"
"github.com/ethersphere/bee/pkg/swarm"
Expand Down Expand Up @@ -64,6 +65,7 @@ const (
optionNameFullNode = "full-node"
optionNamePostageContractAddress = "postage-stamp-address"
optionNameBlockTime = "block-time"
optionWarmUpTime = "warmup-time"
)

func init() {
Expand Down Expand Up @@ -236,6 +238,7 @@ func (c *command) setAllFlags(cmd *cobra.Command) {
cmd.Flags().String(optionNameTransactionHash, "", "proof-of-identity transaction hash")
cmd.Flags().Uint64(optionNameBlockTime, 15, "chain block time")
cmd.Flags().String(optionNameSwapDeploymentGasPrice, "", "gas price in wei to use for deployment and funding")
cmd.Flags().Duration(optionWarmUpTime, time.Minute*10, "time to warmup the node before pull/push protocols can be kicked off.")
}

func newLogger(cmd *cobra.Command, verbosity string) (logging.Logger, error) {
Expand Down
1 change: 1 addition & 0 deletions cmd/bee/cmd/start.go
Original file line number Diff line number Diff line change
Expand Up @@ -154,6 +154,7 @@ Welcome to the Swarm.... Bzzz Bzzzz Bzzzz
PostageContractAddress: c.config.GetString(optionNamePostageContractAddress),
BlockTime: c.config.GetUint64(optionNameBlockTime),
DeployGasPrice: c.config.GetString(optionNameSwapDeploymentGasPrice),
WarmupTime: c.config.GetDuration(optionWarmUpTime),
})
if err != nil {
return err
Expand Down
13 changes: 10 additions & 3 deletions pkg/node/node.go
Original file line number Diff line number Diff line change
Expand Up @@ -146,6 +146,7 @@ type Options struct {
PriceOracleAddress string
BlockTime uint64
DeployGasPrice string
WarmupTime time.Duration
}

const (
Expand Down Expand Up @@ -173,6 +174,12 @@ func NewBee(addr string, swarmAddress swarm.Address, publicKey ecdsa.PublicKey,
}
}()

// light nodes have zero warmup time for pull/pushsync protocols
warmupTime := o.WarmupTime
if !o.FullNodeMode {
warmupTime = 0
}

b = &Bee{
p2pCancel: p2pCancel,
errorLogWriter: logger.WriterLevel(logrus.ErrorLevel),
Expand Down Expand Up @@ -559,7 +566,7 @@ func NewBee(addr string, swarmAddress swarm.Address, publicKey ecdsa.PublicKey,

pinningService := pinning.NewService(storer, stateStore, traversalService)

pushSyncProtocol := pushsync.New(swarmAddress, p2ps, storer, kad, tagService, o.FullNodeMode, pssService.TryUnwrap, validStamp, logger, acc, pricer, signer, tracer)
pushSyncProtocol := pushsync.New(swarmAddress, p2ps, storer, kad, tagService, o.FullNodeMode, pssService.TryUnwrap, validStamp, logger, acc, pricer, signer, tracer, warmupTime)

// set the pushSyncer in the PSS
pssService.SetPushSyncer(pushSyncProtocol)
Expand All @@ -570,7 +577,7 @@ func NewBee(addr string, swarmAddress swarm.Address, publicKey ecdsa.PublicKey,
b.recoveryHandleCleanup = pssService.Register(recovery.Topic, chunkRepairHandler)
}

pusherService := pusher.New(networkID, storer, kad, pushSyncProtocol, tagService, logger, tracer)
pusherService := pusher.New(networkID, storer, kad, pushSyncProtocol, tagService, logger, tracer, warmupTime)
b.pusherCloser = pusherService

pullStorage := pullstorage.New(storer)
Expand All @@ -580,7 +587,7 @@ func NewBee(addr string, swarmAddress swarm.Address, publicKey ecdsa.PublicKey,

var pullerService *puller.Puller
if o.FullNodeMode {
pullerService := puller.New(stateStore, kad, pullSyncProtocol, logger, puller.Options{})
pullerService := puller.New(stateStore, kad, pullSyncProtocol, logger, puller.Options{}, warmupTime)
b.pullerCloser = pullerService
}

Expand Down
16 changes: 13 additions & 3 deletions pkg/puller/puller.go
Original file line number Diff line number Diff line change
Expand Up @@ -50,7 +50,7 @@ type Puller struct {
bins uint8 // how many bins do we support
}

func New(stateStore storage.StateStorer, topology topology.Driver, pullSync pullsync.Interface, logger logging.Logger, o Options) *Puller {
func New(stateStore storage.StateStorer, topology topology.Driver, pullSync pullsync.Interface, logger logging.Logger, o Options, warmupTime time.Duration) *Puller {
var (
bins uint8 = swarm.MaxBins
)
Expand All @@ -77,7 +77,7 @@ func New(stateStore storage.StateStorer, topology topology.Driver, pullSync pull
p.syncPeers[i] = make(map[string]*syncPeer)
}
p.wg.Add(1)
go p.manage()
go p.manage(warmupTime)
return p
}

Expand All @@ -86,7 +86,7 @@ type peer struct {
po uint8
}

func (p *Puller) manage() {
func (p *Puller) manage(warmupTime time.Duration) {
defer p.wg.Done()
c, unsubscribe := p.topology.SubscribePeersChange()
defer unsubscribe()
Expand All @@ -96,6 +96,16 @@ func (p *Puller) manage() {
<-p.quit
cancel()
}()

// wait for warmup duration to complete
select {
case <-time.After(warmupTime):
case <-p.quit:
return
}

p.logger.Info("puller: warmup period complete, worker starting.")

for {
select {
case <-c:
Expand Down
2 changes: 1 addition & 1 deletion pkg/puller/puller_test.go
Original file line number Diff line number Diff line change
Expand Up @@ -597,7 +597,7 @@ func newPuller(ops opts) (*puller.Puller, storage.StateStorer, *mockk.Mock, *moc
o := puller.Options{
Bins: ops.bins,
}
return puller.New(s, kad, ps, logger, o), s, kad, ps
return puller.New(s, kad, ps, logger, o, 0), s, kad, ps
}

type c struct {
Expand Down
16 changes: 13 additions & 3 deletions pkg/pusher/pusher.go
Original file line number Diff line number Diff line change
Expand Up @@ -49,7 +49,7 @@ var (

var ErrInvalidAddress = errors.New("invalid address")

func New(networkID uint64, storer storage.Storer, peerSuggester topology.ClosestPeerer, pushSyncer pushsync.PushSyncer, tagger *tags.Tags, logger logging.Logger, tracer *tracing.Tracer) *Service {
func New(networkID uint64, storer storage.Storer, peerSuggester topology.ClosestPeerer, pushSyncer pushsync.PushSyncer, tagger *tags.Tags, logger logging.Logger, tracer *tracing.Tracer, warmupTime time.Duration) *Service {
service := &Service{
networkID: networkID,
storer: storer,
Expand All @@ -61,13 +61,13 @@ func New(networkID uint64, storer storage.Storer, peerSuggester topology.Closest
quit: make(chan struct{}),
chunksWorkerQuitC: make(chan struct{}),
}
go service.chunksWorker()
go service.chunksWorker(warmupTime)
return service
}

// chunksWorker is a loop that keeps looking for chunks that are locally uploaded ( by monitoring pushIndex )
// and pushes them to the closest peer and get a receipt.
func (s *Service) chunksWorker() {
func (s *Service) chunksWorker(warmupTime time.Duration) {
var (
chunks <-chan swarm.Chunk
unsubscribe func()
Expand All @@ -81,13 +81,23 @@ func (s *Service) chunksWorker() {
span opentracing.Span
logger *logrus.Entry
)

defer timer.Stop()
defer close(s.chunksWorkerQuitC)
go func() {
<-s.quit
cancel()
}()

// wait for warmup duration to complete
select {
case <-time.After(warmupTime):
case <-s.quit:
return
}

s.logger.Info("pusher: warmup period complete, worker starting.")

LOOP:
for {
select {
Expand Down
2 changes: 1 addition & 1 deletion pkg/pusher/pusher_test.go
Original file line number Diff line number Diff line change
Expand Up @@ -379,7 +379,7 @@ func createPusher(t *testing.T, addr swarm.Address, pushSyncService pushsync.Pus
}
peerSuggester := mock.NewTopologyDriver(mockOpts...)

pusherService := pusher.New(1, pusherStorer, peerSuggester, pushSyncService, mtags, logger, nil)
pusherService := pusher.New(1, pusherStorer, peerSuggester, pushSyncService, mtags, logger, nil, 0)
return mtags, pusherService, pusherStorer
}

Expand Down
11 changes: 10 additions & 1 deletion pkg/pushsync/pushsync.go
Original file line number Diff line number Diff line change
Expand Up @@ -45,6 +45,7 @@ const (
var (
ErrOutOfDepthReplication = errors.New("replication outside of the neighborhood")
ErrNoPush = errors.New("could not push chunk")
ErrWarmup = errors.New("node warmup time not complete")
)

type PushSyncer interface {
Expand Down Expand Up @@ -72,13 +73,14 @@ type PushSync struct {
signer crypto.Signer
isFullNode bool
failedRequests *failedRequestCache
warmupPeriod time.Time
}

var defaultTTL = 20 * time.Second // request time to live
var timeToWaitForPushsyncToNeighbor = 3 * time.Second // time to wait to get a receipt for a chunk
var nPeersToPushsync = 3 // number of peers to replicate to as receipt is sent upstream

func New(address swarm.Address, streamer p2p.StreamerDisconnecter, storer storage.Putter, topology topology.Driver, tagger *tags.Tags, isFullNode bool, unwrap func(swarm.Chunk), validStamp func(swarm.Chunk, []byte) (swarm.Chunk, error), logger logging.Logger, accounting accounting.Interface, pricer pricer.Interface, signer crypto.Signer, tracer *tracing.Tracer) *PushSync {
func New(address swarm.Address, streamer p2p.StreamerDisconnecter, storer storage.Putter, topology topology.Driver, tagger *tags.Tags, isFullNode bool, unwrap func(swarm.Chunk), validStamp func(swarm.Chunk, []byte) (swarm.Chunk, error), logger logging.Logger, accounting accounting.Interface, pricer pricer.Interface, signer crypto.Signer, tracer *tracing.Tracer, warmupTime time.Duration) *PushSync {
ps := &PushSync{
address: address,
streamer: streamer,
Expand All @@ -95,6 +97,7 @@ func New(address swarm.Address, streamer p2p.StreamerDisconnecter, storer storag
validStamp: validStamp,
signer: signer,
failedRequests: newFailedRequestCache(),
warmupPeriod: time.Now().Add(warmupTime),
}
return ps
}
Expand Down Expand Up @@ -197,6 +200,12 @@ func (ps *PushSync) handler(ctx context.Context, p p2p.Peer, stream p2p.Stream)
receipt, err := ps.pushToClosest(ctx, chunk, false)
if err != nil {
if errors.Is(err, topology.ErrWantSelf) {

if time.Now().Before(ps.warmupPeriod) {
err = ErrWarmup
return
}

if !storedChunk {
_, err = ps.storer.Put(ctx, storage.ModePutSync, chunk)
if err != nil {
Expand Down
2 changes: 1 addition & 1 deletion pkg/pushsync/pushsync_test.go
Original file line number Diff line number Diff line change
Expand Up @@ -804,7 +804,7 @@ func createPushSyncNodeWithAccounting(t *testing.T, addr swarm.Address, prices p
return ch.WithStamp(postage.NewStamp(nil, nil)), nil
}

return pushsync.New(addr, recorderDisconnecter, storer, mockTopology, mtag, true, unwrap, validStamp, logger, acct, mockPricer, signer, nil), storer, mtag
return pushsync.New(addr, recorderDisconnecter, storer, mockTopology, mtag, true, unwrap, validStamp, logger, acct, mockPricer, signer, nil, 0), storer, mtag
}

func waitOnRecordAndTest(t *testing.T, peer swarm.Address, recorder *streamtest.Recorder, add swarm.Address, data []byte) {
Expand Down