Merge pull request #499 from alouche/fix_potential_race_condition_shutdown

Fix potential race condition on shutdown (pool.reap/server.handleConsulConn)
This commit is contained in:
Armon Dadgar 2014-11-26 11:39:33 -08:00
commit 795bcc57a5
2 changed files with 15 additions and 8 deletions

View File

@ -358,12 +358,12 @@ func (p *ConnPool) RPC(addr net.Addr, version int, method string, args interface
// Reap is used to close conns open over maxTime
func (p *ConnPool) reap() {
for !p.shutdown {
for {
// Sleep for a while
select {
case <-time.After(time.Second):
case <-p.shutdownCh:
return
case <-time.After(time.Second):
}
// Reap all old conns

View File

@ -3,16 +3,17 @@ package consul
import (
"crypto/tls"
"fmt"
"github.com/armon/go-metrics"
"github.com/hashicorp/consul/consul/structs"
"github.com/hashicorp/go-msgpack/codec"
"github.com/hashicorp/yamux"
"github.com/inconshreveable/muxado"
"io"
"math/rand"
"net"
"strings"
"time"
"github.com/armon/go-metrics"
"github.com/hashicorp/consul/consul/structs"
"github.com/hashicorp/go-msgpack/codec"
"github.com/hashicorp/yamux"
"github.com/inconshreveable/muxado"
)
type RPCType byte
@ -149,7 +150,13 @@ func (s *Server) handleMultiplexV2(conn net.Conn) {
func (s *Server) handleConsulConn(conn net.Conn) {
defer conn.Close()
rpcCodec := codec.GoRpc.ServerCodec(conn, msgpackHandle)
for !s.shutdown {
for {
select {
case <-s.shutdownCh:
return
default:
}
if err := s.rpcServer.ServeRequest(rpcCodec); err != nil {
if err != io.EOF && !strings.Contains(err.Error(), "closed") {
s.logger.Printf("[ERR] consul.rpc: RPC error: %v (%v)", err, conn)