issue #84: Fix race condition

httpserver/pool.go

@@ -69,7 +69,7 @@ type PoolServer interface {
 	Filter(field FieldType, pattern, regex string) PoolServer
 
 	IsRunning(asLeast bool) bool
-	WaitNotify(ctx context.Context)
+	WaitNotify(ctx context.Context, cancel context.CancelFunc)
 
 	Listen(handler http.Handler) liberr.Error
 	Restart()
@@ -310,7 +310,7 @@ func (p pool) IsRunning(atLeast bool) bool {
 	return r
 }
 
-func (p pool) WaitNotify(ctx context.Context) {
+func (p pool) WaitNotify(ctx context.Context, cancel context.CancelFunc) {
 	// Wait for interrupt signal to gracefully shutdown the server with
 	// a timeout of 5 seconds.
 	quit := make(chan os.Signal, 1)
@@ -321,8 +321,14 @@ func (p pool) WaitNotify(ctx context.Context) {
 	select {
 	case <-quit:
 		p.Shutdown()
+		if cancel != nil {
+			cancel()
+		}
 	case <-ctx.Done():
 		p.Shutdown()
+		if cancel != nil {
+			cancel()
+		}
 	}
 }
 

httpserver/run.go

@@ -0,0 +1,313 @@
+/***********************************************************************************************************************
+ *
+ *   MIT License
+ *
+ *   Copyright (c) 2021 Nicolas JUHEL
+ *
+ *   Permission is hereby granted, free of charge, to any person obtaining a copy
+ *   of this software and associated documentation files (the "Software"), to deal
+ *   in the Software without restriction, including without limitation the rights
+ *   to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
+ *   copies of the Software, and to permit persons to whom the Software is
+ *   furnished to do so, subject to the following conditions:
+ *
+ *   The above copyright notice and this permission notice shall be included in all
+ *   copies or substantial portions of the Software.
+ *
+ *   THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+ *   IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+ *   FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+ *   AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+ *   LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
+ *   OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
+ *   SOFTWARE.
+ *
+ *
+ **********************************************************************************************************************/
+
+package httpserver
+
+import (
+	"context"
+	"errors"
+	"log"
+	"net"
+	"net/http"
+	"os"
+	"os/signal"
+	"sync/atomic"
+	"syscall"
+	"time"
+
+	liberr "github.com/nabbar/golib/errors"
+	liblog "github.com/nabbar/golib/logger"
+	"golang.org/x/net/http2"
+)
+
+type srvRun struct {
+	run *atomic.Value
+	snm string
+	srv *http.Server
+	ctx context.Context
+	cnl context.CancelFunc
+}
+
+type run interface {
+	IsRunning() bool
+	WaitNotify()
+	Listen(cfg *ServerConfig, handler http.Handler) liberr.Error
+	Restart(cfg *ServerConfig)
+	Shutdown()
+}
+
+func newRun() run {
+	return &srvRun{
+		run: new(atomic.Value),
+		srv: nil,
+	}
+}
+
+func (s *srvRun) getRunning() bool {
+	if s.run == nil {
+		return false
+	} else if i := s.run.Load(); i == nil {
+		return false
+	} else if b, ok := i.(bool); !ok {
+		return false
+	} else {
+		return b
+	}
+}
+
+func (s *srvRun) setRunning(state bool) {
+	s.run.Store(state)
+}
+
+func (s *srvRun) IsRunning() bool {
+	return s.getRunning()
+}
+
+func (s *srvRun) WaitNotify() {
+	// Wait for interrupt signal to gracefully shutdown the server with
+	// a timeout of 5 seconds.
+	quit := make(chan os.Signal, 1)
+	signal.Notify(quit, syscall.SIGINT)
+	signal.Notify(quit, syscall.SIGTERM)
+	signal.Notify(quit, syscall.SIGQUIT)
+
+	select {
+	case <-quit:
+		s.Shutdown()
+	case <-s.ctx.Done():
+		s.Shutdown()
+	}
+}
+
+func (s *srvRun) Merge(srv Server) bool {
+	panic("implement me")
+}
+
+func (s *srvRun) Listen(cfg *ServerConfig, handler http.Handler) liberr.Error {
+	ssl, err := cfg.GetTLS()
+	if err != nil {
+		return err
+	}
+
+	bind := cfg.GetListen().Host
+	name := cfg.Name
+	if name == "" {
+		name = bind
+	}
+
+	srv := &http.Server{
+		Addr:     cfg.GetListen().Host,
+		ErrorLog: liblog.GetLogger(liblog.ErrorLevel, log.LstdFlags|log.Lmicroseconds, "[http/http2 server '%s']", name),
+	}
+
+	if cfg.ReadTimeout > 0 {
+		srv.ReadTimeout = cfg.ReadTimeout
+	}
+
+	if cfg.ReadHeaderTimeout > 0 {
+		srv.ReadHeaderTimeout = cfg.ReadHeaderTimeout
+	}
+
+	if cfg.WriteTimeout > 0 {
+		srv.WriteTimeout = cfg.WriteTimeout
+	}
+
+	if cfg.MaxHeaderBytes > 0 {
+		srv.MaxHeaderBytes = cfg.MaxHeaderBytes
+	}
+
+	if cfg.IdleTimeout > 0 {
+		srv.IdleTimeout = cfg.IdleTimeout
+	}
+
+	if ssl.LenCertificatePair() > 0 {
+		srv.TLSConfig = ssl.TlsConfig("")
+	}
+
+	if handler != nil {
+		srv.Handler = handler
+	} else if s.srv != nil {
+		srv.Handler = s.srv.Handler
+	}
+
+	s2 := &http2.Server{}
+
+	if cfg.MaxHandlers > 0 {
+		s2.MaxHandlers = cfg.MaxHandlers
+	}
+
+	if cfg.MaxConcurrentStreams > 0 {
+		s2.MaxConcurrentStreams = cfg.MaxConcurrentStreams
+	}
+
+	if cfg.PermitProhibitedCipherSuites {
+		s2.PermitProhibitedCipherSuites = true
+	}
+
+	if cfg.IdleTimeout > 0 {
+		s2.IdleTimeout = cfg.IdleTimeout
+	}
+
+	if cfg.MaxUploadBufferPerConnection > 0 {
+		s2.MaxUploadBufferPerConnection = cfg.MaxUploadBufferPerConnection
+	}
+
+	if cfg.MaxUploadBufferPerStream > 0 {
+		s2.MaxUploadBufferPerStream = cfg.MaxUploadBufferPerStream
+	}
+
+	if e := http2.ConfigureServer(srv, s2); e != nil {
+		return ErrorHTTP2Configure.ErrorParent(e)
+	}
+
+	if s.IsRunning() {
+		s.Shutdown()
+	}
+
+	for i := 0; i < 5; i++ {
+		if e := s.PortInUse(cfg.Listen); e != nil {
+			s.Shutdown()
+		} else {
+			break
+		}
+	}
+
+	if s.ctx != nil && s.ctx.Err() == nil && s.cnl != nil {
+		s.cnl()
+		s.ctx = nil
+		s.cnl = nil
+	}
+
+	s.ctx, s.cnl = context.WithCancel(cfg.getContext())
+	s.snm = name
+	s.srv = srv
+
+	go func(name, host string) {
+
+		defer func() {
+			if s.ctx != nil && s.cnl != nil && s.ctx.Err() == nil {
+				s.cnl()
+			}
+			s.setRunning(false)
+		}()
+
+		s.srv.BaseContext = func(listener net.Listener) context.Context {
+			return s.ctx
+		}
+
+		var err error
+
+		if ssl.LenCertificatePair() > 0 {
+			liblog.InfoLevel.Logf("TLS Server '%s' is starting with bindable: %s", name, host)
+
+			s.setRunning(true)
+			err = s.srv.ListenAndServeTLS("", "")
+		} else {
+			liblog.InfoLevel.Logf("Server '%s' is starting with bindable: %s", name, host)
+
+			s.setRunning(true)
+			err = s.srv.ListenAndServe()
+		}
+
+		if err != nil && s.ctx.Err() != nil && s.ctx.Err().Error() == err.Error() {
+			return
+		} else if err != nil && errors.Is(err, http.ErrServerClosed) {
+			return
+		} else if err != nil {
+			liblog.ErrorLevel.LogErrorCtxf(liblog.NilLevel, "Listen Server '%s'", err, name)
+		}
+	}(name, bind)
+
+	return nil
+}
+
+func (s *srvRun) Restart(cfg *ServerConfig) {
+	_ = s.Listen(cfg, nil)
+}
+
+func (s *srvRun) Shutdown() {
+	ctx, cancel := context.WithTimeout(context.Background(), timeoutShutdown)
+
+	defer func() {
+		cancel()
+
+		if s.srv != nil {
+			_ = s.srv.Close()
+		}
+
+		s.setRunning(false)
+	}()
+
+	liblog.InfoLevel.Logf("Shutdown Server '%s'...", s.snm)
+
+	if s.cnl != nil && s.ctx != nil && s.ctx.Err() == nil {
+		s.cnl()
+	}
+
+	if s.srv != nil {
+		err := s.srv.Shutdown(ctx)
+		if err != nil && !errors.Is(err, http.ErrServerClosed) {
+			liblog.ErrorLevel.Logf("Shutdown Server '%s' Error: %v", s.snm, err)
+		}
+	}
+}
+
+func (s *srvRun) PortInUse(listen string) liberr.Error {
+	var (
+		dia = net.Dialer{}
+		con net.Conn
+		err error
+		ctx context.Context
+		cnl context.CancelFunc
+	)
+
+	defer func() {
+		if cnl != nil {
+			cnl()
+		}
+		if con != nil {
+			_ = con.Close()
+		}
+	}()
+
+	ctx, cnl = context.WithTimeout(context.TODO(), 2*time.Second)
+	con, err = dia.DialContext(ctx, "tcp", listen)
+
+	if con != nil {
+		_ = con.Close()
+		con = nil
+	}
+
+	cnl()
+	cnl = nil
+
+	if err != nil {
+		return nil
+	}
+
+	return ErrorPortUse.Error(nil)
+}

httpserver/server.go

@@ -28,20 +28,11 @@ package httpserver
 
 import (
 	"context"
-	"errors"
-	"log"
-	"net"
 	"net/http"
-	"os"
-	"os/signal"
 	"sync/atomic"
-	"syscall"
 	"time"
 
-	"golang.org/x/net/http2"
-
 	liberr "github.com/nabbar/golib/errors"
-	liblog "github.com/nabbar/golib/logger"
 )
 
 const (
@@ -50,9 +41,8 @@ const (
 )
 
 type server struct {
-	run atomic.Value
+	run *atomic.Value
 	cfg *ServerConfig
-	srv *http.Server
 	cnl context.CancelFunc
 }
 
@@ -77,11 +67,27 @@ type Server interface {
 func NewServer(cfg *ServerConfig) Server {
 	return &server{
 		cfg: cfg,
-		srv: nil,
+		run: new(atomic.Value),
 		cnl: nil,
 	}
 }
 
+func (s *server) getRun() run {
+	if s.run == nil {
+		return newRun()
+	} else if i := s.run.Load(); i == nil {
+		return newRun()
+	} else if r, ok := i.(run); !ok {
+		return newRun()
+	} else {
+		return r
+	}
+}
+
+func (s *server) setRun(r run) {
+	s.run.Store(r)
+}
+
 func (s *server) GetConfig() *ServerConfig {
 	return s.cfg
 }
@@ -107,165 +113,23 @@ func (s *server) GetExpose() string {
 }
 
 func (s *server) IsRunning() bool {
-	if i := s.run.Load(); i == nil {
-		return false
-	} else if b, ok := i.(bool); !ok {
-		return false
-	} else {
-		return b
-	}
+	return s.getRun().IsRunning()
 }
 
 func (s *server) IsTLS() bool {
 	return s.cfg.IsTLS()
 }
 
-func (s *server) setRunning() {
-	s.run.Store(true)
-}
-
-func (s *server) setNotRunning() {
-	s.run.Store(false)
-}
-
 func (s *server) Listen(handler http.Handler) liberr.Error {
-	ssl, err := s.cfg.GetTLS()
-	if err != nil {
-		return err
-	}
-
-	srv := &http.Server{
-		Addr:     s.GetBindable(),
-		ErrorLog: liblog.GetLogger(liblog.ErrorLevel, log.LstdFlags|log.Lmicroseconds, "[http/http2 server '%s']", s.GetName()),
-	}
-
-	if s.cfg.ReadTimeout > 0 {
-		srv.ReadTimeout = s.cfg.ReadTimeout
-	}
-
-	if s.cfg.ReadHeaderTimeout > 0 {
-		srv.ReadHeaderTimeout = s.cfg.ReadHeaderTimeout
-	}
-
-	if s.cfg.WriteTimeout > 0 {
-		srv.WriteTimeout = s.cfg.WriteTimeout
-	}
-
-	if s.cfg.MaxHeaderBytes > 0 {
-		srv.MaxHeaderBytes = s.cfg.MaxHeaderBytes
-	}
-
-	if s.cfg.IdleTimeout > 0 {
-		srv.IdleTimeout = s.cfg.IdleTimeout
-	}
-
-	if ssl.LenCertificatePair() > 0 {
-		srv.TLSConfig = ssl.TlsConfig("")
-	}
-
-	if handler != nil {
-		srv.Handler = handler
-	} else if s.srv != nil {
-		srv.Handler = s.srv.Handler
-	}
-
-	cfg := &http2.Server{}
-
-	if s.cfg.MaxHandlers > 0 {
-		cfg.MaxHandlers = s.cfg.MaxHandlers
-	}
-
-	if s.cfg.MaxConcurrentStreams > 0 {
-		cfg.MaxConcurrentStreams = s.cfg.MaxConcurrentStreams
-	}
-
-	if s.cfg.PermitProhibitedCipherSuites {
-		cfg.PermitProhibitedCipherSuites = true
-	}
-
-	if s.cfg.IdleTimeout > 0 {
-		cfg.IdleTimeout = s.cfg.IdleTimeout
-	}
-
-	if s.cfg.MaxUploadBufferPerConnection > 0 {
-		cfg.MaxUploadBufferPerConnection = s.cfg.MaxUploadBufferPerConnection
-	}
-
-	if s.cfg.MaxUploadBufferPerStream > 0 {
-		cfg.MaxUploadBufferPerStream = s.cfg.MaxUploadBufferPerStream
-	}
-
-	if e := http2.ConfigureServer(srv, cfg); e != nil {
-		return ErrorHTTP2Configure.ErrorParent(e)
-	}
-
-	if s.IsRunning() {
-		s.Shutdown()
-	}
-
-	for i := 0; i < 5; i++ {
-		if e := s.PortInUse(); e != nil {
-			s.Shutdown()
-		} else {
-			break
-		}
-	}
-
-	s.srv = srv
-
-	go func() {
-		ctx, cnl := context.WithCancel(s.cfg.getContext())
-		s.cnl = cnl
-
-		defer func() {
-			cnl()
-			s.setNotRunning()
-		}()
-
-		s.srv.BaseContext = func(listener net.Listener) context.Context {
-			return ctx
-		}
-
-		var err error
-
-		if ssl.LenCertificatePair() > 0 {
-			liblog.InfoLevel.Logf("TLS Server '%s' is starting with bindable: %s", s.GetName(), s.GetBindable())
-
-			s.setRunning()
-			err = s.srv.ListenAndServeTLS("", "")
-		} else {
-			liblog.InfoLevel.Logf("Server '%s' is starting with bindable: %s", s.GetName(), s.GetBindable())
-
-			s.setRunning()
-			err = s.srv.ListenAndServe()
-		}
-
-		if err != nil && ctx.Err() != nil && ctx.Err().Error() == err.Error() {
-			return
-		} else if err != nil && errors.Is(err, http.ErrServerClosed) {
-			return
-		} else if err != nil {
-			liblog.ErrorLevel.LogErrorCtxf(liblog.NilLevel, "Listen Server '%s'", err, s.GetName())
-		}
-	}()
-
-	return nil
+	r := s.getRun()
+	e := r.Listen(s.cfg, handler)
+	s.setRun(r)
+	return e
 }
 
 func (s *server) WaitNotify() {
-	// Wait for interrupt signal to gracefully shutdown the server with
-	// a timeout of 5 seconds.
-	quit := make(chan os.Signal, 1)
-	signal.Notify(quit, syscall.SIGINT)
-	signal.Notify(quit, syscall.SIGTERM)
-	signal.Notify(quit, syscall.SIGQUIT)
-
-	select {
-	case <-quit:
-		s.Shutdown()
-	case <-s.cfg.getContext().Done():
-		s.Shutdown()
-	}
+	r := s.getRun()
+	r.WaitNotify()
 }
 
 func (s *server) Restart() {
@@ -273,29 +137,9 @@ func (s *server) Restart() {
 }
 
 func (s *server) Shutdown() {
-	ctx, cancel := context.WithTimeout(context.Background(), timeoutShutdown)
-	defer func() {
-		cancel()
-
-		if s.srv != nil {
-			_ = s.srv.Close()
-		}
-
-		s.setNotRunning()
-	}()
-
-	liblog.InfoLevel.Logf("Shutdown Server '%s'...", s.GetName())
-
-	if s.cnl != nil {
-		s.cnl()
-	}
-
-	if s.srv != nil {
-		err := s.srv.Shutdown(ctx)
-		if err != nil && !errors.Is(err, http.ErrServerClosed) {
-			liblog.ErrorLevel.Logf("Shutdown Server '%s' Error: %v", s.GetName(), err)
-		}
-	}
+	r := s.getRun()
+	r.Shutdown()
+	s.setRun(r)
 }
 
 func (s *server) Merge(srv Server) bool {
@@ -306,31 +150,3 @@ func (s *server) Merge(srv Server) bool {
 
 	return false
 }
-
-func (s *server) PortInUse() liberr.Error {
-	var (
-		dia = net.Dialer{}
-		con net.Conn
-		err error
-		ctx context.Context
-		cnl context.CancelFunc
-	)
-
-	defer func() {
-		if cnl != nil {
-			cnl()
-		}
-		if con != nil {
-			_ = con.Close()
-		}
-	}()
-
-	ctx, cnl = context.WithTimeout(context.TODO(), 2*time.Second)
-	con, err = dia.DialContext(ctx, "tcp", s.cfg.Listen)
-
-	if err != nil {
-		return nil
-	}
-
-	return ErrorPortUse.Error(nil)
-}

test/test-httpserver/main.go

@@ -0,0 +1 @@
+package test_httpserver