diff --git a/nameserver/mdns_client.go b/nameserver/mdns_client.go index 59f89f16..cb53c490 100644 --- a/nameserver/mdns_client.go +++ b/nameserver/mdns_client.go @@ -50,11 +50,11 @@ type inflightQuery struct { type MDNSClient struct { // note unorthodox use of 'Server' class in client logic - using it to // listen on a multicast socket and call callback whenever a message comes in. - listener *dns.Server - conn *net.UDPConn - addr *net.UDPAddr - inflight map[string]*inflightQuery - queryChan chan<- *MDNSInteraction + listener *dns.Server + conn *net.UDPConn + addr *net.UDPAddr + inflight map[string]*inflightQuery + actionChan chan<- MDNSAction } type mDNSQueryInfo struct { @@ -90,9 +90,9 @@ func (c *MDNSClient) Start(ifi *net.Interface) error { c.listener = &dns.Server{Unsafe: true, PacketConn: multicast, Handler: dns.HandlerFunc(handleMDNS)} go c.listener.ActivateAndServe() - queryChan := make(chan *MDNSInteraction, MailboxSize) - c.queryChan = queryChan - go c.queryLoop(queryChan) + actionChan := make(chan MDNSAction, MailboxSize) + c.actionChan = actionChan + go c.actorLoop(actionChan) return nil } @@ -103,34 +103,79 @@ func LinkLocalMulticastListener(ifi *net.Interface) (net.PacketConn, error) { // ACTOR client API -const ( - CSendQuery = iota - CShutdown - CMessageReceived -) - -type MDNSInteraction struct { - code int - resultChan chan<- interface{} - payload interface{} -} +type MDNSAction func() // Async func (c *MDNSClient) Shutdown() { - c.queryChan <- &MDNSInteraction{code: CShutdown} + c.actionChan <- nil } // Async func (c *MDNSClient) SendQuery(name string, querytype uint16, responseCh chan<- *Response) { - c.queryChan <- &MDNSInteraction{ - code: CSendQuery, - payload: mDNSQueryInfo{name, querytype, responseCh}, + c.actionChan <- func() { + query, found := c.inflight[name] + if !found { + m := new(dns.Msg) + m.SetQuestion(name, querytype) + m.RecursionDesired = false + buf, err := m.Pack() + if err != nil { + responseCh <- &Response{Err: err} + close(responseCh) + return + } + query = &inflightQuery{ + name: name, + id: m.Id, + } + if _, err = c.conn.WriteTo(buf, c.addr); err != nil { + responseCh <- &Response{Err: err} + close(responseCh) + return + } + c.inflight[name] = query + } + info := &responseInfo{ + ch: responseCh, + timeout: time.Now().Add(mDNSTimeout), + } + // Invariant on responseInfos: they are in ascending order of + // timeout. Since we use a fixed interval from Now(), this + // must be after all existing timeouts. + query.responseInfos = append(query.responseInfos, info) } } // Async - called from dns library multiplexer func (c *MDNSClient) ResponseCallback(r *dns.Msg) { - c.queryChan <- &MDNSInteraction{code: CMessageReceived, payload: r} + c.actionChan <- func() { + for _, answer := range r.Answer { + var name string + var res *Response + + switch rr := answer.(type) { + case *dns.A: + name = rr.Hdr.Name + res = &Response{Addr: rr.A} + case *dns.PTR: + name = rr.Hdr.Name + res = &Response{Name: rr.Ptr} + default: + return + } + + if query, found := c.inflight[name]; found { + for _, resp := range query.responseInfos { + resp.ch <- res + close(resp.ch) + } + delete(c.inflight, name) + } else { + // We've received a response that didn't match a query + // Do we want to cache it? + } + } + } } // ACTOR server @@ -164,25 +209,18 @@ func (c *MDNSClient) checkInFlightQueries() time.Duration { return after } -func (c *MDNSClient) queryLoop(queryChan <-chan *MDNSInteraction) { +func (c *MDNSClient) actorLoop(actionChan <-chan MDNSAction) { timer := time.NewTimer(MaxDuration) run := func() { timer.Reset(c.checkInFlightQueries()) } terminate := false for !terminate { select { - case query, ok := <-queryChan: - if !ok { - break - } - switch query.code { - case CShutdown: + case action := <-actionChan: + if action == nil { c.listener.Shutdown() terminate = true - case CSendQuery: - c.handleSendQuery(query.payload.(mDNSQueryInfo)) - run() - case CMessageReceived: - c.handleResponse(query.payload.(*dns.Msg)) + } else { + action() run() } case <-timer.C: @@ -197,64 +235,3 @@ func (c *MDNSClient) queryLoop(queryChan <-chan *MDNSInteraction) { } } } - -func (c *MDNSClient) handleSendQuery(q mDNSQueryInfo) { - query, found := c.inflight[q.name] - if !found { - m := new(dns.Msg) - m.SetQuestion(q.name, q.querytype) - m.RecursionDesired = false - buf, err := m.Pack() - if err != nil { - q.responseCh <- &Response{Err: err} - close(q.responseCh) - return - } - query = &inflightQuery{ - name: q.name, - id: m.Id, - } - if _, err = c.conn.WriteTo(buf, c.addr); err != nil { - q.responseCh <- &Response{Err: err} - close(q.responseCh) - return - } - c.inflight[q.name] = query - } - info := &responseInfo{ - ch: q.responseCh, - timeout: time.Now().Add(mDNSTimeout), - } - // Invariant on responseInfos: they are in ascending order of timeout. - // Since we use a fixed interval from Now(), this must be after all existing timeouts. - query.responseInfos = append(query.responseInfos, info) -} - -func (c *MDNSClient) handleResponse(r *dns.Msg) { - for _, answer := range r.Answer { - var name string - var res *Response - - switch rr := answer.(type) { - case *dns.A: - name = rr.Hdr.Name - res = &Response{Addr: rr.A} - case *dns.PTR: - name = rr.Hdr.Name - res = &Response{Name: rr.Ptr} - default: - return - } - - if query, found := c.inflight[name]; found { - for _, resp := range query.responseInfos { - resp.ch <- res - close(resp.ch) - } - delete(c.inflight, name) - } else { - // We've received a response that didn't match a query - // Do we want to cache it? - } - } -} diff --git a/router/connection.go b/router/connection.go index 4d72a286..f3a88c2a 100644 --- a/router/connection.go +++ b/router/connection.go @@ -59,14 +59,11 @@ type LocalConnection struct { Decryptor Decryptor Router *Router uid uint64 - queryChan chan<- *ConnectionInteraction - finished <-chan struct{} // closed to signal that queryLoop has finished + actionChan chan<- ConnectionAction + finished <-chan struct{} // closed to signal that actorLoop has finished } -type ConnectionInteraction struct { - Interaction - payload interface{} -} +type ConnectionAction func() error func NewRemoteConnection(from, to *Peer, tcpAddr string, outbound bool, established bool) *RemoteConnection { return &RemoteConnection{ @@ -118,11 +115,11 @@ func NewLocalConnection(connRemote *RemoteConnection, tcpConn *net.TCPConn, udpA // 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 + actionChan := make(chan ConnectionAction, ChannelSize) + conn.actionChan = actionChan finished := make(chan struct{}) conn.finished = finished - go conn.run(queryChan, finished, acceptNewPeer) + go conn.run(actionChan, finished, acceptNewPeer) } func (conn *LocalConnection) BreakTie(dupConn Connection) ConnectionTieBreak { @@ -182,29 +179,21 @@ func (conn *LocalConnection) SendProtocolMsg(m ProtocolMsg) { } } -// ACTOR client API +// ACTOR methods -const ( - CSetEstablished = iota - CReceivedHeartbeat - CShutdown -) - -// Send an actor request to the queryLoop, but don't block if queryLoop has exited -// - see http://blog.golang.org/pipelines for pattern -func (conn *LocalConnection) sendQuery(code int, payload interface{}) { - select { - case conn.queryChan <- &ConnectionInteraction{ - Interaction: Interaction{code: code}, - payload: payload}: - case <-conn.finished: - } -} +// NB: The conn.* fields are only written by the connection actor +// process, which is the caller of the ConnectionAction funs. Hence we +// do not need locks for reading, and only need write locks for fields +// read by other processes. // Async func (conn *LocalConnection) Shutdown(err error) { - // Run on its own goroutine in case the channel is backed up - go conn.sendQuery(CShutdown, err) + if err == nil { + conn.actionChan <- nil + } else { + // Run on its own goroutine in case the channel is backed up + go func() { conn.sendAction(func() error { return err }) }() + } } // Async @@ -216,18 +205,76 @@ func (conn *LocalConnection) ReceivedHeartbeat(remoteUDPAddr *net.UDPAddr, connU if remoteUDPAddr == nil || connUID != conn.uid { return } - conn.sendQuery(CReceivedHeartbeat, remoteUDPAddr) + conn.sendAction(func() error { + oldRemoteUDPAddr := conn.remoteUDPAddr + old := conn.receivedHeartbeat + conn.Lock() + conn.remoteUDPAddr = remoteUDPAddr + conn.receivedHeartbeat = true + conn.Unlock() + conn.heartbeatTimeout.Reset(HeartbeatTimeout) + if !old { + if err := conn.sendSimpleProtocolMsg(ProtocolConnectionEstablished); err != nil { + return err + } + } + if oldRemoteUDPAddr == nil { + return conn.sendFastHeartbeats() + } else if oldRemoteUDPAddr.String() != remoteUDPAddr.String() { + log.Println("Peer", conn.remote.Name, "moved from", old, "to", remoteUDPAddr) + } + return nil + }) } // Async func (conn *LocalConnection) SetEstablished() { - conn.sendQuery(CSetEstablished, nil) + conn.sendAction(func() error { + stopTicker(conn.heartbeat) + old := conn.established + conn.Lock() + conn.established = true + conn.Unlock() + if old { + return nil + } + conn.Router.Ourself.ConnectionEstablished(conn) + if err := conn.ensureForwarders(); err != nil { + return err + } + // Send a large frame down the DF channel in order to prompt + // PMTU discovery to start. + conn.Forward(true, &ForwardedFrame{ + srcPeer: conn.local, + dstPeer: conn.remote, + frame: PMTUDiscovery}, + nil) + conn.heartbeat = time.NewTicker(SlowHeartbeat) + conn.fragTest = time.NewTicker(FragTestInterval) + // avoid initial waits for timers to fire + conn.Forward(true, conn.heartbeatFrame, nil) + conn.setStackFrag(false) + if err := conn.sendSimpleProtocolMsg(ProtocolStartFragmentationTest); err != nil { + return err + } + return nil + }) +} + +// Send an actor request to the actorLoop, but don't block if +// actorLoop has exited - see http://blog.golang.org/pipelines for +// pattern +func (conn *LocalConnection) sendAction(action ConnectionAction) { + select { + case conn.actionChan <- action: + case <-conn.finished: + } } // ACTOR server -func (conn *LocalConnection) run(queryChan <-chan *ConnectionInteraction, finished chan<- struct{}, acceptNewPeer bool) { - defer conn.handleShutdown() +func (conn *LocalConnection) run(actionChan <-chan ConnectionAction, finished chan<- struct{}, acceptNewPeer bool) { + defer conn.shutdown() defer close(finished) tcpConn := conn.TCPConn @@ -268,7 +315,7 @@ func (conn *LocalConnection) run(queryChan <-chan *ConnectionInteraction, finish conn.heartbeatTimeout = time.NewTimer(HeartbeatTimeout) - if err := conn.queryLoop(queryChan); err != nil { + if err := conn.actorLoop(actionChan); err != nil { conn.Log("connection shutting down due to error:", err) } else { conn.Log("connection shutting down") @@ -286,23 +333,14 @@ func (conn *LocalConnection) run(queryChan <-chan *ConnectionInteraction, finish // AddConnection, which is *synchronous*, before entering the tcp // receive loop. -func (conn *LocalConnection) queryLoop(queryChan <-chan *ConnectionInteraction) (err error) { - terminate := false - for !terminate && err == nil { +func (conn *LocalConnection) actorLoop(actionChan <-chan ConnectionAction) (err error) { + for err == nil { select { - case query, ok := <-queryChan: - if !ok { + case action := <-actionChan: + if action == nil { break } - switch query.code { - case CShutdown: - err = query.payload.(error) - terminate = true - case CReceivedHeartbeat: - err = conn.handleReceivedHeartbeat(query.payload.(*net.UDPAddr)) - case CSetEstablished: - err = conn.handleSetEstablished() - } + err = action() case <-conn.heartbeatTimeout.C: err = fmt.Errorf("timed out waiting for UDP heartbeat") case <-tickerChan(conn.heartbeat): @@ -315,66 +353,7 @@ func (conn *LocalConnection) queryLoop(queryChan <-chan *ConnectionInteraction) return } -// Handlers -// -// NB: The conn.* fields are only written by the connection actor -// process, which is the caller of the handlers. Hence we do not need -// locks for reading, and only need write locks for fields read by -// other processes. - -func (conn *LocalConnection) handleReceivedHeartbeat(remoteUDPAddr *net.UDPAddr) error { - oldRemoteUDPAddr := conn.remoteUDPAddr - old := conn.receivedHeartbeat - conn.Lock() - conn.remoteUDPAddr = remoteUDPAddr - conn.receivedHeartbeat = true - conn.Unlock() - conn.heartbeatTimeout.Reset(HeartbeatTimeout) - if !old { - if err := conn.sendSimpleProtocolMsg(ProtocolConnectionEstablished); err != nil { - return err - } - } - if oldRemoteUDPAddr == nil { - return conn.sendFastHeartbeats() - } else if oldRemoteUDPAddr.String() != remoteUDPAddr.String() { - log.Println("Peer", conn.remote.Name, "moved from", old, "to", remoteUDPAddr) - } - return nil -} - -func (conn *LocalConnection) handleSetEstablished() error { - stopTicker(conn.heartbeat) - old := conn.established - conn.Lock() - conn.established = true - conn.Unlock() - if old { - return nil - } - conn.Router.Ourself.ConnectionEstablished(conn) - if err := conn.ensureForwarders(); err != nil { - return err - } - // Send a large frame down the DF channel in order to prompt - // PMTU discovery to start. - conn.Forward(true, &ForwardedFrame{ - srcPeer: conn.local, - dstPeer: conn.remote, - frame: PMTUDiscovery}, - nil) - conn.heartbeat = time.NewTicker(SlowHeartbeat) - conn.fragTest = time.NewTicker(FragTestInterval) - // avoid initial waits for timers to fire - conn.Forward(true, conn.heartbeatFrame, nil) - conn.setStackFrag(false) - if err := conn.sendSimpleProtocolMsg(ProtocolStartFragmentationTest); err != nil { - return err - } - return nil -} - -func (conn *LocalConnection) handleShutdown() { +func (conn *LocalConnection) shutdown() { if conn.TCPConn != nil { checkWarn(conn.TCPConn.Close()) } diff --git a/router/connection_maker.go b/router/connection_maker.go index aff69115..9ca73945 100644 --- a/router/connection_maker.go +++ b/router/connection_maker.go @@ -14,19 +14,12 @@ const ( MaxInterval = 10 * time.Minute ) -const ( - CMInitiate = iota - CMTerminated - CMRefresh - CMStatus -) - type ConnectionMaker struct { ourself *LocalPeer peers *Peers targets map[string]*Target cmdLineAddress map[string]bool - queryChan chan<- *ConnectionMakerInteraction + actionChan chan<- ConnectionMakerAction } // Information about an address where we may find a peer @@ -36,10 +29,7 @@ type Target struct { tryInterval time.Duration // backoff time on next failure } -type ConnectionMakerInteraction struct { - Interaction - address string -} +type ConnectionMakerAction func() bool func NewConnectionMaker(ourself *LocalPeer, peers *Peers) *ConnectionMaker { return &ConnectionMaker{ @@ -50,61 +40,59 @@ func NewConnectionMaker(ourself *LocalPeer, peers *Peers) *ConnectionMaker { } func (cm *ConnectionMaker) Start() { - queryChan := make(chan *ConnectionMakerInteraction, ChannelSize) - cm.queryChan = queryChan - go cm.queryLoop(queryChan) + actionChan := make(chan ConnectionMakerAction, ChannelSize) + cm.actionChan = actionChan + go cm.queryLoop(actionChan) } func (cm *ConnectionMaker) InitiateConnection(address string) { - cm.queryChan <- &ConnectionMakerInteraction{ - Interaction: Interaction{code: CMInitiate}, - address: address} + cm.actionChan <- func() bool { + cm.cmdLineAddress[NormalisePeerAddr(address)] = true + return true + } } func (cm *ConnectionMaker) ConnectionTerminated(address string) { - cm.queryChan <- &ConnectionMakerInteraction{ - Interaction: Interaction{code: CMTerminated}, - address: address} + cm.actionChan <- func() bool { + if target, found := cm.targets[address]; found { + target.attempting = false + target.tryAfter, target.tryInterval = tryAfter(target.tryInterval) + } + return true + } } func (cm *ConnectionMaker) Refresh() { - cm.queryChan <- &ConnectionMakerInteraction{ - Interaction: Interaction{code: CMRefresh}} + cm.actionChan <- func() bool { return true } } func (cm *ConnectionMaker) String() string { - resultChan := make(chan interface{}, 0) - cm.queryChan <- &ConnectionMakerInteraction{ - Interaction: Interaction{code: CMStatus, resultChan: resultChan}} - result := <-resultChan - return result.(string) + resultChan := make(chan string, 0) + cm.actionChan <- func() bool { + var buf bytes.Buffer + for address, target := range cm.targets { + var fmtStr string + if target.attempting { + fmtStr = "%s (trying since %v)\n" + } else { + fmtStr = "%s (next try at %v)\n" + } + buf.WriteString(fmt.Sprintf(fmtStr, address, target.tryAfter)) + } + resultChan <- buf.String() + return false + } + return <-resultChan } -func (cm *ConnectionMaker) queryLoop(queryChan <-chan *ConnectionMakerInteraction) { +func (cm *ConnectionMaker) queryLoop(actionChan <-chan ConnectionMakerAction) { timer := time.NewTimer(MaxDuration) run := func() { timer.Reset(cm.checkStateAndAttemptConnections()) } for { select { - case query, ok := <-queryChan: - if !ok { - return - } - switch query.code { - case CMInitiate: - cm.cmdLineAddress[NormalisePeerAddr(query.address)] = true + case action := <-actionChan: + if action() { run() - case CMTerminated: - if target, found := cm.targets[query.address]; found { - target.attempting = false - target.tryAfter, target.tryInterval = tryAfter(target.tryInterval) - } - run() - case CMRefresh: - run() - case CMStatus: - query.resultChan <- cm.status() - default: - log.Fatal("Unexpected connection maker query:", query) } case <-timer.C: run() @@ -187,20 +175,6 @@ func (cm *ConnectionMaker) addTarget(address string) { } } -func (cm *ConnectionMaker) status() string { - var buf bytes.Buffer - for address, target := range cm.targets { - var fmtStr string - if target.attempting { - fmtStr = "%s (trying since %v)\n" - } else { - fmtStr = "%s (next try at %v)\n" - } - buf.WriteString(fmt.Sprintf(fmtStr, address, target.tryAfter)) - } - return buf.String() -} - func (cm *ConnectionMaker) attemptConnection(address string, acceptNewPeer bool) { log.Printf("->[%s] attempting connection\n", address) if err := cm.ourself.CreateConnection(address, acceptNewPeer); err != nil { diff --git a/router/gossip_test.go b/router/gossip_test.go index d72a517e..0146eef6 100644 --- a/router/gossip_test.go +++ b/router/gossip_test.go @@ -20,8 +20,8 @@ type mockChannelConnection struct { // channels when Router.OnGossip() calls async methods. func NewTestRouter(name PeerName) *Router { router := NewRouter(nil, name, "", nil, 10, 1024, nil) - router.ConnectionMaker.queryChan = make(chan *ConnectionMakerInteraction, ChannelSize) - router.Routes.queryChan = make(chan *Interaction, ChannelSize) + router.ConnectionMaker.actionChan = make(chan ConnectionMakerAction, ChannelSize) + router.Routes.actionChan = make(chan RoutesAction, ChannelSize) return router } diff --git a/router/local_peer.go b/router/local_peer.go index 3e499aad..4e40795a 100644 --- a/router/local_peer.go +++ b/router/local_peer.go @@ -9,23 +9,20 @@ import ( type LocalPeer struct { *Peer - Router *Router - queryChan chan<- *PeerInteraction + Router *Router + actionChan chan<- LocalPeerAction } -type PeerInteraction struct { - Interaction - payload interface{} -} +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() { - queryChan := make(chan *PeerInteraction, ChannelSize) - peer.queryChan = queryChan - go peer.queryLoop(queryChan) + actionChan := make(chan LocalPeerAction, ChannelSize) + peer.actionChan = actionChan + go peer.actorLoop(actionChan) } func (peer *LocalPeer) Forward(dstPeer *Peer, df bool, frame []byte, dec *EthernetDecoder) error { @@ -128,48 +125,38 @@ const ( // Sync. func (peer *LocalPeer) AddConnection(conn *LocalConnection) { resultChan := make(chan interface{}) - peer.queryChan <- &PeerInteraction{ - Interaction: Interaction{code: PAddConnection, resultChan: resultChan}, - payload: conn} + peer.actionChan <- func() { + peer.handleAddConnection(conn) + resultChan <- nil + } <-resultChan } // Async. func (peer *LocalPeer) ConnectionEstablished(conn *LocalConnection) { - peer.queryChan <- &PeerInteraction{ - Interaction: Interaction{code: PConnectionEstablished}, - payload: conn} + peer.actionChan <- func() { + peer.handleConnectionEstablished(conn) + } } // Sync. func (peer *LocalPeer) DeleteConnection(conn *LocalConnection) { resultChan := make(chan interface{}) - peer.queryChan <- &PeerInteraction{ - Interaction: Interaction{code: PDeleteConnection, resultChan: resultChan}, - payload: conn} + peer.actionChan <- func() { + peer.handleDeleteConnection(conn) + resultChan <- nil + } <-resultChan } // ACTOR server -func (peer *LocalPeer) queryLoop(queryChan <-chan *PeerInteraction) { +func (peer *LocalPeer) actorLoop(actionChan <-chan LocalPeerAction) { gossipTimer := time.Tick(GossipInterval) for { select { - case query, ok := <-queryChan: - if !ok { - return - } - switch query.code { - case PAddConnection: - peer.handleAddConnection(query.payload.(*LocalConnection)) - query.resultChan <- nil - case PConnectionEstablished: - peer.handleConnectionEstablished(query.payload.(*LocalConnection)) - case PDeleteConnection: - peer.handleDeleteConnection(query.payload.(*LocalConnection)) - query.resultChan <- nil - } + case action := <-actionChan: + action() case <-gossipTimer: peer.Router.SendAllGossip() } diff --git a/router/routes.go b/router/routes.go index 89c639a6..ca00721f 100644 --- a/router/routes.go +++ b/router/routes.go @@ -3,17 +3,16 @@ package router import ( "bytes" "fmt" - "log" "sync" ) type Routes struct { sync.RWMutex - ourself *Peer - peers *Peers - unicast map[PeerName]PeerName - broadcast map[PeerName][]PeerName - queryChan chan<- *Interaction + ourself *Peer + peers *Peers + unicast map[PeerName]PeerName + broadcast map[PeerName][]PeerName + actionChan chan<- RoutesAction } func NewRoutes(ourself *Peer, peers *Peers) *Routes { @@ -28,9 +27,9 @@ func NewRoutes(ourself *Peer, peers *Peers) *Routes { } func (routes *Routes) Start() { - queryChan := make(chan *Interaction, ChannelSize) - routes.queryChan = queryChan - go routes.queryLoop(queryChan) + actionChan := make(chan RoutesAction, ChannelSize) + routes.actionChan = actionChan + go routes.actorLoop(actionChan) } func (routes *Routes) Unicast(name PeerName) (PeerName, bool) { @@ -67,34 +66,25 @@ func (routes *Routes) String() string { // ACTOR client API -const ( - RRecalculate = iota -) +type RoutesAction func() // Async. func (routes *Routes) Recalculate() { - routes.queryChan <- &Interaction{code: RRecalculate} + routes.actionChan <- func() { + unicast := routes.calculateUnicast() + broadcast := routes.calculateBroadcast() + routes.Lock() + routes.unicast = unicast + routes.broadcast = broadcast + routes.Unlock() + } } // ACTOR server -func (routes *Routes) queryLoop(queryChan <-chan *Interaction) { +func (routes *Routes) actorLoop(actionChan <-chan RoutesAction) { for { - query, ok := <-queryChan - if !ok { - return - } - switch query.code { - case RRecalculate: - unicast := routes.calculateUnicast() - broadcast := routes.calculateBroadcast() - routes.Lock() - routes.unicast = unicast - routes.broadcast = broadcast - routes.Unlock() - default: - log.Fatal("Unexpected routes query:", query) - } + (<-actionChan)() } } diff --git a/router/utils.go b/router/utils.go index 1387e134..7f1ad1bb 100644 --- a/router/utils.go +++ b/router/utils.go @@ -10,11 +10,6 @@ import ( "net" ) -type Interaction struct { - code int - resultChan chan<- interface{} -} - func checkFatal(e error) { if e != nil { log.Fatal(e)