server: asynchronously handle internal broadcast/send requests

This commit modifies the request handling within the sever’s
queryHandler goroutine to ensure that requests from the ChannelRouter
or other related sub-systems don’t block the main processing loop.

We do this simply by launching a goroutine to handle the dispatch of
the request.
This commit is contained in:
Olaoluwa Osuntokun committed 2017-01-24 17:06:32 -08:00
1 parent ddc3c3ab35
commit 73d5daa2c3
1 file changed
+41 -29
+41 -29
View File
@@ -578,49 +578,60 @@ out:
srvrLog.Debugf("Broadcasting %v messages", len(bMsg.msgs)) srvrLog.Debugf("Broadcasting %v messages", len(bMsg.msgs))
s.peersMtx.RLock() // Launch a new goroutine to handle the broadcast
for _, peer := range s.peersByPub { // request, this allows us process this request
if ignore != nil && // asynchronously without blocking subsequent broadcast
peer.addr.IdentityKey.IsEqual(ignore) { // requests.
go func() {
s.peersMtx.RLock()
for _, peer := range s.peersByPub {
if ignore != nil &&
peer.addr.IdentityKey.IsEqual(ignore) {
srvrLog.Debugf("Skipping %v in broadcast", srvrLog.Debugf("Skipping %v in broadcast",
ignore.SerializeCompressed()) ignore.SerializeCompressed())
continue continue
}
for _, msg := range bMsg.msgs {
peer.queueMsg(msg, nil)
}
} }
s.peersMtx.RUnlock()
for _, msg := range bMsg.msgs { bMsg.errChan <- nil
peer.queueMsg(msg, nil) }()
}
}
s.peersMtx.RUnlock()
bMsg.errChan <- nil
case sMsg := <-s.sendRequests: case sMsg := <-s.sendRequests:
// TODO(roasbeef): use [33]byte everywhere instead // TODO(roasbeef): use [33]byte everywhere instead
// * eliminate usage of mutexes, funnel all peer // * eliminate usage of mutexes, funnel all peer
// mutation to this goroutine // mutation to this goroutine
target := sMsg.target.SerializeCompressed() target := sMsg.target.SerializeCompressed()
srvrLog.Debugf("Attempting to send msgs %v to: %x", srvrLog.Debugf("Attempting to send msgs %v to: %x",
len(sMsg.msgs), target) len(sMsg.msgs), target)
s.peersMtx.RLock() // Launch a new goroutine to handle this send request,
targetPeer, ok := s.peersByPub[string(target)] // this allows us process this request asynchronously
if !ok { // without blocking future send requests.
go func() {
s.peersMtx.RLock()
targetPeer, ok := s.peersByPub[string(target)]
if !ok {
s.peersMtx.RUnlock()
srvrLog.Errorf("unable to send message to %x, "+
"peer not found", target)
sMsg.errChan <- errors.New("peer not found")
return
}
for _, msg := range sMsg.msgs {
targetPeer.queueMsg(msg, nil)
}
s.peersMtx.RUnlock() s.peersMtx.RUnlock()
srvrLog.Errorf("unable to send message to %x, "+
"peer not found", target)
sMsg.errChan <- errors.New("peer not found")
continue
}
for _, msg := range sMsg.msgs { sMsg.errChan <- nil
targetPeer.queueMsg(msg, nil) }()
}
s.peersMtx.RUnlock()
sMsg.errChan <- nil
case query := <-s.queries: case query := <-s.queries:
switch msg := query.(type) { switch msg := query.(type) {
case *connectPeerMsg: case *connectPeerMsg:
@@ -695,6 +706,7 @@ func (s *server) handleConnectPeer(msg *connectPeerMsg) {
Addr: addr, Addr: addr,
Permanent: true, Permanent: true,
}) })
msg.err <- nil
} else { } else {
// If we're not making a persistent connection, then we'll // If we're not making a persistent connection, then we'll
// attempt to connect o the target peer, returning an error // attempt to connect o the target peer, returning an error