refactor listener update/destroy code

Don't close and reopen listeners
This commit is contained in:
Shivaram Lingamneni 2017-09-11 18:40:15 -04:00
parent d5528f6e56
commit 0f0f2d1314
2 changed files with 66 additions and 99 deletions

View File

@ -368,6 +368,7 @@ func (conf *Config) TLSListeners() map[string]*tls.Config {
tlsListeners := make(map[string]*tls.Config) tlsListeners := make(map[string]*tls.Config)
for s, tlsListenersConf := range conf.Server.TLSListeners { for s, tlsListenersConf := range conf.Server.TLSListeners {
config, err := tlsListenersConf.Config() config, err := tlsListenersConf.Config()
config.ClientAuth = tls.RequestClientCert
if err != nil { if err != nil {
log.Fatal(err) log.Fatal(err)
} }

View File

@ -57,26 +57,15 @@ type LineLenLimits struct {
Rest int Rest int
} }
// ListenerInterface represents an interface for a listener. // ListenerWrapper wraps a listener so it can be safely reconfigured or stopped
type ListenerInterface struct { type ListenerWrapper struct {
Listener net.Listener listener net.Listener
Events chan ListenerEvent tlsConfig *tls.Config
} shouldStop bool
// lets the ListenerWrapper inform the server that it has stopped:
const ( stopEvent chan bool
// DestroyListener instructs the listener to destroy itself. // protects atomic update of tlsConfig and shouldStop:
DestroyListener ListenerEventType = iota configMutex sync.Mutex
// UpdateListener instructs the listener to update itself (grab new certs, etc).
UpdateListener = iota
)
// ListenerEventType is the type of event this is.
type ListenerEventType int
// ListenerEvent is an event that's passed to the listener.
type ListenerEvent struct {
Type ListenerEventType
NewConfig *tls.Config
} }
// Server is the main Oragono server. // Server is the main Oragono server.
@ -102,8 +91,7 @@ type Server struct {
isupport *ISupportList isupport *ISupportList
klines *KLineManager klines *KLineManager
limits Limits limits Limits
listenerEventActMutex sync.Mutex listeners map[string]*ListenerWrapper
listeners map[string]*ListenerInterface
logger *logger.Manager logger *logger.Manager
MaxSendQBytes uint64 MaxSendQBytes uint64
monitoring map[string][]*Client monitoring map[string][]*Client
@ -219,7 +207,7 @@ func NewServer(configFilename string, config *Config, logger *logger.Manager) (*
Rest: config.Limits.LineLen.Rest, Rest: config.Limits.LineLen.Rest,
}, },
}, },
listeners: make(map[string]*ListenerInterface), listeners: make(map[string]*ListenerWrapper),
logger: logger, logger: logger,
MaxSendQBytes: config.Server.MaxSendQBytes, MaxSendQBytes: config.Server.MaxSendQBytes,
monitoring: make(map[string][]*Client), monitoring: make(map[string][]*Client),
@ -315,7 +303,7 @@ func NewServer(configFilename string, config *Config, logger *logger.Manager) (*
tlsListeners := config.TLSListeners() tlsListeners := config.TLSListeners()
for _, addr := range config.Server.Listen { for _, addr := range config.Server.Listen {
server.listeners[addr] = server.createListener(addr, tlsListeners) server.listeners[addr] = server.createListener(addr, tlsListeners[addr])
} }
if len(tlsListeners) == 0 { if len(tlsListeners) == 0 {
@ -510,90 +498,63 @@ func (server *Server) Run() {
// //
// createListener starts the given listeners. // createListener starts the given listeners.
func (server *Server) createListener(addr string, tlsMap map[string]*tls.Config) *ListenerInterface { func (server *Server) createListener(addr string, tlsConfig *tls.Config) *ListenerWrapper {
config, listenTLS := tlsMap[addr]
// make listener event channel
listenerEventChannel := make(chan ListenerEvent, 1)
// make listener // make listener
listener, err := net.Listen("tcp", addr) listener, err := net.Listen("tcp", addr)
if err != nil { if err != nil {
log.Fatal(server, "listen error: ", err) log.Fatal(server, "listen error: ", err)
} }
// throw our details to the server so we can be modified/killed later
wrapper := ListenerWrapper{
listener: listener,
tlsConfig: tlsConfig,
shouldStop: false,
stopEvent: make(chan bool, 1),
}
// TODO(slingamn) move all logging of listener status to rehash()
tlsString := "plaintext" tlsString := "plaintext"
if listenTLS { if tlsConfig != nil {
config.ClientAuth = tls.RequestClientCert
listener = tls.NewListener(listener, config)
tlsString = "TLS" tlsString = "TLS"
} }
// throw our details to the server so we can be modified/killed later
li := ListenerInterface{
Events: listenerEventChannel,
Listener: listener,
}
// start listening
server.logger.Info("listeners", fmt.Sprintf("listening on %s using %s.", addr, tlsString)) server.logger.Info("listeners", fmt.Sprintf("listening on %s using %s.", addr, tlsString))
var shouldStop bool
// setup accept goroutine // setup accept goroutine
go func() { go func() {
for { for {
conn, err := listener.Accept() conn, err := listener.Accept()
// synchronously access config data:
// whether TLS is enabled and whether we should stop listening
wrapper.configMutex.Lock()
shouldStop = wrapper.shouldStop
tlsConfig = wrapper.tlsConfig
wrapper.configMutex.Unlock()
if err == nil { if err == nil {
if tlsConfig != nil {
conn = tls.Server(conn, tlsConfig)
}
newConn := clientConn{ newConn := clientConn{
Conn: conn, Conn: conn,
IsTLS: listenTLS, IsTLS: tlsConfig != nil,
} }
// hand off the connection
server.newConns <- newConn server.newConns <- newConn
} }
select { if shouldStop {
case event := <-li.Events: listener.Close()
// this is used to confirm that whoever passed us this event has closed the existing listener correctly (in an attempt to get us to notice the event). wrapper.stopEvent <- true
// this is required to keep REHASH from having a very small race possibility of killing the primary listener return
server.listenerEventActMutex.Lock()
server.listenerEventActMutex.Unlock()
if event.Type == DestroyListener {
// listener should already be closed, this is just for safety
listener.Close()
return
} else if event.Type == UpdateListener {
// close old listener
listener.Close()
// make new listener
listener, err = net.Listen("tcp", addr)
if err != nil {
log.Fatal(server, "listen error: ", err)
}
tlsString := "plaintext"
if event.NewConfig != nil {
config = event.NewConfig
config.ClientAuth = tls.RequestClientCert
listener = tls.NewListener(listener, config)
tlsString = "TLS"
}
// update server ListenerInterface
li.Listener = listener
// print notice
server.logger.Info("listeners", fmt.Sprintf("updated listener %s using %s.", addr, tlsString))
}
default:
// no events waiting for us, fall-through and continue
} }
} }
}() }()
return &li return &wrapper
} }
// //
@ -1649,32 +1610,37 @@ func (server *Server) rehash() error {
tlsListeners := config.TLSListeners() tlsListeners := config.TLSListeners()
for addr := range server.listeners { for addr := range server.listeners {
currentListener := server.listeners[addr] currentListener := server.listeners[addr]
var exists bool var stillConfigured bool
for _, newaddr := range config.Server.Listen { for _, newaddr := range config.Server.Listen {
if newaddr == addr { if newaddr == addr {
exists = true stillConfigured = true
break break
} }
} }
server.listenerEventActMutex.Lock() // pass new config information to the listener, to be picked up after
if exists { // its next Accept(). this is like sending over a buffered channel of
// update old listener // size 1, but where sending a second item overwrites the buffered item
currentListener.Events <- ListenerEvent{ // instead of blocking.
Type: UpdateListener, currentListener.configMutex.Lock()
NewConfig: tlsListeners[addr], currentListener.shouldStop = !stillConfigured
} currentListener.tlsConfig = tlsListeners[addr]
currentListener.configMutex.Unlock()
if stillConfigured {
server.logger.Info("rehash",
fmt.Sprintf("now listening on %s, tls=%t.", addr, (currentListener.tlsConfig != nil)),
)
} else { } else {
// destroy this listener, since it is no longer in the config // tell the listener it should stop by interrupting its Accept() call:
currentListener.Events <- ListenerEvent{ currentListener.listener.Close()
Type: DestroyListener, // XXX there is no guarantee from the API when the address will actually
} // free for bind(2) again; this solution "seems to work". See here:
// https://github.com/golang/go/issues/21833
<-currentListener.stopEvent
delete(server.listeners, addr) delete(server.listeners, addr)
server.logger.Info("rehash", fmt.Sprintf("stopped listening on %s.", addr))
} }
// force listener to apply the event right away
// (this causes its Accept() call to return immediately with an error)
currentListener.Listener.Close()
server.listenerEventActMutex.Unlock()
} }
// create new listeners that were not previously configured // create new listeners that were not previously configured
@ -1682,7 +1648,7 @@ func (server *Server) rehash() error {
_, exists := server.listeners[newaddr] _, exists := server.listeners[newaddr]
if !exists { if !exists {
// make new listener // make new listener
server.listeners[newaddr] = server.createListener(newaddr, tlsListeners) server.listeners[newaddr] = server.createListener(newaddr, tlsListeners[newaddr])
} }
} }