swarm.go 10 KB
Newer Older
1
// Package swarm implements a connection muxer with a pair of channels
Juan Batiz-Benet's avatar
Juan Batiz-Benet committed
2 3 4 5
// to synchronize all network communication.
package swarm

import (
6
	"context"
Juan Batiz-Benet's avatar
Juan Batiz-Benet committed
7
	"fmt"
8
	"io/ioutil"
9 10
	"os"
	"strings"
Juan Batiz-Benet's avatar
Juan Batiz-Benet committed
11 12 13
	"sync"
	"time"

Jeromy's avatar
Jeromy committed
14 15 16 17 18 19 20 21 22 23 24 25 26 27 28
	logging "github.com/ipfs/go-log"
	"github.com/jbenet/goprocess"
	goprocessctx "github.com/jbenet/goprocess/context"
	addrutil "github.com/libp2p/go-addr-util"
	conn "github.com/libp2p/go-libp2p-conn"
	ci "github.com/libp2p/go-libp2p-crypto"
	ipnet "github.com/libp2p/go-libp2p-interface-pnet"
	metrics "github.com/libp2p/go-libp2p-metrics"
	mconn "github.com/libp2p/go-libp2p-metrics/conn"
	inet "github.com/libp2p/go-libp2p-net"
	peer "github.com/libp2p/go-libp2p-peer"
	pstore "github.com/libp2p/go-libp2p-peerstore"
	transport "github.com/libp2p/go-libp2p-transport"
	filter "github.com/libp2p/go-maddr-filter"
	ps "github.com/libp2p/go-peerstream"
Steven Allen's avatar
Steven Allen committed
29
	pst "github.com/libp2p/go-stream-muxer"
Jeromy's avatar
Jeromy committed
30
	tcpt "github.com/libp2p/go-tcp-transport"
Lars Gierth's avatar
Lars Gierth committed
31
	ws "github.com/libp2p/go-ws-transport"
Jeromy's avatar
Jeromy committed
32 33 34 35 36
	ma "github.com/multiformats/go-multiaddr"
	psmss "github.com/whyrusleeping/go-smux-multistream"
	spdy "github.com/whyrusleeping/go-smux-spdystream"
	yamux "github.com/whyrusleeping/go-smux-yamux"
	mafilter "github.com/whyrusleeping/multiaddr-filter"
Juan Batiz-Benet's avatar
Juan Batiz-Benet committed
37 38 39 40
)

var log = logging.Logger("swarm2")

41 42
// PSTransport is the default peerstream transport that will be used by
// any libp2p swarms.
Juan Batiz-Benet's avatar
Juan Batiz-Benet committed
43 44 45
var PSTransport pst.Transport

func init() {
46 47 48
	msstpt := psmss.NewBlankTransport()

	ymxtpt := &yamux.Transport{
49
		AcceptBacklog:          8192,
50 51 52
		ConnectionWriteTimeout: time.Second * 10,
		KeepAliveInterval:      time.Second * 30,
		EnableKeepAlive:        true,
53
		MaxStreamWindowSize:    uint32(1024 * 512),
54 55 56
		LogOutput:              ioutil.Discard,
	}

Jeromy's avatar
Jeromy committed
57 58
	msstpt.AddTransport("/yamux/1.0.0", ymxtpt)
	msstpt.AddTransport("/spdy/3.1.0", spdy.Transport)
59

60 61 62 63 64
	// allow overriding of muxer preferences
	if prefs := os.Getenv("LIBP2P_MUX_PREFS"); prefs != "" {
		msstpt.OrderPreference = strings.Fields(prefs)
	}

65
	PSTransport = msstpt
Juan Batiz-Benet's avatar
Juan Batiz-Benet committed
66 67 68 69 70 71 72 73 74 75 76
}

// Swarm is a connection muxer, allowing connections to other peers to
// be opened and closed, while still using the same Chan for all
// communication. The Chan sends/receives Messages, which note the
// destination or source Peer.
//
// Uses peerstream.Swarm
type Swarm struct {
	swarm *ps.Swarm
	local peer.ID
Jeromy's avatar
Jeromy committed
77
	peers pstore.Peerstore
Juan Batiz-Benet's avatar
Juan Batiz-Benet committed
78 79
	connh ConnHandler

80
	dsync *DialSync
Juan Batiz-Benet's avatar
Juan Batiz-Benet committed
81 82 83
	backf dialbackoff
	dialT time.Duration // mainly for tests

Jeromy's avatar
Jeromy committed
84 85
	dialer *conn.Dialer

Juan Batiz-Benet's avatar
Juan Batiz-Benet committed
86 87 88
	notifmu sync.RWMutex
	notifs  map[inet.Notifiee]ps.Notifiee

Jeromy's avatar
Jeromy committed
89 90
	transports []transport.Transport

Juan Batiz-Benet's avatar
Juan Batiz-Benet committed
91 92 93
	// filters for addresses that shouldnt be dialed
	Filters *filter.Filters

Jeromy's avatar
Jeromy committed
94 95 96
	// file descriptor rate limited
	fdRateLimit chan struct{}

Juan Batiz-Benet's avatar
Juan Batiz-Benet committed
97 98 99
	proc goprocess.Process
	ctx  context.Context
	bwc  metrics.Reporter
Jeromy's avatar
Jeromy committed
100 101

	limiter *dialLimiter
102 103

	protec ipnet.Protector
Juan Batiz-Benet's avatar
Juan Batiz-Benet committed
104 105
}

106
func NewSwarm(ctx context.Context, listenAddrs []ma.Multiaddr, local peer.ID,
107
	peers pstore.Peerstore, bwc metrics.Reporter) (*Swarm, error) {
108
	return NewSwarmWithProtector(ctx, listenAddrs, local, peers, nil, PSTransport, bwc)
109 110 111 112
}

// NewSwarm constructs a Swarm, with a Chan.
func NewSwarmWithProtector(ctx context.Context, listenAddrs []ma.Multiaddr, local peer.ID,
113
	peers pstore.Peerstore, protec ipnet.Protector, tpt pst.Transport, bwc metrics.Reporter) (*Swarm, error) {
Juan Batiz-Benet's avatar
Juan Batiz-Benet committed
114 115 116 117 118 119

	listenAddrs, err := filterAddrs(listenAddrs)
	if err != nil {
		return nil, err
	}

120 121 122 123 124
	var wrap func(c transport.Conn) transport.Conn
	if bwc != nil {
		wrap = func(c transport.Conn) transport.Conn {
			return mconn.WrapConn(bwc, c)
		}
Jeromy's avatar
Jeromy committed
125 126
	}

Juan Batiz-Benet's avatar
Juan Batiz-Benet committed
127
	s := &Swarm{
128
		swarm:  ps.NewSwarm(tpt),
129 130 131 132 133 134
		local:  local,
		peers:  peers,
		ctx:    ctx,
		dialT:  DialTimeout,
		notifs: make(map[inet.Notifiee]ps.Notifiee),
		transports: []transport.Transport{
135
			tcpt.NewTCPTransport(),
Jeromy's avatar
Jeromy committed
136
			new(ws.WebsocketTransport),
137
		},
Jeromy's avatar
Jeromy committed
138 139 140
		bwc:         bwc,
		fdRateLimit: make(chan struct{}, concurrentFdDials),
		Filters:     filter.NewFilters(),
141
		dialer:      conn.NewDialer(local, peers.PrivKey(local), wrap),
142
		protec:      protec,
Juan Batiz-Benet's avatar
Juan Batiz-Benet committed
143
	}
144
	s.dialer.Protector = protec
Juan Batiz-Benet's avatar
Juan Batiz-Benet committed
145

146
	s.dsync = NewDialSync(s.doDial)
Jeromy's avatar
Jeromy committed
147 148
	s.limiter = newDialLimiter(s.dialAddr)

Juan Batiz-Benet's avatar
Juan Batiz-Benet committed
149 150 151 152
	// configure Swarm
	s.proc = goprocessctx.WithContextAndTeardown(ctx, s.teardown)
	s.SetConnHandler(nil) // make sure to setup our own conn handler.

Jeromy's avatar
Jeromy committed
153 154 155 156 157 158
	err = s.setupInterfaces(listenAddrs)
	if err != nil {
		return nil, err
	}

	return s, nil
Juan Batiz-Benet's avatar
Juan Batiz-Benet committed
159 160
}

161
func NewBlankSwarm(ctx context.Context, id peer.ID, privkey ci.PrivKey, pstpt pst.Transport) *Swarm {
162
	s := &Swarm{
163
		swarm:       ps.NewSwarm(pstpt),
164 165 166 167 168 169 170
		local:       id,
		peers:       pstore.NewPeerstore(),
		ctx:         ctx,
		dialT:       DialTimeout,
		notifs:      make(map[inet.Notifiee]ps.Notifiee),
		fdRateLimit: make(chan struct{}, concurrentFdDials),
		Filters:     filter.NewFilters(),
171
		dialer:      conn.NewDialer(id, privkey, nil),
172 173 174 175 176 177 178 179 180 181 182 183 184 185
	}

	// configure Swarm
	s.limiter = newDialLimiter(s.dialAddr)
	s.proc = goprocessctx.WithContextAndTeardown(ctx, s.teardown)
	s.SetConnHandler(nil) // make sure to setup our own conn handler.

	return s
}

func (s *Swarm) AddTransport(t transport.Transport) {
	s.transports = append(s.transports, t)
}

Juan Batiz-Benet's avatar
Juan Batiz-Benet committed
186 187 188 189
func (s *Swarm) teardown() error {
	return s.swarm.Close()
}

190 191
// AddAddrFilter adds a multiaddr filter to the set of filters the swarm will
// use to determine which addresses not to dial to.
Juan Batiz-Benet's avatar
Juan Batiz-Benet committed
192 193 194 195 196 197 198 199 200
func (s *Swarm) AddAddrFilter(f string) error {
	m, err := mafilter.NewMask(f)
	if err != nil {
		return err
	}

	s.Filters.AddDialFilter(m)
	return nil
}
Jeromy's avatar
Jeromy committed
201

Juan Batiz-Benet's avatar
Juan Batiz-Benet committed
202 203 204 205 206 207 208 209
func filterAddrs(listenAddrs []ma.Multiaddr) ([]ma.Multiaddr, error) {
	if len(listenAddrs) > 0 {
		filtered := addrutil.FilterUsableAddrs(listenAddrs)
		if len(filtered) < 1 {
			return nil, fmt.Errorf("swarm cannot use any addr in: %s", listenAddrs)
		}
		listenAddrs = filtered
	}
Jeromy's avatar
Jeromy committed
210

Juan Batiz-Benet's avatar
Juan Batiz-Benet committed
211 212 213
	return listenAddrs, nil
}

214
// Listen sets up listeners for all of the given addresses
Juan Batiz-Benet's avatar
Juan Batiz-Benet committed
215 216 217 218 219 220
func (s *Swarm) Listen(addrs ...ma.Multiaddr) error {
	addrs, err := filterAddrs(addrs)
	if err != nil {
		return err
	}

Jeromy's avatar
Jeromy committed
221
	return s.setupInterfaces(addrs)
Juan Batiz-Benet's avatar
Juan Batiz-Benet committed
222 223 224 225 226 227 228 229 230 231 232 233 234 235 236 237 238 239 240 241 242 243 244 245 246 247 248 249 250 251 252 253 254 255 256 257 258 259 260 261 262 263 264 265 266 267 268
}

// Process returns the Process of the swarm
func (s *Swarm) Process() goprocess.Process {
	return s.proc
}

// Context returns the context of the swarm
func (s *Swarm) Context() context.Context {
	return s.ctx
}

// Close stops the Swarm.
func (s *Swarm) Close() error {
	return s.proc.Close()
}

// StreamSwarm returns the underlying peerstream.Swarm
func (s *Swarm) StreamSwarm() *ps.Swarm {
	return s.swarm
}

// SetConnHandler assigns the handler for new connections.
// See peerstream. You will rarely use this. See SetStreamHandler
func (s *Swarm) SetConnHandler(handler ConnHandler) {

	// handler is nil if user wants to clear the old handler.
	if handler == nil {
		s.swarm.SetConnHandler(func(psconn *ps.Conn) {
			s.connHandler(psconn)
		})
		return
	}

	s.swarm.SetConnHandler(func(psconn *ps.Conn) {
		// sc is nil if closed in our handler.
		if sc := s.connHandler(psconn); sc != nil {
			// call the user's handler. in a goroutine for sync safety.
			go handler(sc)
		}
	})
}

// SetStreamHandler assigns the handler for new streams.
// See peerstream.
func (s *Swarm) SetStreamHandler(handler inet.StreamHandler) {
	s.swarm.SetStreamHandler(func(s *ps.Stream) {
269
		handler((*Stream)(s))
Juan Batiz-Benet's avatar
Juan Batiz-Benet committed
270 271 272 273
	})
}

// NewStreamWithPeer creates a new stream on any available connection to p
274
func (s *Swarm) NewStreamWithPeer(ctx context.Context, p peer.ID) (*Stream, error) {
Juan Batiz-Benet's avatar
Juan Batiz-Benet committed
275 276 277
	// if we have no connections, try connecting.
	if len(s.ConnectionsToPeer(p)) == 0 {
		log.Debug("Swarm: NewStreamWithPeer no connections. Attempting to connect...")
278
		if _, err := s.Dial(ctx, p); err != nil {
Juan Batiz-Benet's avatar
Juan Batiz-Benet committed
279 280 281 282 283
			return nil, err
		}
	}
	log.Debug("Swarm: NewStreamWithPeer...")

284
	// TODO: think about passing a context down to NewStreamWithGroup
Juan Batiz-Benet's avatar
Juan Batiz-Benet committed
285
	st, err := s.swarm.NewStreamWithGroup(p)
286
	return (*Stream)(st), err
Juan Batiz-Benet's avatar
Juan Batiz-Benet committed
287 288 289 290 291 292 293
}

// ConnectionsToPeer returns all the live connections to p
func (s *Swarm) ConnectionsToPeer(p peer.ID) []*Conn {
	return wrapConns(ps.ConnsWithGroup(p, s.swarm.Conns()))
}

294 295 296 297 298 299 300 301 302
func (s *Swarm) HaveConnsToPeer(p peer.ID) bool {
	for _, c := range s.swarm.Conns() {
		if c.InGroup(p) {
			return true
		}
	}
	return false
}

Juan Batiz-Benet's avatar
Juan Batiz-Benet committed
303 304 305 306 307 308 309 310 311 312 313 314 315 316 317 318 319 320 321 322 323 324 325 326 327 328 329 330 331 332 333 334 335 336 337 338 339
// Connections returns a slice of all connections.
func (s *Swarm) Connections() []*Conn {
	return wrapConns(s.swarm.Conns())
}

// CloseConnection removes a given peer from swarm + closes the connection
func (s *Swarm) CloseConnection(p peer.ID) error {
	conns := s.swarm.ConnsWithGroup(p) // boom.
	for _, c := range conns {
		c.Close()
	}
	return nil
}

// Peers returns a copy of the set of peers swarm is connected to.
func (s *Swarm) Peers() []peer.ID {
	conns := s.Connections()

	seen := make(map[peer.ID]struct{})
	peers := make([]peer.ID, 0, len(conns))
	for _, c := range conns {
		p := c.RemotePeer()
		if _, found := seen[p]; found {
			continue
		}

		seen[p] = struct{}{}
		peers = append(peers, p)
	}
	return peers
}

// LocalPeer returns the local peer swarm is associated to.
func (s *Swarm) LocalPeer() peer.ID {
	return s.local
}

340
// Backoff returns the dialbackoff object for this swarm.
Jeromy's avatar
Jeromy committed
341 342 343 344
func (s *Swarm) Backoff() *dialbackoff {
	return &s.backf
}

Juan Batiz-Benet's avatar
Juan Batiz-Benet committed
345 346 347 348 349 350 351 352 353 354 355 356 357 358 359 360 361 362 363 364 365 366 367 368 369 370 371 372 373 374 375 376 377 378 379 380 381 382 383 384 385 386 387 388 389 390 391 392 393 394
// notifyAll sends a signal to all Notifiees
func (s *Swarm) notifyAll(notify func(inet.Notifiee)) {
	s.notifmu.RLock()
	for f := range s.notifs {
		go notify(f)
	}
	s.notifmu.RUnlock()
}

// Notify signs up Notifiee to receive signals when events happen
func (s *Swarm) Notify(f inet.Notifiee) {
	// wrap with our notifiee, to translate function calls
	n := &ps2netNotifee{net: (*Network)(s), not: f}

	s.notifmu.Lock()
	s.notifs[f] = n
	s.notifmu.Unlock()

	// register for notifications in the peer swarm.
	s.swarm.Notify(n)
}

// StopNotify unregisters Notifiee fromr receiving signals
func (s *Swarm) StopNotify(f inet.Notifiee) {
	s.notifmu.Lock()
	n, found := s.notifs[f]
	if found {
		delete(s.notifs, f)
	}
	s.notifmu.Unlock()

	if found {
		s.swarm.StopNotify(n)
	}
}

type ps2netNotifee struct {
	net *Network
	not inet.Notifiee
}

func (n *ps2netNotifee) Connected(c *ps.Conn) {
	n.not.Connected(n.net, inet.Conn((*Conn)(c)))
}

func (n *ps2netNotifee) Disconnected(c *ps.Conn) {
	n.not.Disconnected(n.net, inet.Conn((*Conn)(c)))
}

func (n *ps2netNotifee) OpenedStream(s *ps.Stream) {
395
	n.not.OpenedStream(n.net, (*Stream)(s))
Juan Batiz-Benet's avatar
Juan Batiz-Benet committed
396 397 398
}

func (n *ps2netNotifee) ClosedStream(s *ps.Stream) {
399
	n.not.ClosedStream(n.net, (*Stream)(s))
Juan Batiz-Benet's avatar
Juan Batiz-Benet committed
400
}