swarm.go 11.7 KB
Newer Older
Juan Batiz-Benet's avatar
Juan Batiz-Benet committed
1 2 3
package swarm

import (
4
	"context"
Steven Allen's avatar
Steven Allen committed
5
	"errors"
Juan Batiz-Benet's avatar
Juan Batiz-Benet committed
6
	"fmt"
7
	"strings"
Juan Batiz-Benet's avatar
Juan Batiz-Benet committed
8
	"sync"
Steven Allen's avatar
Steven Allen committed
9
	"sync/atomic"
Juan Batiz-Benet's avatar
Juan Batiz-Benet committed
10 11
	"time"

Jeromy's avatar
Jeromy committed
12 13 14 15 16 17 18 19 20 21
	logging "github.com/ipfs/go-log"
	"github.com/jbenet/goprocess"
	goprocessctx "github.com/jbenet/goprocess/context"
	metrics "github.com/libp2p/go-libp2p-metrics"
	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"
	mafilter "github.com/whyrusleeping/multiaddr-filter"
Juan Batiz-Benet's avatar
Juan Batiz-Benet committed
22 23
)

24 25
// DialTimeoutLocal is the maximum duration a Dial to local network address
// is allowed to take.
Steven Allen's avatar
Steven Allen committed
26 27
// This includes the time between dialing the raw network connection,
// protocol selection as well the handshake, if applicable.
28
var DialTimeoutLocal = 5 * time.Second
29

Steven Allen's avatar
Steven Allen committed
30
var log = logging.Logger("swarm2")
31

Steven Allen's avatar
Steven Allen committed
32 33
// ErrSwarmClosed is returned when one attempts to operate on a closed swarm.
var ErrSwarmClosed = errors.New("swarm closed")
34

Steven Allen's avatar
Steven Allen committed
35 36 37 38
// ErrAddrFiltered is returned when trying to register a connection to a
// filtered address. You shouldn't see this error unless some underlying
// transport is misbehaving.
var ErrAddrFiltered = errors.New("address filtered")
Juan Batiz-Benet's avatar
Juan Batiz-Benet committed
39 40 41 42 43 44

// 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.
type Swarm struct {
Steven Allen's avatar
Steven Allen committed
45 46 47 48
	// Close refcount. This allows us to fully wait for the swarm to be torn
	// down before continuing.
	refs sync.WaitGroup

Juan Batiz-Benet's avatar
Juan Batiz-Benet committed
49
	local peer.ID
Jeromy's avatar
Jeromy committed
50
	peers pstore.Peerstore
Juan Batiz-Benet's avatar
Juan Batiz-Benet committed
51

Steven Allen's avatar
Steven Allen committed
52 53 54 55
	conns struct {
		sync.RWMutex
		m map[peer.ID][]*Conn
	}
Juan Batiz-Benet's avatar
Juan Batiz-Benet committed
56

Steven Allen's avatar
Steven Allen committed
57 58 59 60
	listeners struct {
		sync.RWMutex
		m map[transport.Listener]struct{}
	}
Jeromy's avatar
Jeromy committed
61

Steven Allen's avatar
Steven Allen committed
62 63 64 65
	notifs struct {
		sync.RWMutex
		m map[inet.Notifiee]struct{}
	}
Juan Batiz-Benet's avatar
Juan Batiz-Benet committed
66

Steven Allen's avatar
Steven Allen committed
67 68 69 70
	transports struct {
		sync.RWMutex
		m map[int]transport.Transport
	}
Jeromy's avatar
Jeromy committed
71

Steven Allen's avatar
Steven Allen committed
72 73 74
	// new connection and stream handlers
	connh   atomic.Value
	streamh atomic.Value
Juan Batiz-Benet's avatar
Juan Batiz-Benet committed
75

Steven Allen's avatar
Steven Allen committed
76 77 78 79 80 81 82
	// dialing helpers
	dsync   *DialSync
	backf   DialBackoff
	limiter *dialLimiter

	// filters for addresses that shouldnt be dialed (or accepted)
	Filters *filter.Filters
Jeromy's avatar
Jeromy committed
83

Juan Batiz-Benet's avatar
Juan Batiz-Benet committed
84 85 86
	proc goprocess.Process
	ctx  context.Context
	bwc  metrics.Reporter
87 88
}

Steven Allen's avatar
Steven Allen committed
89 90
// NewSwarm constructs a Swarm
func NewSwarm(ctx context.Context, local peer.ID, peers pstore.Peerstore, bwc metrics.Reporter) *Swarm {
Juan Batiz-Benet's avatar
Juan Batiz-Benet committed
91
	s := &Swarm{
Steven Allen's avatar
Steven Allen committed
92 93 94 95
		local:   local,
		peers:   peers,
		bwc:     bwc,
		Filters: filter.NewFilters(),
Juan Batiz-Benet's avatar
Juan Batiz-Benet committed
96
	}
Steven Allen's avatar
Steven Allen committed
97 98 99 100 101

	s.conns.m = make(map[peer.ID][]*Conn)
	s.listeners.m = make(map[transport.Listener]struct{})
	s.transports.m = make(map[int]transport.Transport)
	s.notifs.m = make(map[inet.Notifiee]struct{})
Juan Batiz-Benet's avatar
Juan Batiz-Benet committed
102

103
	s.dsync = NewDialSync(s.doDial)
Jeromy's avatar
Jeromy committed
104
	s.limiter = newDialLimiter(s.dialAddr)
Juan Batiz-Benet's avatar
Juan Batiz-Benet committed
105
	s.proc = goprocessctx.WithContextAndTeardown(ctx, s.teardown)
Steven Allen's avatar
Steven Allen committed
106
	s.ctx = goprocessctx.OnClosingContext(s.proc)
Jeromy's avatar
Jeromy committed
107

Steven Allen's avatar
Steven Allen committed
108
	return s
Juan Batiz-Benet's avatar
Juan Batiz-Benet committed
109 110
}

Steven Allen's avatar
Steven Allen committed
111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132
func (s *Swarm) teardown() error {
	// Prevents new connections and/or listeners from being added to the swarm.

	s.listeners.Lock()
	listeners := s.listeners.m
	s.listeners.m = nil
	s.listeners.Unlock()

	s.conns.Lock()
	conns := s.conns.m
	s.conns.m = nil
	s.conns.Unlock()

	// Lots of go routines but we might as well do this in parallel. We want
	// to shut down as fast as possible.

	for l := range listeners {
		go func(l transport.Listener) {
			if err := l.Close(); err != nil {
				log.Errorf("error when shutting down listener: %s", err)
			}
		}(l)
133 134
	}

Steven Allen's avatar
Steven Allen committed
135 136 137 138 139 140 141 142 143
	for _, cs := range conns {
		for _, c := range cs {
			go func(c *Conn) {
				if err := c.Close(); err != nil {
					log.Errorf("error when shutting down connection: %s", err)
				}
			}(c)
		}
	}
144

Steven Allen's avatar
Steven Allen committed
145 146
	// Wait for everything to finish.
	s.refs.Wait()
147

Steven Allen's avatar
Steven Allen committed
148
	return nil
Juan Batiz-Benet's avatar
Juan Batiz-Benet committed
149 150
}

151 152
// 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
153 154 155 156 157 158 159 160 161
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
162

Steven Allen's avatar
Steven Allen committed
163 164 165 166 167
// Process returns the Process of the swarm
func (s *Swarm) Process() goprocess.Process {
	return s.proc
}

168
func (s *Swarm) addConn(tc transport.Conn, dir inet.Direction) (*Conn, error) {
Steven Allen's avatar
Steven Allen committed
169 170 171 172 173 174
	// The underlying transport (or the dialer) *should* filter it's own
	// connections but we should double check anyways.
	raddr := tc.RemoteMultiaddr()
	if s.Filters.AddrBlocked(raddr) {
		tc.Close()
		return nil, ErrAddrFiltered
Juan Batiz-Benet's avatar
Juan Batiz-Benet committed
175
	}
Jeromy's avatar
Jeromy committed
176

Steven Allen's avatar
Steven Allen committed
177
	p := tc.RemotePeer()
Juan Batiz-Benet's avatar
Juan Batiz-Benet committed
178

Steven Allen's avatar
Steven Allen committed
179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196
	// Add the public key.
	if pk := tc.RemotePublicKey(); pk != nil {
		s.peers.AddPubKey(p, pk)
	}

	// Clear any backoffs
	s.backf.Clear(p)

	// Finally, add the peer.
	s.conns.Lock()
	// Check if we're still online
	if s.conns.m == nil {
		s.conns.Unlock()
		tc.Close()
		return nil, ErrSwarmClosed
	}

	// Wrap and register the connection.
197
	stat := inet.Stat{Direction: dir}
Steven Allen's avatar
Steven Allen committed
198 199 200
	c := &Conn{
		conn:  tc,
		swarm: s,
201
		stat:  stat,
Juan Batiz-Benet's avatar
Juan Batiz-Benet committed
202
	}
Steven Allen's avatar
Steven Allen committed
203 204 205 206 207 208 209 210 211 212 213 214 215 216 217 218 219 220 221 222 223
	c.streams.m = make(map[*Stream]struct{})
	s.conns.m[p] = append(s.conns.m[p], c)

	// Add two swarm refs:
	// * One will be decremented after the close notifications fire in Conn.doClose
	// * The other will be decremented when Conn.start exits.
	s.refs.Add(2)

	// Take the notification lock before releasing the conns lock to block
	// Disconnect notifications until after the Connect notifications done.
	c.notifyLk.Lock()
	s.conns.Unlock()

	// We have a connection now. Cancel all other in-progress dials.
	// This should be fast, no reason to wait till later.
	s.dsync.CancelDial(p)

	s.notifyAll(func(f inet.Notifiee) {
		f.Connected(s, c)
	})
	c.notifyLk.Unlock()
Juan Batiz-Benet's avatar
Juan Batiz-Benet committed
224

Steven Allen's avatar
Steven Allen committed
225 226 227 228 229 230 231 232 233 234
	c.start()

	// TODO: Get rid of this. We use it for identify but that happen much
	// earlier (really, inside the transport and, if not then, during the
	// notifications).
	if h := s.ConnHandler(); h != nil {
		go h(c)
	}

	return c, nil
Juan Batiz-Benet's avatar
Juan Batiz-Benet committed
235 236
}

Steven Allen's avatar
Steven Allen committed
237 238 239
// Peerstore returns this swarms internal Peerstore.
func (s *Swarm) Peerstore() pstore.Peerstore {
	return s.peers
Juan Batiz-Benet's avatar
Juan Batiz-Benet committed
240 241 242 243 244 245 246 247 248 249 250 251
}

// 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()
}

Steven Allen's avatar
Steven Allen committed
252
// TODO: We probably don't need the conn handlers.
Juan Batiz-Benet's avatar
Juan Batiz-Benet committed
253 254

// SetConnHandler assigns the handler for new connections.
Steven Allen's avatar
Steven Allen committed
255 256 257 258
// You will rarely use this. See SetStreamHandler
func (s *Swarm) SetConnHandler(handler inet.ConnHandler) {
	s.connh.Store(handler)
}
Juan Batiz-Benet's avatar
Juan Batiz-Benet committed
259

Steven Allen's avatar
Steven Allen committed
260 261 262 263
// ConnHandler gets the handler for new connections.
func (s *Swarm) ConnHandler() inet.ConnHandler {
	handler, _ := s.connh.Load().(inet.ConnHandler)
	return handler
Juan Batiz-Benet's avatar
Juan Batiz-Benet committed
264 265 266 267
}

// SetStreamHandler assigns the handler for new streams.
func (s *Swarm) SetStreamHandler(handler inet.StreamHandler) {
Steven Allen's avatar
Steven Allen committed
268 269 270 271 272 273 274 275 276 277 278 279 280 281 282 283 284 285 286 287 288 289 290 291 292 293 294 295 296 297 298 299 300 301 302 303 304 305 306 307 308 309 310 311 312 313
	s.streamh.Store(handler)
}

// StreamHandler gets the handler for new streams.
func (s *Swarm) StreamHandler() inet.StreamHandler {
	handler, _ := s.streamh.Load().(inet.StreamHandler)
	return handler
}

// NewStream creates a new stream on any available connection to peer, dialing
// if necessary.
func (s *Swarm) NewStream(ctx context.Context, p peer.ID) (inet.Stream, error) {
	log.Debugf("[%s] opening stream to peer [%s]", s.local, p)

	// Algorithm:
	// 1. Find the best connection, otherwise, dial.
	// 2. Try opening a stream.
	// 3. If the underlying connection is, in fact, closed, close the outer
	//    connection and try again. We do this in case we have a closed
	//    connection but don't notice it until we actually try to open a
	//    stream.
	//
	// Note: We only dial once.
	//
	// TODO: Try all connections even if we get an error opening a stream on
	// a non-closed connection.
	dials := 0
	for {
		c := s.bestConnToPeer(p)
		if c == nil {
			if dials >= DialAttempts {
				return nil, errors.New("max dial attempts exceeded")
			}
			dials++

			var err error
			c, err = s.dialPeer(ctx, p)
			if err != nil {
				return nil, err
			}
		}
		s, err := c.NewStream()
		if err != nil {
			if c.conn.IsClosed() {
				continue
			}
Juan Batiz-Benet's avatar
Juan Batiz-Benet committed
314 315
			return nil, err
		}
Steven Allen's avatar
Steven Allen committed
316
		return s, nil
Juan Batiz-Benet's avatar
Juan Batiz-Benet committed
317 318 319
	}
}

Steven Allen's avatar
Steven Allen committed
320 321 322 323 324 325 326 327 328 329 330 331 332 333 334 335 336 337 338 339 340 341 342 343 344 345 346 347 348 349 350 351
// ConnsToPeer returns all the live connections to peer.
func (s *Swarm) ConnsToPeer(p peer.ID) []inet.Conn {
	// TODO: Consider sorting the connection list best to worst. Currently,
	// it's sorted oldest to newest.
	s.conns.RLock()
	defer s.conns.RUnlock()
	conns := s.conns.m[p]
	output := make([]inet.Conn, len(conns))
	for i, c := range conns {
		output[i] = c
	}
	return output
}

// bestConnToPeer returns the best connection to peer.
func (s *Swarm) bestConnToPeer(p peer.ID) *Conn {
	// Selects the best connection we have to the peer.
	// TODO: Prefer some transports over others. Currently, we just select
	// the newest non-closed connection with the most streams.
	s.conns.RLock()
	defer s.conns.RUnlock()

	var best *Conn
	bestLen := 0
	for _, c := range s.conns.m[p] {
		if c.conn.IsClosed() {
			// We *will* garbage collect this soon anyways.
			continue
		}
		c.streams.Lock()
		cLen := len(c.streams.m)
		c.streams.Unlock()
Juan Batiz-Benet's avatar
Juan Batiz-Benet committed
352

Steven Allen's avatar
Steven Allen committed
353 354 355 356
		if cLen >= bestLen {
			best = c
			bestLen = cLen
		}
357

Steven Allen's avatar
Steven Allen committed
358 359
	}
	return best
Juan Batiz-Benet's avatar
Juan Batiz-Benet committed
360 361
}

Steven Allen's avatar
Steven Allen committed
362 363 364 365 366 367 368
// Connectedness returns our "connectedness" state with the given peer.
//
// To check if we have an open connection, use `s.Connectedness(p) ==
// inet.Connected`.
func (s *Swarm) Connectedness(p peer.ID) inet.Connectedness {
	if s.bestConnToPeer(p) != nil {
		return inet.Connected
Juan Batiz-Benet's avatar
Juan Batiz-Benet committed
369
	}
Steven Allen's avatar
Steven Allen committed
370
	return inet.NotConnected
Juan Batiz-Benet's avatar
Juan Batiz-Benet committed
371 372
}

Steven Allen's avatar
Steven Allen committed
373 374 375 376
// Conns returns a slice of all connections.
func (s *Swarm) Conns() []inet.Conn {
	s.conns.RLock()
	defer s.conns.RUnlock()
Juan Batiz-Benet's avatar
Juan Batiz-Benet committed
377

Steven Allen's avatar
Steven Allen committed
378 379 380 381 382 383 384 385 386 387 388 389 390 391 392 393 394 395 396 397 398 399 400
	conns := make([]inet.Conn, 0, len(s.conns.m))
	for _, cs := range s.conns.m {
		for _, c := range cs {
			conns = append(conns, c)
		}
	}
	return conns
}

// ClosePeer closes all connections to the given peer.
func (s *Swarm) ClosePeer(p peer.ID) error {
	conns := s.ConnsToPeer(p)
	switch len(conns) {
	case 0:
		return nil
	case 1:
		return conns[0].Close()
	default:
		errCh := make(chan error)
		for _, c := range conns {
			go func(c inet.Conn) {
				errCh <- c.Close()
			}(c)
Juan Batiz-Benet's avatar
Juan Batiz-Benet committed
401 402
		}

Steven Allen's avatar
Steven Allen committed
403 404 405 406 407 408 409 410 411 412 413 414 415 416 417 418 419 420 421 422
		var errs []string
		for _ = range conns {
			err := <-errCh
			if err != nil {
				errs = append(errs, err.Error())
			}
		}
		if len(errs) > 0 {
			return fmt.Errorf("when disconnecting from peer %s: %s", p, strings.Join(errs, ", "))
		}
		return nil
	}
}

// Peers returns a copy of the set of peers swarm is connected to.
func (s *Swarm) Peers() []peer.ID {
	s.conns.RLock()
	defer s.conns.RUnlock()
	peers := make([]peer.ID, 0, len(s.conns.m))
	for p := range s.conns.m {
Juan Batiz-Benet's avatar
Juan Batiz-Benet committed
423 424
		peers = append(peers, p)
	}
Steven Allen's avatar
Steven Allen committed
425

Juan Batiz-Benet's avatar
Juan Batiz-Benet committed
426 427 428 429 430 431 432 433
	return peers
}

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

Steven Allen's avatar
Steven Allen committed
434 435
// Backoff returns the DialBackoff object for this swarm.
func (s *Swarm) Backoff() *DialBackoff {
Jeromy's avatar
Jeromy committed
436 437 438
	return &s.backf
}

Juan Batiz-Benet's avatar
Juan Batiz-Benet committed
439 440
// notifyAll sends a signal to all Notifiees
func (s *Swarm) notifyAll(notify func(inet.Notifiee)) {
Steven Allen's avatar
Steven Allen committed
441 442 443 444 445 446 447 448 449
	var wg sync.WaitGroup

	s.notifs.RLock()
	wg.Add(len(s.notifs.m))
	for f := range s.notifs.m {
		go func(f inet.Notifiee) {
			defer wg.Done()
			notify(f)
		}(f)
Juan Batiz-Benet's avatar
Juan Batiz-Benet committed
450
	}
Steven Allen's avatar
Steven Allen committed
451 452 453

	wg.Wait()
	s.notifs.RUnlock()
Juan Batiz-Benet's avatar
Juan Batiz-Benet committed
454 455 456 457
}

// Notify signs up Notifiee to receive signals when events happen
func (s *Swarm) Notify(f inet.Notifiee) {
Steven Allen's avatar
Steven Allen committed
458 459 460
	s.notifs.Lock()
	s.notifs.m[f] = struct{}{}
	s.notifs.Unlock()
Juan Batiz-Benet's avatar
Juan Batiz-Benet committed
461 462 463 464
}

// StopNotify unregisters Notifiee fromr receiving signals
func (s *Swarm) StopNotify(f inet.Notifiee) {
Steven Allen's avatar
Steven Allen committed
465 466 467 468 469 470 471 472 473 474 475 476 477 478 479 480 481 482 483 484 485 486 487 488 489
	s.notifs.Lock()
	delete(s.notifs.m, f)
	s.notifs.Unlock()
}

func (s *Swarm) removeConn(c *Conn) {
	p := c.RemotePeer()

	s.conns.Lock()
	defer s.conns.Unlock()
	cs := s.conns.m[p]
	for i, ci := range cs {
		if ci == c {
			if len(cs) == 1 {
				delete(s.conns.m, p)
			} else {
				// NOTE: We're intentionally preserving order.
				// This way, connections to a peer are always
				// sorted oldest to newest.
				copy(cs[i:], cs[i+1:])
				cs[len(cs)-1] = nil
				s.conns.m[p] = cs[:len(cs)-1]
			}
			return
		}
Juan Batiz-Benet's avatar
Juan Batiz-Benet committed
490 491 492
	}
}

Steven Allen's avatar
Steven Allen committed
493 494 495
// String returns a string representation of Network.
func (s *Swarm) String() string {
	return fmt.Sprintf("<Swarm %s>", s.LocalPeer())
Juan Batiz-Benet's avatar
Juan Batiz-Benet committed
496 497
}

Steven Allen's avatar
Steven Allen committed
498 499 500
// Swarm is a Network.
var _ inet.Network = (*Swarm)(nil)
var _ transport.Network = (*Swarm)(nil)