fix: use default HTTP routers when FullRT DHT client is used (#9841)
Gus Eggert committed
May 1, 2023 at 15:29 UTC
ecf20f540b25905dc0a5b1505f72238de704bba2
4 files changed
+65
-53
cmd/ipfs/daemon.go
+2
-12
@@ -413,19 +413,9 @@ func daemonFunc(req *cmds.Request, re cmds.ResponseEmitter, env cmds.Environment
413
case routingOptionSupernodeKwd:
414
return errors.New("supernode routing was never fully implemented and has been removed")
415
case routingOptionDefaultKwd, routingOptionAutoKwd:
416
- ncfg.Routing = libp2p.ConstructDefaultRouting(
417
- cfg.Identity.PeerID,
418
- cfg.Addresses.Swarm,
419
- cfg.Identity.PrivKey,
420
- libp2p.DHTOption,
421
- )
416
+ ncfg.Routing = libp2p.ConstructDefaultRouting(cfg, libp2p.DHTOption)
417
case routingOptionAutoClientKwd:
423
- ncfg.Routing = libp2p.ConstructDefaultRouting(
424
- cfg.Identity.PeerID,
425
- cfg.Addresses.Swarm,
426
- cfg.Identity.PrivKey,
427
- libp2p.DHTClientOption,
428
- )
418
+ ncfg.Routing = libp2p.ConstructDefaultRouting(cfg, libp2p.DHTClientOption)
419
case routingOptionDHTClientKwd:
420
ncfg.Routing = libp2p.DHTClientOption
421
case routingOptionDHTKwd:
core/node/groups.go
+1
-1
@@ -165,7 +165,7 @@ func LibP2P(bcfg *BuildCfg, cfg *config.Config, userResourceOverrides rcmgr.Part
165
fx.Provide(libp2p.Routing),
166
fx.Provide(libp2p.ContentRouting),
167
168
- fx.Provide(libp2p.BaseRouting(cfg.Experimental.AcceleratedDHTClient)),
168
+ fx.Provide(libp2p.BaseRouting(cfg)),
169
maybeProvide(libp2p.PubsubRouter, bcfg.getOpt("ipnsps")),
170
171
maybeProvide(libp2p.BandwidthCounter, !cfg.Swarm.DisableBandwidthMetrics),
core/node/libp2p/routing.go
+29
-18
@@ -63,35 +63,34 @@ type processInitialRoutingOut struct {
63
64
type AddrInfoChan chan peer.AddrInfo
65
66
-func BaseRouting(experimentalDHTClient bool) interface{} {
66
+func BaseRouting(cfg *config.Config) interface{} {
67
return func(lc fx.Lifecycle, in processInitialRoutingIn) (out processInitialRoutingOut, err error) {
68
- var dr *ddht.DHT
68
+ var dualDHT *ddht.DHT
69
if dht, ok := in.Router.(*ddht.DHT); ok {
70
- dr = dht
70
+ dualDHT = dht
71
72
lc.Append(fx.Hook{
73
OnStop: func(ctx context.Context) error {
74
- return dr.Close()
74
+ return dualDHT.Close()
75
},
76
})
77
}
78
79
- if pr, ok := in.Router.(routinghelpers.ComposableRouter); ok {
80
- for _, r := range pr.Routers() {
79
+ if cr, ok := in.Router.(routinghelpers.ComposableRouter); ok {
80
+ for _, r := range cr.Routers() {
81
if dht, ok := r.(*ddht.DHT); ok {
82
- dr = dht
82
+ dualDHT = dht
83
lc.Append(fx.Hook{
84
OnStop: func(ctx context.Context) error {
85
- return dr.Close()
85
+ return dualDHT.Close()
86
},
87
})
88
-
88
break
89
}
90
}
91
}
92
94
- if dr != nil && experimentalDHTClient {
93
+ if dualDHT != nil && cfg.Experimental.AcceleratedDHTClient {
94
cfg, err := in.Repo.Config()
95
if err != nil {
96
return out, err
@@ -101,7 +100,7 @@ func BaseRouting(experimentalDHTClient bool) interface{} {
100
return out, err
101
}
102
104
- expClient, err := fullrt.NewFullRT(in.Host,
103
+ fullRTClient, err := fullrt.NewFullRT(in.Host,
104
dht.DefaultPrefix,
105
fullrt.DHTOption(
106
dht.Validator(in.Validator),
@@ -116,18 +115,30 @@ func BaseRouting(experimentalDHTClient bool) interface{} {
115
116
lc.Append(fx.Hook{
117
OnStop: func(ctx context.Context) error {
119
- return expClient.Close()
118
+ return fullRTClient.Close()
119
},
120
})
121
122
+ // we want to also use the default HTTP routers, so wrap the FullRT client
123
+ // in a parallel router that calls them in parallel
124
+ httpRouters, err := constructDefaultHTTPRouters(cfg)
125
+ if err != nil {
126
+ return out, err
127
+ }
128
+ routers := []*routinghelpers.ParallelRouter{
129
+ {Router: fullRTClient},
130
+ }
131
+ routers = append(routers, httpRouters...)
132
+ router := routinghelpers.NewComposableParallel(routers)
133
+
134
return processInitialRoutingOut{
135
Router: Router{
125
- Routing: expClient,
136
Priority: 1000,
137
+ Routing: router,
138
},
128
- DHT: dr,
129
- DHTClient: expClient,
130
- ContentRouter: expClient,
139
+ DHT: dualDHT,
140
+ DHTClient: fullRTClient,
141
+ ContentRouter: fullRTClient,
142
}, nil
143
}
144
@@ -136,8 +147,8 @@ func BaseRouting(experimentalDHTClient bool) interface{} {
147
Priority: 1000,
148
Routing: in.Router,
149
},
139
- DHT: dr,
140
- DHTClient: dr,
150
+ DHT: dualDHT,
151
+ DHTClient: dualDHT,
152
ContentRouter: in.Router,
153
}, nil
154
}
core/node/libp2p/routingopt.go
+33
-22
@@ -39,8 +39,35 @@ func init() {
39
}
40
}
41
42
+func constructDefaultHTTPRouters(cfg *config.Config) ([]*routinghelpers.ParallelRouter, error) {
43
+ var routers []*routinghelpers.ParallelRouter
44
+ // Append HTTP routers for additional speed
45
+ for _, endpoint := range defaultHTTPRouters {
46
+ httpRouter, err := irouting.ConstructHTTPRouter(endpoint, cfg.Identity.PeerID, cfg.Addresses.Swarm, cfg.Identity.PrivKey)
47
+ if err != nil {
48
+ return nil, err
49
+ }
50
+
51
+ r := &irouting.Composer{
52
+ GetValueRouter: routinghelpers.Null{},
53
+ PutValueRouter: routinghelpers.Null{},
54
+ ProvideRouter: routinghelpers.Null{}, // modify this when indexers supports provide
55
+ FindPeersRouter: routinghelpers.Null{},
56
+ FindProvidersRouter: httpRouter,
57
+ }
58
+
59
+ routers = append(routers, &routinghelpers.ParallelRouter{
60
+ Router: r,
61
+ IgnoreError: true, // https://github.com/ipfs/kubo/pull/9475#discussion_r1042507387
62
+ Timeout: 15 * time.Second, // 5x server value from https://github.com/ipfs/kubo/pull/9475#discussion_r1042428529
63
+ ExecuteAfter: 0,
64
+ })
65
+ }
66
+ return routers, nil
67
+}
68
+
69
// ConstructDefaultRouting returns routers used when Routing.Type is unset or set to "auto"
43
-func ConstructDefaultRouting(peerID string, addrs []string, privKey string, routingOpt RoutingOption) func(
70
+func ConstructDefaultRouting(cfg *config.Config, routingOpt RoutingOption) func(
71
ctx context.Context,
72
host host.Host,
73
dstore datastore.Batching,
@@ -68,29 +95,13 @@ func ConstructDefaultRouting(peerID string, addrs []string, privKey string, rout
95
ExecuteAfter: 0,
96
})
97
71
- // Append HTTP routers for additional speed
72
- for _, endpoint := range defaultHTTPRouters {
73
- httpRouter, err := irouting.ConstructHTTPRouter(endpoint, peerID, addrs, privKey)
74
- if err != nil {
75
- return nil, err
76
- }
77
-
78
- r := &irouting.Composer{
79
- GetValueRouter: routinghelpers.Null{},
80
- PutValueRouter: routinghelpers.Null{},
81
- ProvideRouter: routinghelpers.Null{}, // modify this when indexers supports provide
82
- FindPeersRouter: routinghelpers.Null{},
83
- FindProvidersRouter: httpRouter,
84
- }
85
-
86
- routers = append(routers, &routinghelpers.ParallelRouter{
87
- Router: r,
88
- IgnoreError: true, // https://github.com/ipfs/kubo/pull/9475#discussion_r1042507387
89
- Timeout: 15 * time.Second, // 5x server value from https://github.com/ipfs/kubo/pull/9475#discussion_r1042428529
90
- ExecuteAfter: 0,
91
- })
98
+ httpRouters, err := constructDefaultHTTPRouters(cfg)
99
+ if err != nil {
100
+ return nil, err
101
}
102
103
+ routers = append(routers, httpRouters...)
104
+
105
routing := routinghelpers.NewComposableParallel(routers)
106
return routing, nil
107
}