@cryptotaxi247 / kubo / commits / 1a3752b81

core: setup peerhost + listen separate steps

We had a problem: we were starting all the services with the network live, and so would miss early messages. We were noticing bitswap messages not handled (not in muxer). Many of the subsystems expect the network to _exist_ when they start up, so we split construction and starting to listen into two separate steps.

Juan Batiz-Benet committed Jan 30, 2015 at 20:25 UTC 1a3752b81f939b80a6429e170b9acad4e06db92e
1 file changed +48 -27
core/core.go
+48 -27
@@ -214,30 +214,39 @@ func (n *IpfsNode) StartOnlineServices(ctx context.Context) error {
214 return err
215 }
216
217 - peerhost, err := constructPeerHost(ctx, n.Repo.Config(), n.Identity, n.Peerstore)
217 + peerhost, err := constructPeerHost(ctx, n.Identity, n.Peerstore)
218 if err != nil {
219 return debugerror.Wrap(err)
220 }
221 n.PeerHost = peerhost
222
223 - // setup diagnostics service
224 - n.Diagnostics = diag.NewDiagnostics(n.Identity, n.PeerHost)
223 + // this block is the set of services which need to be initialized with the host
224 + // and _before_ we start listening.
225 + {
226 + // setup diagnostics service
227 + n.Diagnostics = diag.NewDiagnostics(n.Identity, n.PeerHost)
228
226 - // setup routing service
227 - dhtRouting, err := constructDHTRouting(ctx, n.PeerHost, n.Repo.Datastore())
228 - if err != nil {
229 - return debugerror.Wrap(err)
230 - }
231 - n.Routing = dhtRouting
229 + // setup routing service
230 + dhtRouting, err := constructDHTRouting(ctx, n.PeerHost, n.Repo.Datastore())
231 + if err != nil {
232 + return debugerror.Wrap(err)
233 + }
234 + n.Routing = dhtRouting
235
233 - // setup exchange service
234 - const alwaysSendToPeer = true // use YesManStrategy
235 - bitswapNetwork := bsnet.NewFromIpfsHost(n.PeerHost, n.Routing)
236 - n.Exchange = bitswap.New(ctx, n.Identity, bitswapNetwork, n.Blockstore, alwaysSendToPeer)
236 + // setup exchange service
237 + const alwaysSendToPeer = true // use YesManStrategy
238 + bitswapNetwork := bsnet.NewFromIpfsHost(n.PeerHost, n.Routing)
239 + n.Exchange = bitswap.New(ctx, n.Identity, bitswapNetwork, n.Blockstore, alwaysSendToPeer)
240
238 - // setup name system
239 - // TODO implement an offline namesys that serves only local names.
240 - n.Namesys = namesys.NewNameSystem(n.Routing)
241 + // setup name system
242 + // TODO implement an offline namesys that serves only local names.
243 + n.Namesys = namesys.NewNameSystem(n.Routing)
244 + }
245 +
246 + // Ok, now we're ready to listen.
247 + if err := startListening(ctx, n.PeerHost, n.Repo.Config()); err != nil {
248 + return debugerror.Wrap(err)
249 + }
250
251 n.Reprovider = rp.NewReprovider(n.Routing, n.Blockstore)
252 go n.Reprovider.ProvideEvery(ctx, kReprovideFrequency)
@@ -405,37 +414,49 @@ func listenAddresses(cfg *config.Config) ([]ma.Multiaddr, error) {
414 }
415
416 // isolates the complex initialization steps
408 -func constructPeerHost(ctx context.Context, cfg *config.Config, id peer.ID, ps peer.Peerstore) (p2phost.Host, error) {
409 - listenAddrs, err := listenAddresses(cfg)
417 +func constructPeerHost(ctx context.Context, id peer.ID, ps peer.Peerstore) (p2phost.Host, error) {
418 +
419 + // no addresses to begin with. we'll start later.
420 + network, err := swarm.NewNetwork(ctx, nil, id, ps)
421 if err != nil {
422 return nil, debugerror.Wrap(err)
423 }
424
425 + host := p2pbhost.New(network, p2pbhost.NATPortMap)
426 + return host, nil
427 +}
428 +
429 +// startListening on the network addresses
430 +func startListening(ctx context.Context, host p2phost.Host, cfg *config.Config) error {
431 + listenAddrs, err := listenAddresses(cfg)
432 + if err != nil {
433 + return debugerror.Wrap(err)
434 + }
435 +
436 // make sure we error out if our config does not have addresses we can use
437 log.Debugf("Config.Addresses.Swarm:%s", listenAddrs)
438 filteredAddrs := addrutil.FilterUsableAddrs(listenAddrs)
439 log.Debugf("Config.Addresses.Swarm:%s (filtered)", filteredAddrs)
440 if len(filteredAddrs) < 1 {
419 - return nil, debugerror.Errorf("addresses in config not usable: %s", listenAddrs)
441 + return debugerror.Errorf("addresses in config not usable: %s", listenAddrs)
442 }
443
422 - network, err := swarm.NewNetwork(ctx, filteredAddrs, id, ps)
423 - if err != nil {
424 - return nil, debugerror.Wrap(err)
444 + // Actually start listening:
445 + if err := host.Network().Listen(filteredAddrs...); err != nil {
446 + return err
447 }
448
427 - peerhost := p2pbhost.New(network, p2pbhost.NATPortMap)
449 // explicitly set these as our listen addrs.
450 // (why not do it inside inet.NewNetwork? because this way we can
451 // listen on addresses without necessarily advertising those publicly.)
431 - addrs, err := peerhost.Network().InterfaceListenAddresses()
452 + addrs, err := host.Network().InterfaceListenAddresses()
453 if err != nil {
433 - return nil, debugerror.Wrap(err)
454 + return debugerror.Wrap(err)
455 }
456 log.Infof("Swarm listening at: %s", addrs)
457
437 - ps.AddAddresses(id, addrs)
438 - return peerhost, nil
458 + host.Peerstore().AddAddresses(host.ID(), addrs)
459 + return nil
460 }
461
462 func constructDHTRouting(ctx context.Context, host p2phost.Host, ds datastore.ThreadSafeDatastore) (*dht.IpfsDHT, error) {