From 48c73598c46123e6f154275222dd279fbffe5daa Mon Sep 17 00:00:00 2001 From: Matthias Radestock Date: Tue, 24 Mar 2015 18:53:51 +0000 Subject: [PATCH 1/6] convert the connection actor to use closures The only slight trickiness is Shutdown. Generally connections shut down when an error is encountered, but we also need to be able to shut down connections w/o an error. Rather than making all the actor methods return a (terminated bool, error) tuple, we signal the non-error shutdown by sending nil as the action (instead of a closure). --- router/connection.go | 193 +++++++++++++++++++------------------------ 1 file changed, 86 insertions(+), 107 deletions(-) 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()) } From c54b3bcacf21e45149955e315a48d413cff08c42 Mon Sep 17 00:00:00 2001 From: Matthias Radestock Date: Tue, 24 Mar 2015 19:55:18 +0000 Subject: [PATCH 2/6] convert LocalPeer actor to use closures for methods --- router/local_peer.go | 53 +++++++++++++++++--------------------------- 1 file changed, 20 insertions(+), 33 deletions(-) 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() } From f0731f5d52454e1097609f8f30df982346010a56 Mon Sep 17 00:00:00 2001 From: Matthias Radestock Date: Tue, 24 Mar 2015 21:43:11 +0000 Subject: [PATCH 3/6] convert Routes actor to use closures for methods --- router/gossip_test.go | 2 +- router/routes.go | 48 +++++++++++++++++-------------------------- 2 files changed, 20 insertions(+), 30 deletions(-) diff --git a/router/gossip_test.go b/router/gossip_test.go index d72a517e..b6494be3 100644 --- a/router/gossip_test.go +++ b/router/gossip_test.go @@ -21,7 +21,7 @@ type mockChannelConnection struct { 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.Routes.actionChan = make(chan RoutesAction, ChannelSize) return router } 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)() } } From a06797dd957715f5a2bcb6e0f3b27a56151c0587 Mon Sep 17 00:00:00 2001 From: Matthias Radestock Date: Wed, 25 Mar 2015 00:10:42 +0000 Subject: [PATCH 4/6] convert ConnectionMaker actor to use closures for methods --- router/connection_maker.go | 98 ++++++++++++++------------------------ router/gossip_test.go | 2 +- 2 files changed, 37 insertions(+), 63 deletions(-) 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 b6494be3..0146eef6 100644 --- a/router/gossip_test.go +++ b/router/gossip_test.go @@ -20,7 +20,7 @@ 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.ConnectionMaker.actionChan = make(chan ConnectionMakerAction, ChannelSize) router.Routes.actionChan = make(chan RoutesAction, ChannelSize) return router } From 0e43a308584f01cdac60a9546e4435ca4a5d5590 Mon Sep 17 00:00:00 2001 From: Matthias Radestock Date: Wed, 25 Mar 2015 05:05:18 +0000 Subject: [PATCH 5/6] convert MDNSClient actor to use closures for methods We use a 'nil' closure to indicate shutdown. --- nameserver/mdns_client.go | 171 +++++++++++++++++--------------------- 1 file changed, 74 insertions(+), 97 deletions(-) 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? - } - } -} From ee6325d488ec6d1503288e8f34539ec1ddd81d8d Mon Sep 17 00:00:00 2001 From: Matthias Radestock Date: Wed, 25 Mar 2015 05:07:15 +0000 Subject: [PATCH 6/6] get rid of Interaction it's no longer needed in the closure-based actor method scheme --- router/utils.go | 5 ----- 1 file changed, 5 deletions(-) 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)