diff options
| -rw-r--r-- | server.go | 10 | ||||
| -rw-r--r-- | service/http/handler.go (renamed from service/http/server.go) | 48 | ||||
| -rw-r--r-- | service/http/handler_test.go (renamed from service/http/server_test.go) | 18 | ||||
| -rw-r--r-- | service/http/rpc.go | 7 | ||||
| -rw-r--r-- | service/http/service.go | 4 | ||||
| -rw-r--r-- | socket_factory.go | 2 | ||||
| -rw-r--r-- | static_pool.go | 154 | ||||
| -rw-r--r-- | worker.go | 1 |
8 files changed, 119 insertions, 125 deletions
@@ -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) } } @@ -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 |