fix: use default HTTP routers when FullRT DHT client is used (#9841)
Gus Eggert committed
May 1, 2023 at 15:29 UTC
9482d7529233ff8942601a875c705fac55376c06
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
@@ -43,8 +43,35 @@ func init() {
43
}
44
}
45
46
+func constructDefaultHTTPRouters(cfg *config.Config) ([]*routinghelpers.ParallelRouter, error) {
47
+ var routers []*routinghelpers.ParallelRouter
48
+ // Append HTTP routers for additional speed
49
+ for _, endpoint := range defaultHTTPRouters {
50
+ httpRouter, err := irouting.ConstructHTTPRouter(endpoint, cfg.Identity.PeerID, cfg.Addresses.Swarm, cfg.Identity.PrivKey)
51
+ if err != nil {
52
+ return nil, err
53
+ }
54
+
55
+ r := &irouting.Composer{
56
+ GetValueRouter: routinghelpers.Null{},
57
+ PutValueRouter: routinghelpers.Null{},
58
+ ProvideRouter: routinghelpers.Null{}, // modify this when indexers supports provide
59
+ FindPeersRouter: routinghelpers.Null{},
60
+ FindProvidersRouter: httpRouter,
61
+ }
62
+
63
+ routers = append(routers, &routinghelpers.ParallelRouter{
64
+ Router: r,
65
+ IgnoreError: true, // https://github.com/ipfs/kubo/pull/9475#discussion_r1042507387
66
+ Timeout: 15 * time.Second, // 5x server value from https://github.com/ipfs/kubo/pull/9475#discussion_r1042428529
67
+ ExecuteAfter: 0,
68
+ })
69
+ }
70
+ return routers, nil
71
+}
72
+
73
// ConstructDefaultRouting returns routers used when Routing.Type is unset or set to "auto"
47
-func ConstructDefaultRouting(peerID string, addrs []string, privKey string, routingOpt RoutingOption) RoutingOption {
74
+func ConstructDefaultRouting(cfg *config.Config, routingOpt RoutingOption) RoutingOption {
75
return func(args RoutingOptionArgs) (routing.Routing, error) {
76
// Defined routers will be queried in parallel (optimizing for response speed)
77
// Different trade-offs can be made by setting Routing.Type = "custom" with own Routing.Routers
@@ -60,29 +87,13 @@ func ConstructDefaultRouting(peerID string, addrs []string, privKey string, rout
87
ExecuteAfter: 0,
88
})
89
63
- // Append HTTP routers for additional speed
64
- for _, endpoint := range defaultHTTPRouters {
65
- httpRouter, err := irouting.ConstructHTTPRouter(endpoint, peerID, addrs, privKey)
66
- if err != nil {
67
- return nil, err
68
- }
69
-
70
- r := &irouting.Composer{
71
- GetValueRouter: routinghelpers.Null{},
72
- PutValueRouter: routinghelpers.Null{},
73
- ProvideRouter: routinghelpers.Null{}, // modify this when indexers supports provide
74
- FindPeersRouter: routinghelpers.Null{},
75
- FindProvidersRouter: httpRouter,
76
- }
77
-
78
- routers = append(routers, &routinghelpers.ParallelRouter{
79
- Router: r,
80
- IgnoreError: true, // https://github.com/ipfs/kubo/pull/9475#discussion_r1042507387
81
- Timeout: 15 * time.Second, // 5x server value from https://github.com/ipfs/kubo/pull/9475#discussion_r1042428529
82
- ExecuteAfter: 0,
83
- })
90
+ httpRouters, err := constructDefaultHTTPRouters(cfg)
91
+ if err != nil {
92
+ return nil, err
93
}
94
95
+ routers = append(routers, httpRouters...)
96
+
97
routing := routinghelpers.NewComposableParallel(routers)
98
return routing, nil
99
}