diff --git a/router/connection.go b/router/connection.go index 9cfaac86..a79abca8 100644 --- a/router/connection.go +++ b/router/connection.go @@ -49,23 +49,25 @@ func (conn *RemoteConnection) String() string { return fmt.Sprint("Connection ", from, "->", to) } -// Async. Does not return anything. If the connection is successful, -// it will end up in the local peer's connections map. -func NewLocalConnection(connRemote *RemoteConnection, acceptNewPeer bool, tcpConn *net.TCPConn, udpAddr *net.UDPAddr, router *Router) { +func NewLocalConnection(connRemote *RemoteConnection, tcpConn *net.TCPConn, udpAddr *net.UDPAddr, router *Router) *LocalConnection { if connRemote.local != router.Ourself.Peer { log.Fatal("Attempt to create local connection from a peer which is not ourself") } - - queryChan := make(chan *ConnectionInteraction, ChannelSize) // NB, we're taking a copy of connRemote here. - connLocal := &LocalConnection{ + return &LocalConnection{ RemoteConnection: *connRemote, Router: router, TCPConn: tcpConn, remoteUDPAddr: udpAddr, - effectivePMTU: DefaultPMTU, - queryChan: queryChan} - go connLocal.run(queryChan, acceptNewPeer) + 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) { + queryChan := make(chan *ConnectionInteraction, ChannelSize) + conn.queryChan = queryChan + go conn.run(queryChan, acceptNewPeer) } func (conn *LocalConnection) Established() bool { diff --git a/router/connection_maker.go b/router/connection_maker.go index 191d7dae..eb261458 100644 --- a/router/connection_maker.go +++ b/router/connection_maker.go @@ -41,16 +41,18 @@ type ConnectionMakerInteraction struct { address string } -func StartConnectionMaker(ourself *LocalPeer, peers *Peers) *ConnectionMaker { - queryChan := make(chan *ConnectionMakerInteraction, ChannelSize) - state := &ConnectionMaker{ +func NewConnectionMaker(ourself *LocalPeer, peers *Peers) *ConnectionMaker { + return &ConnectionMaker{ ourself: ourself, peers: peers, - queryChan: queryChan, cmdLineAddress: make(map[string]bool), targets: make(map[string]*Target)} - go state.queryLoop(queryChan) - return state +} + +func (cm *ConnectionMaker) Start() { + queryChan := make(chan *ConnectionMakerInteraction, ChannelSize) + cm.queryChan = queryChan + go cm.queryLoop(queryChan) } func (cm *ConnectionMaker) InitiateConnection(address string) { diff --git a/router/forwarder.go b/router/forwarder.go index 94eed9f7..fafb5115 100644 --- a/router/forwarder.go +++ b/router/forwarder.go @@ -50,8 +50,8 @@ func (conn *LocalConnection) ensureForwarders() error { conn.effectivePMTU = forwarder.unverifiedPMTU conn.Unlock() - go forwarder.Run() - go forwarderDF.Run() + forwarder.Start() + forwarderDF.Start() return nil } @@ -204,7 +204,11 @@ func NewForwarder(conn *LocalConnection, ch <-chan *ForwardedFrame, stop <-chan return fwd } -func (fwd *Forwarder) Run() { +func (fwd *Forwarder) Start() { + go fwd.run() +} + +func (fwd *Forwarder) run() { defer fwd.udpSender.Shutdown() var flushed, ok bool var frame *ForwardedFrame diff --git a/router/local_peer.go b/router/local_peer.go index 5bebd85c..b88f672d 100644 --- a/router/local_peer.go +++ b/router/local_peer.go @@ -17,12 +17,14 @@ type PeerInteraction struct { payload interface{} } -func StartLocalPeer(name PeerName, router *Router) *LocalPeer { - peer := &LocalPeer{Peer: NewPeer(name, 0, 0), Router: router} +func NewLocalPeer(name PeerName, router *Router) *LocalPeer { + return &LocalPeer{Peer: NewPeer(name, 0, 0), Router: router} +} + +func (peer *LocalPeer) Start() { queryChan := make(chan *PeerInteraction, ChannelSize) peer.queryChan = queryChan go peer.queryLoop(queryChan) - return peer } func (peer *LocalPeer) Forward(dstPeer *Peer, df bool, frame []byte, dec *EthernetDecoder) error { @@ -106,7 +108,8 @@ func (peer *LocalPeer) CreateConnection(peerAddr string, acceptNewPeer bool) err return err } connRemote := NewRemoteConnection(peer.Peer, nil, tcpConn.RemoteAddr().String()) - NewLocalConnection(connRemote, acceptNewPeer, tcpConn, udpAddr, peer.Router) + connLocal := NewLocalConnection(connRemote, tcpConn, udpAddr, peer.Router) + connLocal.Start(acceptNewPeer) return nil } diff --git a/router/router.go b/router/router.go index fa0ca37e..0d361bfd 100644 --- a/router/router.go +++ b/router/router.go @@ -33,31 +33,32 @@ func NewRouter(iface *net.Interface, name PeerName, password []byte, connLimit i if len(password) > 0 { router.Password = &password } - router.Ourself = StartLocalPeer(name, router) + router.Ourself = NewLocalPeer(name, router) router.Peers = NewPeers(router.Ourself.Peer, router.Macs, onPeerGC) router.Peers.FetchWithDefault(router.Ourself.Peer) - log.Println("Our name is", router.Ourself.Name) - + router.Routes = NewRoutes(router.Ourself.Peer, router.Peers) + router.ConnectionMaker = NewConnectionMaker(router.Ourself, router.Peers) return router } -func (router *Router) UsingPassword() bool { - return router.Password != nil -} - func (router *Router) Start() { // we need two pcap handles since they aren't thread-safe pio, err := NewPcapIO(router.Iface.Name, router.BufSz) checkFatal(err) po, err := NewPcapO(router.Iface.Name) checkFatal(err) - router.Routes = StartRoutes(router.Ourself.Peer, router.Peers) - router.ConnectionMaker = StartConnectionMaker(router.Ourself, router.Peers) + router.Ourself.Start() + router.Routes.Start() + router.ConnectionMaker.Start() router.UDPListener = router.listenUDP(Port, po) router.listenTCP(Port) router.sniff(pio) } +func (router *Router) UsingPassword() bool { + return router.Password != nil +} + func (router *Router) Status() string { var buf bytes.Buffer buf.WriteString(fmt.Sprintln("Our name is", router.Ourself.Name)) @@ -159,7 +160,8 @@ 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) - NewLocalConnection(connRemote, true, tcpConn, nil, router) + connLocal := NewLocalConnection(connRemote, tcpConn, nil, router) + connLocal.Start(true) } func (router *Router) listenUDP(localPort int, po PacketSink) *net.UDPConn { diff --git a/router/routes.go b/router/routes.go index 4e46ec6a..f8abac18 100644 --- a/router/routes.go +++ b/router/routes.go @@ -16,6 +16,23 @@ type Routes struct { queryChan chan<- *Interaction } +func NewRoutes(ourself *Peer, peers *Peers) *Routes { + routes := &Routes{ + ourself: ourself, + peers: peers, + unicast: make(map[PeerName]PeerName), + broadcast: make(map[PeerName][]PeerName)} + routes.unicast[ourself.Name] = UnknownPeerName + routes.broadcast[ourself.Name] = []PeerName{} + return routes +} + +func (routes *Routes) Start() { + queryChan := make(chan *Interaction, ChannelSize) + routes.queryChan = queryChan + go routes.queryLoop(queryChan) +} + func (routes *Routes) Unicast(name PeerName) (PeerName, bool) { routes.RLock() defer routes.RUnlock() @@ -48,20 +65,6 @@ func (routes *Routes) String() string { return buf.String() } -func StartRoutes(ourself *Peer, peers *Peers) *Routes { - queryChan := make(chan *Interaction, ChannelSize) - state := &Routes{ - ourself: ourself, - peers: peers, - unicast: make(map[PeerName]PeerName), - broadcast: make(map[PeerName][]PeerName), - queryChan: queryChan} - state.unicast[ourself.Name] = UnknownPeerName - state.broadcast[ourself.Name] = []PeerName{} - go state.queryLoop(queryChan) - return state -} - // ACTOR client API const ( diff --git a/weaver/main.go b/weaver/main.go index 5d52fbec..64f44a3b 100644 --- a/weaver/main.go +++ b/weaver/main.go @@ -122,6 +122,7 @@ func main() { } router := weave.NewRouter(iface, ourName, []byte(password), connLimit, bufSz*1024*1024, logFrame) + log.Println("Our name is", router.Ourself.Name) router.Start() for _, peer := range peers { if addr, err := net.ResolveTCPAddr("tcp4", weave.NormalisePeerAddr(peer)); err == nil {