static pool logic update, better watching techniques

8 files changed, 119 insertions, 125 deletions

diff --git a/server.go b/server.go
index c194bc78..75cf3691 100644
--- a/server.go
+++ b/server.go
@@ -29,7 +29,7 @@ type Server struct {
 	cfg *ServerConfig
 
 	// observes pool events (can be attached to multiple pools at the same time)
-	listener func(event int, ctx interface{})
+	lsn func(event int, ctx interface{})
 
 	// protects pool while the re-configuration
 	mu sync.Mutex
@@ -51,7 +51,7 @@ func NewServer(cfg *ServerConfig) *Server {
 
 // AddListener attaches server event watcher.
 func (srv *Server) Listen(l func(event int, ctx interface{})) {
-	srv.listener = l
+	srv.lsn = l
 }
 
 // Start underlying worker pool, configure factory and command provider.
@@ -171,7 +171,7 @@ func (srv *Server) Pool() Pool {
 
 // AddListener pool events.
 func (srv *Server) poolListener(event int, ctx interface{}) {
-	// bypassing to user specified listener
+	// bypassing to user specified lsn
 	srv.throw(event, ctx)
 
 	if event == EventPoolError {
@@ -192,7 +192,7 @@ func (srv *Server) poolListener(event int, ctx interface{}) {
 
 // throw invokes event handler if any.
 func (srv *Server) throw(event int, ctx interface{}) {
-	if srv.listener != nil {
-		srv.listener(event, ctx)
+	if srv.lsn != nil {
+		srv.lsn(event, ctx)
 	}
 }
diff --git a/service/http/server.go b/service/http/handler.go
index cad2e46d..1df00e5a 100644
--- a/service/http/server.go
+++ b/service/http/handler.go
@@ -30,84 +30,84 @@ type Event struct {
 	Error error
 }
 
-// Server serves http connections to underlying PHP application using PSR-7 protocol. Context will include request headers,
+// Handler serves http connections to underlying PHP application using PSR-7 protocol. Context will include request headers,
 // parsed files and query, payload will include parsed form dataTree (if any).
-type Server struct {
+type Handler struct {
 	cfg      *Config
 	listener func(event int, ctx interface{})
 	rr       *roadrunner.Server
 }
 
 // AddListener attaches pool event watcher.
-func (s *Server) Listen(l func(event int, ctx interface{})) {
-	s.listener = l
+func (h *Handler) Listen(l func(event int, ctx interface{})) {
+	h.listener = l
 }
 
 // Handle serve using PSR-7 requests passed to underlying application. Attempts to serve static files first if enabled.
-func (s *Server) ServeHTTP(w http.ResponseWriter, r *http.Request) {
+func (h *Handler) ServeHTTP(w http.ResponseWriter, r *http.Request) {
 	// validating request size
-	if s.cfg.MaxRequest != 0 {
+	if h.cfg.MaxRequest != 0 {
 		if length := r.Header.Get("content-length"); length != "" {
 			if size, err := strconv.ParseInt(length, 10, 64); err != nil {
-				s.handleError(w, r, err)
+				h.handleError(w, r, err)
 				return
-			} else if size > s.cfg.MaxRequest*1024*1024 {
-				s.handleError(w, r, errors.New("request body max size is exceeded"))
+			} else if size > h.cfg.MaxRequest*1024*1024 {
+				h.handleError(w, r, errors.New("request body max size is exceeded"))
 				return
 			}
 		}
 	}
 
-	req, err := NewRequest(r, s.cfg.Uploads)
+	req, err := NewRequest(r, h.cfg.Uploads)
 	if err != nil {
-		s.handleError(w, r, err)
+		h.handleError(w, r, err)
 		return
 	}
 
 	if err = req.Open(); err != nil {
-		s.handleError(w, r, err)
+		h.handleError(w, r, err)
 		return
 	}
 	defer req.Close()
 
 	p, err := req.Payload()
 	if err != nil {
-		s.handleError(w, r, err)
+		h.handleError(w, r, err)
 		return
 	}
 
-	rsp, err := s.rr.Exec(p)
+	rsp, err := h.rr.Exec(p)
 	if err != nil {
-		s.handleError(w, r, err)
+		h.handleError(w, r, err)
 		return
 	}
 
 	resp, err := NewResponse(rsp)
 	if err != nil {
-		s.handleError(w, r, err)
+		h.handleError(w, r, err)
 		return
 	}
 
-	s.handleResponse(req, resp)
+	h.handleResponse(req, resp)
 	resp.Write(w)
 }
 
 // handleResponse triggers response event.
-func (s *Server) handleResponse(req *Request, resp *Response) {
-	s.throw(EventResponse, &Event{Method: req.Method, Uri: req.Uri, Status: resp.Status})
+func (h *Handler) handleResponse(req *Request, resp *Response) {
+	h.throw(EventResponse, &Event{Method: req.Method, Uri: req.Uri, Status: resp.Status})
 }
 
 // handleError sends error.
-func (s *Server) handleError(w http.ResponseWriter, r *http.Request, err error) {
-	s.throw(EventError, &Event{Method: r.Method, Uri: uri(r), Status: 500, Error: err})
+func (h *Handler) handleError(w http.ResponseWriter, r *http.Request, err error) {
+	h.throw(EventError, &Event{Method: r.Method, Uri: uri(r), Status: 500, Error: err})
 
 	w.WriteHeader(500)
 	w.Write([]byte(err.Error()))
 }
 
 // throw invokes event srv if any.
-func (s *Server) throw(event int, ctx interface{}) {
-	if s.listener != nil {
-		s.listener(event, ctx)
+func (h *Handler) throw(event int, ctx interface{}) {
+	if h.listener != nil {
+		h.listener(event, ctx)
 	}
 }
diff --git a/service/http/server_test.go b/service/http/handler_test.go
index 35ee940c..73e587cc 100644
--- a/service/http/server_test.go
+++ b/service/http/handler_test.go
@@ -26,7 +26,7 @@ func get(url string) (string, *http.Response, error) {
 }
 
 func TestServer_Echo(t *testing.T) {
-	st := &Server{
+	st := &Handler{
 		cfg: &Config{
 			MaxRequest: 1024,
 			Uploads: &UploadsConfig{
@@ -60,7 +60,7 @@ func TestServer_Echo(t *testing.T) {
 }
 
 func TestServer_Headers(t *testing.T) {
-	st := &Server{
+	st := &Handler{
 		cfg: &Config{
 			MaxRequest: 1024,
 			Uploads: &UploadsConfig{
@@ -106,7 +106,7 @@ func TestServer_Headers(t *testing.T) {
 }
 
 func TestServer_Cookies(t *testing.T) {
-	st := &Server{
+	st := &Handler{
 		cfg: &Config{
 			MaxRequest: 1024,
 			Uploads: &UploadsConfig{
@@ -156,7 +156,7 @@ func TestServer_Cookies(t *testing.T) {
 }
 
 func TestServer_JsonPayload_POST(t *testing.T) {
-	st := &Server{
+	st := &Handler{
 		cfg: &Config{
 			MaxRequest: 1024,
 			Uploads: &UploadsConfig{
@@ -205,7 +205,7 @@ func TestServer_JsonPayload_POST(t *testing.T) {
 }
 
 func TestServer_JsonPayload_PUT(t *testing.T) {
-	st := &Server{
+	st := &Handler{
 		cfg: &Config{
 			MaxRequest: 1024,
 			Uploads: &UploadsConfig{
@@ -254,7 +254,7 @@ func TestServer_JsonPayload_PUT(t *testing.T) {
 }
 
 func TestServer_JsonPayload_PATCH(t *testing.T) {
-	st := &Server{
+	st := &Handler{
 		cfg: &Config{
 			MaxRequest: 1024,
 			Uploads: &UploadsConfig{
@@ -303,7 +303,7 @@ func TestServer_JsonPayload_PATCH(t *testing.T) {
 }
 
 func TestServer_FormData_POST(t *testing.T) {
-	st := &Server{
+	st := &Handler{
 		cfg: &Config{
 			MaxRequest: 1024,
 			Uploads: &UploadsConfig{
@@ -364,7 +364,7 @@ func TestServer_FormData_POST(t *testing.T) {
 }
 
 func TestServer_FormData_PUT(t *testing.T) {
-	st := &Server{
+	st := &Handler{
 		cfg: &Config{
 			MaxRequest: 1024,
 			Uploads: &UploadsConfig{
@@ -425,7 +425,7 @@ func TestServer_FormData_PUT(t *testing.T) {
 }
 
 func TestServer_FormData_PATCH(t *testing.T) {
-	st := &Server{
+	st := &Handler{
 		cfg: &Config{
 			MaxRequest: 1024,
 			Uploads: &UploadsConfig{
diff --git a/service/http/rpc.go b/service/http/rpc.go
index aa68a826..fc6aa65b 100644
--- a/service/http/rpc.go
+++ b/service/http/rpc.go
@@ -34,7 +34,12 @@ func (rpc *rpcServer) Reset(reset bool, r *string) error {
 	}
 
 	*r = "OK"
-	return rpc.svc.srv.rr.Reset()
+
+	for _, w := range rpc.svc.rr.Workers() {
+		w.Kill()
+	}
+
+	return nil //rpc.svc.rr.Reset()
 }
 
 // Workers returns list of active workers and their stats.
diff --git a/service/http/service.go b/service/http/service.go
index 40eec579..a3773468 100644
--- a/service/http/service.go
+++ b/service/http/service.go
@@ -22,7 +22,7 @@ type Service struct {
 	listeners  []func(event int, ctx interface{})
 	middleware []middleware
 	rr         *roadrunner.Server
-	srv        *Server
+	srv        *Handler
 	http       *http.Server
 }
 
@@ -68,7 +68,7 @@ func (s *Service) Serve() error {
 	rr := roadrunner.NewServer(s.cfg.Workers)
 
 	s.rr = rr
-	s.srv = &Server{cfg: s.cfg, rr: s.rr}
+	s.srv = &Handler{cfg: s.cfg, rr: s.rr}
 	s.http = &http.Server{Addr: s.cfg.Address}
 
 	s.rr.Listen(s.listener)
diff --git a/socket_factory.go b/socket_factory.go
index 70f70f30..43059e8a 100644
--- a/socket_factory.go
+++ b/socket_factory.go
@@ -25,7 +25,7 @@ type SocketFactory struct {
 	relays map[int]chan *goridge.SocketRelay
 }
 
-// NewSocketFactory returns SocketFactory attached to a given socket listener.
+// NewSocketFactory returns SocketFactory attached to a given socket lsn.
 // tout specifies for how long factory should serve for incoming relay connection
 func NewSocketFactory(ls net.Listener, tout time.Duration) *SocketFactory {
 	f := &SocketFactory{
diff --git a/static_pool.go b/static_pool.go
index b043b022..a972b04a 100644
--- a/static_pool.go
+++ b/static_pool.go
@@ -6,6 +6,7 @@ import (
 	"os/exec"
 	"sync"
 	"time"
+	"sync/atomic"
 )
 
 const (
@@ -21,8 +22,8 @@ type StaticPool struct {
 	// worker command creator
 	cmd func() *exec.Cmd
 
-	// listener is optional callback to handle worker create/destruct/error events.
-	listener func(event int, ctx interface{})
+	// lsn is optional callback to handle worker create/destruct/error events.
+	lsn func(event int, ctx interface{})
 
 	// creates and connects to workers
 	factory Factory
@@ -38,6 +39,9 @@ type StaticPool struct {
 
 	// all registered workers
 	workers []*Worker
+
+	// pool is being destroying
+	inDestroy int32
 }
 
 // NewPool creates new worker pool and task multiplexer. StaticPool will initiate with one worker.
@@ -71,7 +75,7 @@ func NewPool(cmd func() *exec.Cmd, factory Factory, cfg Config) (*StaticPool, er
 
 // AddListener attaches pool event watcher.
 func (p *StaticPool) Listen(l func(event int, ctx interface{})) {
-	p.listener = l
+	p.lsn = l
 }
 
 // Config returns associated pool configuration. Immutable.
@@ -110,13 +114,13 @@ func (p *StaticPool) Exec(rqs *Payload) (rsp *Payload, err error) {
 			return nil, err
 		}
 
-		go p.replaceWorker(w, err)
+		go p.destroyWorker(w, err)
 		return nil, err
 	}
 
 	// worker want's to be terminated
 	if rsp.Body == nil && rsp.Context != nil && string(rsp.Context) == StopRequest {
-		go p.replaceWorker(w, err)
+		go p.destroyWorker(w, err)
 		return p.Exec(rqs)
 	}
 
@@ -126,14 +130,18 @@ func (p *StaticPool) Exec(rqs *Payload) (rsp *Payload, err error) {
 
 // Destroy all underlying workers (but let them to complete the task).
 func (p *StaticPool) Destroy() {
+	atomic.AddInt32(&p.inDestroy, 1)
+	defer atomic.AddInt32(&p.inDestroy, -1)
+
 	p.tasks.Wait()
 
 	var wg sync.WaitGroup
 	for _, w := range p.Workers() {
 		wg.Add(1)
+		go w.Stop()
 		go func(w *Worker) {
 			defer wg.Done()
-			p.destroyWorker(w)
+			p.destroyWorker(w, nil)
 		}(w)
 	}
 
@@ -143,33 +151,19 @@ func (p *StaticPool) Destroy() {
 // finds free worker in a given time interval or creates new if allowed.
 func (p *StaticPool) allocateWorker() (w *Worker, err error) {
 	// this loop is required to skip issues with dead workers still being in a ring.
-	for i := int64(0); i < p.cfg.NumWorkers; i++ {
-		select {
-		case w = <-p.free:
-			if w.state.Value() == StateReady {
-				return w, nil
-			} else {
-				go p.replaceWorker(w, w.state.Value())
-				continue
-			}
-		default:
-			// enable timeout handler
-		}
+	select {
+	case w = <-p.free:
+		return w, nil
+	default:
+		// enable timeout handler
+	}
 
-		timeout := time.NewTimer(p.cfg.AllocateTimeout)
-		select {
-		case <-timeout.C:
-			return nil, fmt.Errorf("worker timeout (%s)", p.cfg.AllocateTimeout)
-		case w := <-p.free:
-			timeout.Stop()
-
-			if w.state.Value() == StateReady {
-				return w, nil
-			} else {
-				go p.replaceWorker(w, w.state.Value())
-				continue
-			}
-		}
+	timeout := time.NewTimer(p.cfg.AllocateTimeout)
+	select {
+	case <-timeout.C:
+		return nil, fmt.Errorf("worker timeout (%s)", p.cfg.AllocateTimeout)
+	case w = <-p.free:
+		timeout.Stop()
 	}
 
 	return w, nil
@@ -178,32 +172,13 @@ func (p *StaticPool) allocateWorker() (w *Worker, err error) {
 // release releases or replaces the worker.
 func (p *StaticPool) release(w *Worker) {
 	if p.cfg.MaxJobs != 0 && w.State().NumExecs() >= p.cfg.MaxJobs {
-		go p.replaceWorker(w, p.cfg.MaxJobs)
+		go p.destroyWorker(w, p.cfg.MaxJobs)
 		return
 	}
 
 	p.free <- w
 }
 
-// replaceWorker replaces dead or expired worker with new instance.
-func (p *StaticPool) replaceWorker(w *Worker, caused interface{}) error {
-	go p.destroyWorker(w)
-
-	if nw, err := p.createWorker(); err != nil {
-		p.throw(EventWorkerError, WorkerError{Worker: w, Caused: err})
-
-		if len(p.Workers()) == 0 {
-			// possible situation when major error causes all PHP scripts to die (for example dead DB)
-			p.throw(EventPoolError, fmt.Errorf("all workers are dead"))
-		}
-		return err
-	} else {
-		p.free <- nw
-	}
-
-	return nil
-}
-
 // creates new worker using associated factory. automatically
 // adds worker to the worker list (background)
 func (p *StaticPool) createWorker() (*Worker, error) {
@@ -214,42 +189,16 @@ func (p *StaticPool) createWorker() (*Worker, error) {
 
 	p.throw(EventWorkerConstruct, w)
 
-	go func(w *Worker) {
-		err := w.Wait()
-
-		// worker have died unexpectedly, pool should attempt to replace it with alive version safely
-		if w.state.Value() == StateErrored {
-			if err != nil {
-				p.throw(EventWorkerError, WorkerError{Worker: w, Caused: err})
-			}
-
-			// attempting to replace worker
-			if err := p.replaceWorker(w, err); err != nil {
-				p.throw(EventPoolError, fmt.Errorf("unable to replace: %s", err))
-			}
-		}
-	}(w)
-
 	p.muw.Lock()
-	defer p.muw.Unlock()
-
 	p.workers = append(p.workers, w)
+	p.muw.Unlock()
 
+	go p.watchWorker(w)
 	return w, nil
 }
 
 // destroyWorker destroys workers and removes it from the pool.
-func (p *StaticPool) destroyWorker(w *Worker) {
-	// detaching
-	p.muw.Lock()
-	for i, wc := range p.workers {
-		if wc == w {
-			p.workers = append(p.workers[:i], p.workers[i+1:]...)
-			break
-		}
-	}
-	p.muw.Unlock()
-
+func (p *StaticPool) destroyWorker(w *Worker, caused interface{}) {
 	go w.Stop()
 
 	select {
@@ -267,9 +216,48 @@ func (p *StaticPool) destroyWorker(w *Worker) {
 	}
 }
 
+// watchWorker watches worker state and replaces it if worker fails.
+func (p *StaticPool) watchWorker(w *Worker) {
+	err := w.Wait()
+
+	// detaching
+	p.muw.Lock()
+	for i, wc := range p.workers {
+		if wc == w {
+			p.workers = append(p.workers[:i], p.workers[i+1:]...)
+			break
+		}
+	}
+	p.muw.Unlock()
+
+	// worker have died unexpectedly, pool should attempt to replace it with alive version safely
+	if err != nil {
+		p.throw(EventWorkerError, WorkerError{Worker: w, Caused: err})
+	}
+
+	if !p.destroying() {
+		nw, err := p.createWorker()
+		if err == nil {
+			p.free <- nw
+			return
+		}
+
+		p.throw(EventWorkerError, WorkerError{Worker: w, Caused: err})
+
+		// possible situation when major error causes all PHP scripts to die (for example dead DB)
+		if len(p.Workers()) == 0 {
+			p.throw(EventPoolError, fmt.Errorf("unable to replace (last worker): %s", err))
+		}
+	}
+}
+
+func (p *StaticPool) destroying() bool {
+	return atomic.LoadInt32(&p.inDestroy) != 0
+}
+
 // throw invokes event handler if any.
 func (p *StaticPool) throw(event int, ctx interface{}) {
-	if p.listener != nil {
-		p.listener(event, ctx)
+	if p.lsn != nil {
+		p.lsn(event, ctx)
 	}
 }
diff --git a/worker.go b/worker.go
index 811bda5f..bbb541c8 100644
--- a/worker.go
+++ b/worker.go
@@ -109,6 +109,7 @@ func (w *Worker) Wait() error {
 		w.cmd.Wait()
 	}
 
+	// todo: is it ok?
 	if w.endState.Success() {
 		w.state.set(StateStopped)
 		return nil