Skip to content

Commit

Permalink
Make it async
Browse files Browse the repository at this point in the history
  • Loading branch information
samsondav committed Oct 8, 2020
1 parent 099c43e commit 98222a4
Show file tree
Hide file tree
Showing 2 changed files with 21 additions and 12 deletions.
31 changes: 20 additions & 11 deletions dht.go
Original file line number Diff line number Diff line change
Expand Up @@ -138,7 +138,8 @@ type IpfsDHT struct {
// networks).
enableProviders, enableValues bool

fixLowPeersChan chan struct{}
disableFixLowPeers bool
fixLowPeersChan chan struct{}

addPeerToRTChan chan addPeerRTReq
refreshFinishedCh chan struct{}
Expand Down Expand Up @@ -186,6 +187,7 @@ func New(ctx context.Context, h host.Host, options ...Option) (*IpfsDHT, error)
dht.maxRecordAge = cfg.maxRecordAge
dht.enableProviders = cfg.enableProviders
dht.enableValues = cfg.enableValues
dht.disableFixLowPeers = cfg.disableFixLowPeers

dht.Validator = cfg.validator

Expand Down Expand Up @@ -216,21 +218,12 @@ func New(ctx context.Context, h host.Host, options ...Option) (*IpfsDHT, error)
// handle providers
dht.proc.AddChild(dht.ProviderManager.Process())

dht.fixLowPeers()

if err := dht.rtRefreshManager.Start(); err != nil {
return nil, err
}
dht.proc.Go(dht.populatePeers)

// go-routine to make sure we ALWAYS have RT peer addresses in the peerstore
// since RT membership is decoupled from connectivity
go dht.persistRTPeersInPeerStore()

// listens to the fix low peers chan and tries to fix the Routing Table
if !cfg.disableFixLowPeers {
dht.proc.Go(dht.fixLowPeersRoutine)
}

dht.proc.Go(dht.rtPeerLoop)

return dht, nil
Expand Down Expand Up @@ -419,6 +412,22 @@ func (dht *IpfsDHT) Mode() ModeOpt {
return dht.auto
}

func (dht *IpfsDHT) populatePeers(_ goprocess.Process) {
if !dht.disableFixLowPeers {
dht.fixLowPeers(dht.ctx)
}

if err := dht.rtRefreshManager.Start(); err != nil {
logger.Error(err)
}

// listens to the fix low peers chan and tries to fix the Routing Table
if !dht.disableFixLowPeers {
dht.proc.Go(dht.fixLowPeersRoutine)
}

}

// fixLowPeersRouting manages simultaneous requests to fixLowPeers
func (dht *IpfsDHT) fixLowPeersRoutine(proc goprocess.Process) {
ticker := time.NewTicker(periodicBootstrapInterval)
Expand Down
2 changes: 1 addition & 1 deletion dht_bootstrap.go
Original file line number Diff line number Diff line change
Expand Up @@ -58,7 +58,7 @@ func GetDefaultBootstrapPeerAddrInfos() []peer.AddrInfo {
// Bootstrap tells the DHT to get into a bootstrapped state satisfying the
// IpfsRouter interface.
func (dht *IpfsDHT) Bootstrap(ctx context.Context) error {
dht.fixLowPeers(ctx)
dht.fixRTIfNeeded()
dht.rtRefreshManager.RefreshNoWait()
return nil
}
Expand Down

0 comments on commit 98222a4

Please sign in to comment.