[engine] Exits gateway normally if stop it during startup

src/engine/gateway.go

@@ -21,6 +21,7 @@ import (
 
 type Gateway struct {
  sync.Mutex
+ runLock sync.Mutex
  repo config.Store
  mod *model.Model
  gc *cluster.GatewayCluster
@@ -69,16 +70,25 @@ func NewGateway() (*Gateway, error) {
  repo: repo,
  mod: mod,
  gc: gc,
- schedulers: make(map[string]pipelineScheduler, 100),
+ schedulers: make(map[string]pipelineScheduler),
  done: make(chan error, 1),
  }, nil
 }
 
 func (gw *Gateway) Close() {
+ gw.runLock.Lock()
+ defer gw.runLock.Unlock()
+
+ gw.Lock()
+ defer gw.Unlock()
+
  close(gw.done)
 }
 
 func (gw *Gateway) Run() (<-chan error, error) {
+ gw.runLock.Lock()
+ defer gw.runLock.Unlock()
+
  if !gw.startAt.IsZero() {
  return nil, fmt.Errorf("gateway already started")
  }
@@ -105,6 +115,9 @@ func (gw *Gateway) Run() (<-chan error, error) {
 }
 
 func (gw *Gateway) Stop() {
+ gw.runLock.Lock()
+ defer gw.runLock.Unlock()
+
  gw.Lock()
  defer gw.Unlock()
 

src/rest/api.go

@@ -5,6 +5,7 @@ import (
  "fmt"
  "net"
  "net/http"
+ "sync"
  "time"
 
  "github.com/ant0ine/go-json-rest/rest"
@@ -48,6 +49,7 @@ func (cam *clusterAvailabilityMiddleware) MiddlewareFunc(h rest.HandlerFunc) res
 ////
 
 type Rest struct {
+ sync.Mutex
  gateway *engine.Gateway
  gc *gateway.GatewayCluster
  server *http.Server
@@ -68,6 +70,9 @@ func NewRest(gateway *engine.Gateway) (*Rest, error) {
 }
 
 func (s *Rest) Start() (<-chan error, string, error) {
+ s.Lock()
+ defer s.Unlock()
+
  listenAddr := fmt.Sprintf("%s:9090", option.RestHost)
 
  adminServer, err := newAdminServer(s.gateway)
@@ -165,6 +170,7 @@ func (s *Rest) Start() (<-chan error, string, error) {
  // server exits after closing channel, ignore safely
  recover()
  }()
+
  err := s.server.Serve(tcpKeepAliveListener{ln.(*net.TCPListener)})
  if err != nil && !s.stopped {
  s.done <- err
@@ -176,6 +182,9 @@ func (s *Rest) Start() (<-chan error, string, error) {
 }
 
 func (s *Rest) Stop() {
+ s.Lock()
+ defer s.Unlock()
+
  s.stopped = true
 
  if s.server != nil {
@@ -185,10 +194,15 @@ func (s *Rest) Stop() {
  } else {
  logger.Debugf("[rest interface is shut down gracefully]")
  }
+ } else {
+ s.done <- nil
  }
 }
 
 func (s *Rest) Close() {
+ s.Lock()
+ defer s.Unlock()
+
  close(s.done)
 }
 

src/server/main.go

@@ -182,15 +182,16 @@ func setupHeapDumpSignalHandler() {
  }
  defer f.Close()
 
- logger.Debugf("[memory profiling started, heap dump to %s]",
+ logger.Debugf("[memory profiling started, heap dumps to %s]",
  option.MemProfileFile)
 
  // get up-to-date statistics
+ logger.Infof("[full gc is executing for heap dump, this may block the entire program]")
  runtime.GC()
 
  pprof.WriteHeapProfile(f)
 
- logger.Infof("[memory profiling finished, heap dump to %s]",
+ logger.Infof("[memory profiling finished, heap dumps to %s]",
  option.MemProfileFile)
  }()
  }