package blossomsub import ( "context" "math/rand" "time" "github.com/libp2p/go-libp2p/core/discovery" "github.com/libp2p/go-libp2p/core/host" "github.com/libp2p/go-libp2p/core/peer" discimpl "github.com/libp2p/go-libp2p/p2p/discovery/backoff" ) var ( // poll interval // DiscoveryPollInitialDelay is how long the discovery system waits after it first starts before polling DiscoveryPollInitialDelay = 0 * time.Millisecond // DiscoveryPollInterval is approximately how long the discovery system waits in between checks for whether the // more peers are needed for any bitmask DiscoveryPollInterval = 1 * time.Second ) // interval at which to retry advertisements when they fail. const discoveryAdvertiseRetryInterval = 2 * time.Minute type DiscoverOpt func(*discoverOptions) error type discoverOptions struct { connFactory BackoffConnectorFactory opts []discovery.Option } func defaultDiscoverOptions() *discoverOptions { rngSrc := rand.NewSource(rand.Int63()) minBackoff, maxBackoff := time.Second*10, time.Hour cacheSize := 100 dialTimeout := time.Minute * 2 discoverOpts := &discoverOptions{ connFactory: func(host host.Host) (*discimpl.BackoffConnector, error) { backoff := discimpl.NewExponentialBackoff(minBackoff, maxBackoff, discimpl.FullJitter, time.Second, 5.0, 0, rand.New(rngSrc)) return discimpl.NewBackoffConnector(host, cacheSize, dialTimeout, backoff) }, } return discoverOpts } // discover represents the discovery pipeline. // The discovery pipeline handles advertising and discovery of peers type discover struct { p *PubSub // discovery assists in discovering and advertising peers for a bitmask discovery discovery.Discovery // advertising tracks which bitmasks are being advertised advertising map[string]context.CancelFunc // discoverQ handles continuing peer discovery discoverQ chan *discoverReq // ongoing tracks ongoing discovery requests ongoing map[string]struct{} // done handles completion of a discovery request done chan string // connector handles connecting to new peers found via discovery connector *discimpl.BackoffConnector // options are the set of options to be used to complete struct construction in Start options *discoverOptions } // MinBitmaskSize returns a function that checks if a router is ready for publishing based on the bitmask size. // The router ultimately decides the whether it is ready or not, the given size is just a suggestion. Note // that the bitmask size does not include the router in the count. func MinBitmaskSize(size int) RouterReady { return func(rt PubSubRouter, bitmask []byte) (bool, error) { return rt.EnoughPeers(bitmask, size), nil } } // Start attaches the discovery pipeline to a pubsub instance, initializes discovery and starts event loop func (d *discover) Start(p *PubSub, opts ...DiscoverOpt) error { if d.discovery == nil || p == nil { return nil } d.p = p d.advertising = make(map[string]context.CancelFunc) d.discoverQ = make(chan *discoverReq, 32) d.ongoing = make(map[string]struct{}) d.done = make(chan string) conn, err := d.options.connFactory(p.host) if err != nil { return err } d.connector = conn go d.discoverLoop() go d.pollTimer() return nil } func (d *discover) pollTimer() { select { case <-time.After(DiscoveryPollInitialDelay): case <-d.p.ctx.Done(): return } select { case d.p.eval <- d.requestDiscovery: case <-d.p.ctx.Done(): return } ticker := time.NewTicker(DiscoveryPollInterval) defer ticker.Stop() for { select { case <-ticker.C: select { case d.p.eval <- d.requestDiscovery: case <-d.p.ctx.Done(): return } case <-d.p.ctx.Done(): return } } } func (d *discover) requestDiscovery() { for _, b := range d.p.myBitmasks { if !d.p.rt.EnoughPeers(b.bitmask, 0) { d.discoverQ <- &discoverReq{bitmask: b.bitmask, done: make(chan struct{}, 1)} } } } func (d *discover) discoverLoop() { for { select { case discover := <-d.discoverQ: bitmask := discover.bitmask if _, ok := d.ongoing[string(bitmask)]; ok { discover.done <- struct{}{} continue } d.ongoing[string(bitmask)] = struct{}{} go func() { d.handleDiscovery(d.p.ctx, bitmask, discover.opts) select { case d.done <- string(bitmask): case <-d.p.ctx.Done(): } discover.done <- struct{}{} }() case bitmask := <-d.done: delete(d.ongoing, bitmask) case <-d.p.ctx.Done(): return } } } // Advertise advertises this node's interest in a bitmask to a discovery service. Advertise is not thread-safe. func (d *discover) Advertise(bitmask []byte) { if d.discovery == nil { return } advertisingCtx, cancel := context.WithCancel(d.p.ctx) if _, ok := d.advertising[string(bitmask)]; ok { cancel() return } d.advertising[string(bitmask)] = cancel go func() { next, err := d.discovery.Advertise(advertisingCtx, string(bitmask)) if err != nil { log.Warnf("bootstrap: error providing rendezvous for %s: %s", bitmask, err.Error()) if next == 0 { next = discoveryAdvertiseRetryInterval } } t := time.NewTimer(next) defer t.Stop() for advertisingCtx.Err() == nil { select { case <-t.C: next, err = d.discovery.Advertise(advertisingCtx, string(bitmask)) if err != nil { log.Warnf("bootstrap: error providing rendezvous for %s: %s", bitmask, err.Error()) if next == 0 { next = discoveryAdvertiseRetryInterval } } t.Reset(next) case <-advertisingCtx.Done(): return } } }() } // StopAdvertise stops advertising this node's interest in a bitmask. StopAdvertise is not thread-safe. func (d *discover) StopAdvertise(bitmask []byte) { if d.discovery == nil { return } if advertiseCancel, ok := d.advertising[string(bitmask)]; ok { advertiseCancel() delete(d.advertising, string(bitmask)) } } // Discover searches for additional peers interested in a given bitmask func (d *discover) Discover(bitmask []byte, opts ...discovery.Option) { if d.discovery == nil { return } d.discoverQ <- &discoverReq{bitmask, opts, make(chan struct{}, 1)} } // Bootstrap attempts to bootstrap to a given bitmask. Returns true if bootstrapped successfully, false otherwise. func (d *discover) Bootstrap(ctx context.Context, bitmask []byte, ready RouterReady, opts ...discovery.Option) bool { if d.discovery == nil { return true } t := time.NewTimer(time.Hour) if !t.Stop() { <-t.C } defer t.Stop() for { // Check if ready for publishing bootstrapped := make(chan bool, 1) select { case d.p.eval <- func() { done, _ := ready(d.p.rt, bitmask) bootstrapped <- done }: if <-bootstrapped { return true } case <-d.p.ctx.Done(): return false case <-ctx.Done(): return false } // If not ready discover more peers disc := &discoverReq{bitmask, opts, make(chan struct{}, 1)} select { case d.discoverQ <- disc: case <-d.p.ctx.Done(): return false case <-ctx.Done(): return false } select { case <-disc.done: case <-d.p.ctx.Done(): return false case <-ctx.Done(): return false } t.Reset(time.Millisecond * 100) select { case <-t.C: case <-d.p.ctx.Done(): return false case <-ctx.Done(): return false } } } func (d *discover) handleDiscovery(ctx context.Context, bitmask []byte, opts []discovery.Option) { discoverCtx, cancel := context.WithTimeout(ctx, time.Second*10) defer cancel() peerCh, err := d.discovery.FindPeers(discoverCtx, string(bitmask), opts...) if err != nil { log.Debugf("error finding peers for bitmask %s: %v", bitmask, err) return } d.connector.Connect(ctx, peerCh) } type discoverReq struct { bitmask []byte opts []discovery.Option done chan struct{} } type pubSubDiscovery struct { discovery.Discovery opts []discovery.Option } func (d *pubSubDiscovery) Advertise(ctx context.Context, ns string, opts ...discovery.Option) (time.Duration, error) { return d.Discovery.Advertise(ctx, "floodsub:"+ns, append(opts, d.opts...)...) } func (d *pubSubDiscovery) FindPeers(ctx context.Context, ns string, opts ...discovery.Option) (<-chan peer.AddrInfo, error) { return d.Discovery.FindPeers(ctx, "floodsub:"+ns, append(opts, d.opts...)...) } // WithDiscoveryOpts passes libp2p Discovery options into the PubSub discovery subsystem func WithDiscoveryOpts(opts ...discovery.Option) DiscoverOpt { return func(d *discoverOptions) error { d.opts = opts return nil } } // BackoffConnectorFactory creates a BackoffConnector that is attached to a given host type BackoffConnectorFactory func(host host.Host) (*discimpl.BackoffConnector, error) // WithDiscoverConnector adds a custom connector that deals with how the discovery subsystem connects to peers func WithDiscoverConnector(connFactory BackoffConnectorFactory) DiscoverOpt { return func(d *discoverOptions) error { d.connFactory = connFactory return nil } }