add option to enable go-multiplex experiment
License: MIT Signed-off-by: Jeromy <why@ipfs.io>
Jeromy committed
Nov 29, 2016 at 17:43 UTC
10ddd40f7b47a78cd28743c6dcd11cf97f3748ae
5 files changed
+62
-11
cmd/ipfs/daemon.go
+4
@@ -48,6 +48,7 @@ const (
48
unrestrictedApiAccessKwd = "unrestricted-api"
49
writableKwd = "writable"
50
enableFloodSubKwd = "enable-pubsub-experiment"
51
+ enableMultiplexKwd = "enable-mplex-experiment"
52
// apiAddrKwd = "address-api"
53
// swarmAddrKwd = "address-swarm"
54
)
@@ -158,6 +159,7 @@ Headers.
159
cmds.BoolOption(offlineKwd, "Run offline. Do not connect to the rest of the network but provide local API.").Default(false),
160
cmds.BoolOption(migrateKwd, "If true, assume yes at the migrate prompt. If false, assume no."),
161
cmds.BoolOption(enableFloodSubKwd, "Instantiate the ipfs daemon with the experimental pubsub feature enabled."),
162
+ cmds.BoolOption(enableMultiplexKwd, "Add the experimental 'go-multiplex' stream muxer to libp2p on construction."),
163
164
// TODO: add way to override addresses. tricky part: updating the config if also --init.
165
// cmds.StringOption(apiAddrKwd, "Address for the daemon rpc API (overrides config)"),
@@ -288,6 +290,7 @@ func daemonFunc(req cmds.Request, res cmds.Response) {
290
291
offline, _, _ := req.Option(offlineKwd).Bool()
292
pubsub, _, _ := req.Option(enableFloodSubKwd).Bool()
293
+ mplex, _, _ := req.Option(enableMultiplexKwd).Bool()
294
295
// Start assembling node config
296
ncfg := &core.BuildCfg{
@@ -296,6 +299,7 @@ func daemonFunc(req cmds.Request, res cmds.Response) {
299
Online: !offline,
300
ExtraOpts: map[string]bool{
301
"pubsub": pubsub,
302
+ "mplex": mplex,
303
},
304
//TODO(Kubuxu): refactor Online vs Offline by adding Permanent vs Ephemeral
305
}
core/builder.go
+1
-1
@@ -197,7 +197,7 @@ func setupNode(ctx context.Context, n *IpfsNode, cfg *BuildCfg) error {
197
198
if cfg.Online {
199
do := setupDiscoveryOption(rcfg.Discovery)
200
- if err := n.startOnlineServices(ctx, cfg.Routing, cfg.Host, do, cfg.getOpt("pubsub")); err != nil {
200
+ if err := n.startOnlineServices(ctx, cfg.Routing, cfg.Host, do, cfg.getOpt("pubsub"), cfg.getOpt("mplex")); err != nil {
201
return err
202
}
203
} else {
core/core.go
+45
-5
@@ -14,7 +14,10 @@ import (
14
"errors"
15
"fmt"
16
"io"
17
+ "io/ioutil"
18
"net"
19
+ "os"
20
+ "strings"
21
"time"
22
23
bstore "github.com/ipfs/go-ipfs/blocks/blockstore"
@@ -44,19 +47,24 @@ import (
47
mamask "gx/ipfs/QmSMZwvs3n4GBikZ7hKzT17c3bk65FmyZo2JqtJ16swqCv/multiaddr-filter"
48
logging "gx/ipfs/QmSpJByNKFX1sCsHBEp3R73FL4NF6FnQTEGyNAXHm2GS52/go-log"
49
b58 "gx/ipfs/QmT8rehPR3F6bmwL6zjUN8XpiDBFFpMP2myPdC6ApsWfJf/go-base58"
50
+ mssmux "gx/ipfs/QmTfjLsou9ic6L4KqCcmbLSZcdiFu8q1v6njKp121pbbXx/go-smux-multistream"
51
ma "gx/ipfs/QmUAQaWbKxGCUTuoQVvvicbQNZ9APF5pDGWyAZSe93AtKH/go-multiaddr"
52
floodsub "gx/ipfs/QmV5jot2GfVXmgvetHExJCa2hprebf3AKjprZtuwaXSr1v/floodsub"
53
addrutil "gx/ipfs/QmVDnc2zvyQm8LhT72n22THcshvH7j3qPMnhvjerQER62T/go-addr-util"
54
+ spdy "gx/ipfs/QmWUNsat6Jb19nC5CiJCDXepTkxjdxi3eZqeoB6mrmmaGu/go-smux-spdystream"
55
swarm "gx/ipfs/QmWfxnAiQ5TnnCgiX9ikVUKFNHRgGhbgKdx5DoKPELD7P4/go-libp2p-swarm"
56
+ mplex "gx/ipfs/QmXGevGDVTqeKdisBzaxEK4CJZqfxeXiVSWLaXaVWcG5on/go-smux-multiplex"
57
metrics "gx/ipfs/QmY2otvyPM2sTaDsczo7Yuosg98sUMCJ9qx1gpPaAPTS9B/go-libp2p-metrics"
58
u "gx/ipfs/Qmb912gdngC1UWwTkhuW8knyRbcWeu5kqkxBpveLmW8bSr/go-ipfs-util"
59
routing "gx/ipfs/QmbkGVaN9W6RYJK4Ws5FvMKXKDqdRQ5snhtaa92qP6L8eU/go-libp2p-routing"
60
+ yamux "gx/ipfs/Qmbn7RYyWzBVXiUp9jZ1dA4VADHy9DtS7iZLwfhEUQvm3U/go-smux-yamux"
61
discovery "gx/ipfs/QmbzCT1CwxVZ2ednptC9RavuJe7Bv8DDi2Ne89qUrA37XM/go-libp2p/p2p/discovery"
62
p2pbhost "gx/ipfs/QmbzCT1CwxVZ2ednptC9RavuJe7Bv8DDi2Ne89qUrA37XM/go-libp2p/p2p/host/basic"
63
rhost "gx/ipfs/QmbzCT1CwxVZ2ednptC9RavuJe7Bv8DDi2Ne89qUrA37XM/go-libp2p/p2p/host/routed"
64
ping "gx/ipfs/QmbzCT1CwxVZ2ednptC9RavuJe7Bv8DDi2Ne89qUrA37XM/go-libp2p/p2p/protocol/ping"
65
cid "gx/ipfs/QmcTcsTvfaeEBRFo1TkFgT8sRmgi1n1LTZpecfVP8fzpGD/go-cid"
66
pstore "gx/ipfs/QmeXj9VAjmYQZxpmVz7VzccbJrpmr8qkCDSjfVNsPTWTYU/go-libp2p-peerstore"
67
+ smux "gx/ipfs/QmeZBgYBHvxMukGK5ojg28BCNLB9SeXqT7XXg6o7r2GbJy/go-stream-muxer"
68
peer "gx/ipfs/QmfMmLGoKzCHDN7cGgk64PJr4iipzidDRME8HABSJqvmhC/go-libp2p-peer"
69
ic "gx/ipfs/QmfWDLQjGjVe4fr5CoztYW2DYYjRysMJrFe1RCsXLPTf46/go-libp2p-crypto"
70
)
@@ -129,7 +137,7 @@ type Mounts struct {
137
Ipns mount.Mount
138
}
139
132
-func (n *IpfsNode) startOnlineServices(ctx context.Context, routingOption RoutingOption, hostOption HostOption, do DiscoveryOption, pubsub bool) error {
140
+func (n *IpfsNode) startOnlineServices(ctx context.Context, routingOption RoutingOption, hostOption HostOption, do DiscoveryOption, pubsub, mplex bool) error {
141
142
if n.PeerHost != nil { // already online.
143
return errors.New("node already online")
@@ -159,7 +167,9 @@ func (n *IpfsNode) startOnlineServices(ctx context.Context, routingOption Routin
167
n.Reporter = metrics.NewBandwidthCounter()
168
}
169
162
- peerhost, err := hostOption(ctx, n.Identity, n.Peerstore, n.Reporter, addrfilter)
170
+ tpt := makeSmuxTransport(mplex)
171
+
172
+ peerhost, err := hostOption(ctx, n.Identity, n.Peerstore, n.Reporter, addrfilter, tpt)
173
if err != nil {
174
return err
175
}
@@ -207,6 +217,34 @@ func (n *IpfsNode) startOnlineServices(ctx context.Context, routingOption Routin
217
return n.Bootstrap(DefaultBootstrapConfig)
218
}
219
220
+func makeSmuxTransport(mplexExp bool) smux.Transport {
221
+ mstpt := mssmux.NewBlankTransport()
222
+
223
+ ymxtpt := &yamux.Transport{
224
+ AcceptBacklog: 8192,
225
+ ConnectionWriteTimeout: time.Second * 10,
226
+ KeepAliveInterval: time.Second * 30,
227
+ EnableKeepAlive: true,
228
+ MaxStreamWindowSize: uint32(1024 * 512),
229
+ LogOutput: ioutil.Discard,
230
+ }
231
+
232
+ mstpt.AddTransport("/yamux/1.0.0", ymxtpt)
233
+
234
+ mstpt.AddTransport("/spdy/3.1.0", spdy.Transport)
235
+
236
+ if mplexExp {
237
+ mstpt.AddTransport("/mplex/6.7.0", mplex.DefaultTransport)
238
+ }
239
+
240
+ // Allow muxer preference order overriding
241
+ if prefs := os.Getenv("LIBP2P_MUX_PREFS"); prefs != "" {
242
+ mstpt.OrderPreference = strings.Fields(prefs)
243
+ }
244
+
245
+ return mstpt
246
+}
247
+
248
func setupDiscoveryOption(d config.Discovery) DiscoveryOption {
249
if d.MDNS.Enabled {
250
return func(ctx context.Context, h p2phost.Host) (discovery.Service, error) {
@@ -616,19 +654,21 @@ func listenAddresses(cfg *config.Config) ([]ma.Multiaddr, error) {
654
return listen, nil
655
}
656
619
-type HostOption func(ctx context.Context, id peer.ID, ps pstore.Peerstore, bwr metrics.Reporter, fs []*net.IPNet) (p2phost.Host, error)
657
+type HostOption func(ctx context.Context, id peer.ID, ps pstore.Peerstore, bwr metrics.Reporter, fs []*net.IPNet, tpt smux.Transport) (p2phost.Host, error)
658
659
var DefaultHostOption HostOption = constructPeerHost
660
661
// isolates the complex initialization steps
624
-func constructPeerHost(ctx context.Context, id peer.ID, ps pstore.Peerstore, bwr metrics.Reporter, fs []*net.IPNet) (p2phost.Host, error) {
662
+func constructPeerHost(ctx context.Context, id peer.ID, ps pstore.Peerstore, bwr metrics.Reporter, fs []*net.IPNet, tpt smux.Transport) (p2phost.Host, error) {
663
664
// no addresses to begin with. we'll start later.
627
- network, err := swarm.NewNetwork(ctx, nil, id, ps, bwr)
665
+ swrm, err := swarm.NewSwarmWithProtector(ctx, nil, id, ps, nil, tpt, bwr)
666
if err != nil {
667
return nil, err
668
}
669
670
+ network := (*swarm.Network)(swrm)
671
+
672
for _, f := range fs {
673
network.Swarm().Filters.AddDialFilter(f)
674
}
core/mock/mock.go
+6
-5
@@ -1,22 +1,23 @@
1
package coremock
2
3
import (
4
+ "context"
5
"net"
6
6
- context "context"
7
- "gx/ipfs/QmRWDav6mzWseLWeYfVd5fvUKiVe9xNH29YfMF438fG364/go-datastore"
8
- syncds "gx/ipfs/QmRWDav6mzWseLWeYfVd5fvUKiVe9xNH29YfMF438fG364/go-datastore/sync"
9
-
7
commands "github.com/ipfs/go-ipfs/commands"
8
core "github.com/ipfs/go-ipfs/core"
9
"github.com/ipfs/go-ipfs/repo"
10
config "github.com/ipfs/go-ipfs/repo/config"
11
ds2 "github.com/ipfs/go-ipfs/thirdparty/datastore2"
12
testutil "github.com/ipfs/go-ipfs/thirdparty/testutil"
13
+
14
host "gx/ipfs/QmPTGbC34bPKaUm9wTxBo7zSCac7pDuG42ZmnXC718CKZZ/go-libp2p-host"
15
+ "gx/ipfs/QmRWDav6mzWseLWeYfVd5fvUKiVe9xNH29YfMF438fG364/go-datastore"
16
+ syncds "gx/ipfs/QmRWDav6mzWseLWeYfVd5fvUKiVe9xNH29YfMF438fG364/go-datastore/sync"
17
metrics "gx/ipfs/QmY2otvyPM2sTaDsczo7Yuosg98sUMCJ9qx1gpPaAPTS9B/go-libp2p-metrics"
18
mocknet "gx/ipfs/QmbzCT1CwxVZ2ednptC9RavuJe7Bv8DDi2Ne89qUrA37XM/go-libp2p/p2p/net/mock"
19
pstore "gx/ipfs/QmeXj9VAjmYQZxpmVz7VzccbJrpmr8qkCDSjfVNsPTWTYU/go-libp2p-peerstore"
20
+ smux "gx/ipfs/QmeZBgYBHvxMukGK5ojg28BCNLB9SeXqT7XXg6o7r2GbJy/go-stream-muxer"
21
peer "gx/ipfs/QmfMmLGoKzCHDN7cGgk64PJr4iipzidDRME8HABSJqvmhC/go-libp2p-peer"
22
)
23
@@ -32,7 +33,7 @@ func NewMockNode() (*core.IpfsNode, error) {
33
}
34
35
func MockHostOption(mn mocknet.Mocknet) core.HostOption {
35
- return func(ctx context.Context, id peer.ID, ps pstore.Peerstore, bwr metrics.Reporter, fs []*net.IPNet) (host.Host, error) {
36
+ return func(ctx context.Context, id peer.ID, ps pstore.Peerstore, bwr metrics.Reporter, fs []*net.IPNet, _ smux.Transport) (host.Host, error) {
37
return mn.AddPeerWithPeerstore(id, ps)
38
}
39
}
package.json
+6
@@ -287,6 +287,12 @@
287
"hash": "QmXuBJ7DR6k3rmUEKtvVMhwjmXDuJgXXPUt4LQXKBMsU93",
288
"name": "go-os-helper",
289
"version": "0.0.0"
290
+ },
291
+ {
292
+ "author": "whyrusleeping",
293
+ "hash": "QmXGevGDVTqeKdisBzaxEK4CJZqfxeXiVSWLaXaVWcG5on",
294
+ "name": "go-smux-multiplex",
295
+ "version": "1.1.4"
296
}
297
],
298
"gxVersion": "0.4.0",