diff --git a/router/connection.go b/router/connection.go index adf9e04c..3d15c269 100644 --- a/router/connection.go +++ b/router/connection.go @@ -98,26 +98,23 @@ func (conn *RemoteConnection) String() string { return fmt.Sprint("Connection ", from, "->", to) } -func NewLocalConnection(connRemote *RemoteConnection, tcpConn *net.TCPConn, udpAddr *net.UDPAddr, router *Router) *LocalConnection { +// Does not return anything. If the connection is successful, it will +// end up in the local peer's connections map. +func StartLocalConnection(connRemote *RemoteConnection, tcpConn *net.TCPConn, udpAddr *net.UDPAddr, router *Router, acceptNewPeer bool) { if connRemote.local != router.Ourself.Peer { log.Fatal("Attempt to create local connection from a peer which is not ourself") } // NB, we're taking a copy of connRemote here. - return &LocalConnection{ + actionChan := make(chan ConnectionAction, ChannelSize) + finished := make(chan struct{}) + conn := &LocalConnection{ RemoteConnection: *connRemote, Router: router, TCPConn: tcpConn, remoteUDPAddr: udpAddr, - effectivePMTU: DefaultPMTU} -} - -// Async. Does not return anything. If the connection is successful, -// it will end up in the local peer's connections map. -func (conn *LocalConnection) Start(acceptNewPeer bool) { - actionChan := make(chan ConnectionAction, ChannelSize) - conn.actionChan = actionChan - finished := make(chan struct{}) - conn.finished = finished + effectivePMTU: DefaultPMTU, + actionChan: actionChan, + finished: finished} go conn.run(actionChan, finished, acceptNewPeer) } diff --git a/router/connection_maker.go b/router/connection_maker.go index 57a93a06..2e36b69f 100644 --- a/router/connection_maker.go +++ b/router/connection_maker.go @@ -37,19 +37,17 @@ type Target struct { type ConnectionMakerAction func() bool func NewConnectionMaker(ourself *LocalPeer, peers *Peers, port int, discovery bool) *ConnectionMaker { - return &ConnectionMaker{ + actionChan := make(chan ConnectionMakerAction, ChannelSize) + cm := &ConnectionMaker{ ourself: ourself, peers: peers, port: port, discovery: discovery, directPeers: peerAddrs{}, - targets: make(map[string]*Target)} -} - -func (cm *ConnectionMaker) Start() { - actionChan := make(chan ConnectionMakerAction, ChannelSize) - cm.actionChan = actionChan + targets: make(map[string]*Target), + actionChan: actionChan} go cm.queryLoop(actionChan) + return cm } func (cm *ConnectionMaker) InitiateConnections(peers []string, replace bool) []error { diff --git a/router/forwarder.go b/router/forwarder.go index 802369bd..3c9ba0ed 100644 --- a/router/forwarder.go +++ b/router/forwarder.go @@ -40,16 +40,12 @@ func (conn *LocalConnection) ensureForwarders() error { forwarder := NewForwarder(conn, encryptor, udpSender, DefaultPMTU) forwarderDF := NewForwarderDF(conn, encryptorDF, udpSenderDF, DefaultPMTU) - effectivePMTU := forwarderDF.unverifiedPMTU - forwarder.Start() - forwarderDF.Start() // Various fields in the conn struct are read by other processes, // so we have to use locks. conn.Lock() conn.forwarder = forwarder conn.forwarderDF = forwarderDF - conn.effectivePMTU = effectivePMTU conn.Unlock() return nil @@ -195,20 +191,18 @@ type Forwarder struct { } func NewForwarder(conn *LocalConnection, enc Encryptor, udpSender UDPSender, pmtu int) *Forwarder { - return &Forwarder{ + ch := make(chan *ForwardedFrame, ChannelSize) + finished := make(chan struct{}) + fwd := &Forwarder{ conn: conn, + ch: ch, + finished: finished, enc: enc, udpSender: udpSender, maxPayload: pmtu - UDPOverhead, processSendError: func(err error) error { return err }} -} - -func (fwd *Forwarder) Start() { - ch := make(chan *ForwardedFrame, ChannelSize) - fwd.ch = ch - finished := make(chan struct{}) - fwd.finished = finished go fwd.run(ch, finished) + return fwd } func (fwd *Forwarder) Shutdown() { @@ -310,25 +304,23 @@ type ForwarderDF struct { } func NewForwarderDF(conn *LocalConnection, enc Encryptor, udpSender UDPSender, pmtu int) *ForwarderDF { + ch := make(chan *ForwardedFrame, ChannelSize) + finished := make(chan struct{}) + verifyPMTU := make(chan int, ChannelSize) fwd := &ForwarderDF{ Forwarder: Forwarder{ conn: conn, + ch: ch, + finished: finished, enc: enc, udpSender: udpSender, - maxPayload: pmtu - UDPOverhead}} + maxPayload: pmtu - UDPOverhead}, + verifyPMTU: verifyPMTU} fwd.Forwarder.processSendError = fwd.processSendError fwd.unverifiedPMTU = pmtu - fwd.effectiveOverhead() - return fwd -} - -func (fwd *ForwarderDF) Start() { - ch := make(chan *ForwardedFrame, ChannelSize) - fwd.ch = ch - finished := make(chan struct{}) - fwd.finished = finished - verifyPMTU := make(chan int, ChannelSize) - fwd.verifyPMTU = verifyPMTU + conn.setEffectivePMTU(fwd.unverifiedPMTU) go fwd.run(ch, finished, verifyPMTU) + return fwd } func (fwd *ForwarderDF) PMTUVerified(pmtu int) { diff --git a/router/gossip.go b/router/gossip.go index f66ea9d0..a3af241b 100644 --- a/router/gossip.go +++ b/router/gossip.go @@ -40,13 +40,15 @@ type GossipSender struct { } func NewGossipSender(send func(GossipData)) *GossipSender { - return &GossipSender{send: send} -} - -func (sender *GossipSender) Start() { - sender.cell = make(chan GossipData, 1) - sender.flushch = make(chan chan bool) + cell := make(chan GossipData, 1) + flushch := make(chan chan bool) + sender := &GossipSender{ + send: send, + cell: cell, + flushch: flushch, + } go sender.run() + return sender } func (sender *GossipSender) run() { diff --git a/router/gossip_channel.go b/router/gossip_channel.go index da7dc0ec..8dc1f510 100644 --- a/router/gossip_channel.go +++ b/router/gossip_channel.go @@ -147,7 +147,6 @@ func (c *GossipChannel) sendDown(conn Connection, data GossipData) { } }) c.senders[conn] = sender - sender.Start() } sender.Send(data) } @@ -198,7 +197,6 @@ func (c *GossipChannel) relayBroadcast(srcName PeerName, update GossipData) erro if !found { broadcaster = NewGossipSender(func(pending GossipData) { c.sendBroadcast(srcName, pending) }) c.broadcasters[srcName] = broadcaster - broadcaster.Start() } broadcaster.Send(update) return nil diff --git a/router/gossip_test.go b/router/gossip_test.go index c1ace338..c6936a09 100644 --- a/router/gossip_test.go +++ b/router/gossip_test.go @@ -18,15 +18,10 @@ type mockChannelConnection struct { dest *Router } -// Construct a "passive" Router, i.e. without any goroutines, except -// for Routes and GossipSenders. func NewTestRouter(name string) *Router { peerName, _ := PeerNameFromString(name) - router := NewRouter(Config{}, peerName, "") - // need to create a dummy channel otherwise tests hang on nil - // channels when the Router invoked ConnectionMaker.Refresh - router.ConnectionMaker.actionChan = make(chan ConnectionMakerAction, ChannelSize) - router.Routes.Start() + router := NewRouter(Config{}, peerName, "nick") + router.Start() return router } diff --git a/router/local_peer.go b/router/local_peer.go index 09498629..b9ea939b 100644 --- a/router/local_peer.go +++ b/router/local_peer.go @@ -18,13 +18,14 @@ type LocalPeer struct { type LocalPeerAction func() func NewLocalPeer(name PeerName, nickName string, router *Router) *LocalPeer { - return &LocalPeer{Peer: NewPeer(name, nickName, 0, 0), router: router} -} - -func (peer *LocalPeer) Start() { actionChan := make(chan LocalPeerAction, ChannelSize) - peer.actionChan = actionChan + peer := &LocalPeer{ + Peer: NewPeer(name, nickName, 0, 0), + router: router, + actionChan: actionChan, + } go peer.actorLoop(actionChan) + return peer } func (peer *LocalPeer) Forward(dstPeer *Peer, df bool, frame []byte, dec *EthernetDecoder) error { @@ -132,8 +133,7 @@ func (peer *LocalPeer) CreateConnection(peerAddr string, acceptNewPeer bool) err return err } connRemote := NewRemoteConnection(peer.Peer, nil, tcpConn.RemoteAddr().String(), true, false) - connLocal := NewLocalConnection(connRemote, tcpConn, udpAddr, peer.router) - connLocal.Start(acceptNewPeer) + StartLocalConnection(connRemote, tcpConn, udpAddr, peer.router, acceptNewPeer) return nil } diff --git a/router/mac_cache.go b/router/mac_cache.go index d251d194..5eb03f41 100644 --- a/router/mac_cache.go +++ b/router/mac_cache.go @@ -22,14 +22,12 @@ type MacCache struct { } func NewMacCache(maxAge time.Duration, onExpiry func(net.HardwareAddr, *Peer)) *MacCache { - return &MacCache{ + cache := &MacCache{ table: make(map[uint64]*MacCacheEntry), maxAge: maxAge, onExpiry: onExpiry} -} - -func (cache *MacCache) Start() { cache.setExpiryTimer() + return cache } func (cache *MacCache) Enter(mac net.HardwareAddr, peer *Peer) bool { diff --git a/router/router.go b/router/router.go index ab671a0d..9f9e626f 100644 --- a/router/router.go +++ b/router/router.go @@ -84,6 +84,9 @@ func NewRouter(config Config, name PeerName, nickName string) *Router { return router } +// Start listening for TCP connections, locally captured packets, and +// packets forwarded over UDP. This is separate from NewRouter so +// that gossipers can register before we start forming connections. func (router *Router) Start() { // we need two pcap handles since they aren't thread-safe var pio PacketSourceSink @@ -95,10 +98,6 @@ func (router *Router) Start() { po, err = NewPcapO(router.Iface.Name) checkFatal(err) } - router.Ourself.Start() - router.Macs.Start() - router.Routes.Start() - router.ConnectionMaker.Start() router.UDPListener = router.listenUDP(router.Port, po) router.listenTCP(router.Port) if pio != nil { @@ -224,8 +223,7 @@ func (router *Router) acceptTCP(tcpConn *net.TCPConn) { remoteAddrStr := tcpConn.RemoteAddr().String() log.Printf("->[%s] connection accepted\n", remoteAddrStr) connRemote := NewRemoteConnection(router.Ourself.Peer, nil, remoteAddrStr, false, false) - connLocal := NewLocalConnection(connRemote, tcpConn, nil, router) - connLocal.Start(true) + StartLocalConnection(connRemote, tcpConn, nil, router, true) } func (router *Router) listenUDP(localPort int, po PacketSink) *net.UDPConn { diff --git a/router/routes.go b/router/routes.go index f244f28f..1f73833d 100644 --- a/router/routes.go +++ b/router/routes.go @@ -22,26 +22,23 @@ type Routes struct { } func NewRoutes(ourself *LocalPeer, peers *Peers) *Routes { + recalculate := make(chan *struct{}, 1) + wait := make(chan chan struct{}) routes := &Routes{ ourself: ourself, peers: peers, unicast: make(map[PeerName]PeerName), unicastAll: make(map[PeerName]PeerName), broadcast: make(map[PeerName][]PeerName), - broadcastAll: make(map[PeerName][]PeerName)} + broadcastAll: make(map[PeerName][]PeerName), + recalculate: recalculate, + wait: wait} routes.unicast[ourself.Name] = UnknownPeerName routes.unicastAll[ourself.Name] = UnknownPeerName routes.broadcast[ourself.Name] = []PeerName{} routes.broadcastAll[ourself.Name] = []PeerName{} - return routes -} - -func (routes *Routes) Start() { - recalculate := make(chan *struct{}, 1) - wait := make(chan chan struct{}) - routes.recalculate = recalculate - routes.wait = wait go routes.run(recalculate, wait) + return routes } func (routes *Routes) PeerNames() PeerNameSet {