From 99c1121d8fd31b7b61b6c22e181dc3c05a3f9f82 Mon Sep 17 00:00:00 2001 From: Wolfy-J Date: Tue, 12 Jun 2018 20:45:24 +0300 Subject: static pool logic update, better watching techniques --- server.go | 10 +- service/http/handler.go | 113 ++++++++++ service/http/handler_test.go | 486 +++++++++++++++++++++++++++++++++++++++++++ service/http/rpc.go | 7 +- service/http/server.go | 113 ---------- service/http/server_test.go | 486 ------------------------------------------- service/http/service.go | 4 +- socket_factory.go | 2 +- static_pool.go | 154 +++++++------- worker.go | 1 + 10 files changed, 685 insertions(+), 691 deletions(-) create mode 100644 service/http/handler.go create mode 100644 service/http/handler_test.go delete mode 100644 service/http/server.go delete mode 100644 service/http/server_test.go 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/handler.go b/service/http/handler.go new file mode 100644 index 00000000..1df00e5a --- /dev/null +++ b/service/http/handler.go @@ -0,0 +1,113 @@ +package http + +import ( + "net/http" + "strconv" + "github.com/spiral/roadrunner" + "github.com/pkg/errors" +) + +const ( + // EventResponse thrown after the request been processed. See Event as payload. + EventResponse = iota + 500 + + // EventError thrown on any non job error provided by road runner server. + EventError +) + +// Event represents singular http response event. +type Event struct { + // Method of the request. + Method string + + // Uri requested by the client. + Uri string + + // Status is response status. + Status int + + // Associated error, if any. + Error error +} + +// 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 Handler struct { + cfg *Config + listener func(event int, ctx interface{}) + rr *roadrunner.Server +} + +// AddListener attaches pool event watcher. +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 (h *Handler) ServeHTTP(w http.ResponseWriter, r *http.Request) { + // validating request size + if h.cfg.MaxRequest != 0 { + if length := r.Header.Get("content-length"); length != "" { + if size, err := strconv.ParseInt(length, 10, 64); err != nil { + h.handleError(w, r, err) + return + } 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, h.cfg.Uploads) + if err != nil { + h.handleError(w, r, err) + return + } + + if err = req.Open(); err != nil { + h.handleError(w, r, err) + return + } + defer req.Close() + + p, err := req.Payload() + if err != nil { + h.handleError(w, r, err) + return + } + + rsp, err := h.rr.Exec(p) + if err != nil { + h.handleError(w, r, err) + return + } + + resp, err := NewResponse(rsp) + if err != nil { + h.handleError(w, r, err) + return + } + + h.handleResponse(req, resp) + resp.Write(w) +} + +// handleResponse triggers response event. +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 (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 (h *Handler) throw(event int, ctx interface{}) { + if h.listener != nil { + h.listener(event, ctx) + } +} diff --git a/service/http/handler_test.go b/service/http/handler_test.go new file mode 100644 index 00000000..73e587cc --- /dev/null +++ b/service/http/handler_test.go @@ -0,0 +1,486 @@ +package http + +import ( + "net/http" + "io/ioutil" + "github.com/spiral/roadrunner" + "testing" + "os" + "github.com/stretchr/testify/assert" + "net/url" + "strings" + "context" + "bytes" +) + +// get request and return body +func get(url string) (string, *http.Response, error) { + r, err := http.Get(url) + if err != nil { + return "", nil, err + } + defer r.Body.Close() + + b, err := ioutil.ReadAll(r.Body) + return string(b), r, err +} + +func TestServer_Echo(t *testing.T) { + st := &Handler{ + cfg: &Config{ + MaxRequest: 1024, + Uploads: &UploadsConfig{ + Dir: os.TempDir(), + Forbid: []string{}, + }, + }, + rr: roadrunner.NewServer(&roadrunner.ServerConfig{ + Command: "php ../../php-src/tests/http/client.php echo pipes", + Relay: "pipes", + Pool: &roadrunner.Config{ + NumWorkers: 1, + AllocateTimeout: 10000000, + DestroyTimeout: 10000000, + }, + }), + } + + assert.NoError(t, st.rr.Start()) + defer st.rr.Stop() + + hs := &http.Server{Addr: ":8077", Handler: st} + defer hs.Shutdown(context.Background()) + + go func() { hs.ListenAndServe() }() + + body, r, err := get("http://localhost:8077/?hello=world") + assert.NoError(t, err) + assert.Equal(t, 200, r.StatusCode) + assert.Equal(t, "WORLD", body) +} + +func TestServer_Headers(t *testing.T) { + st := &Handler{ + cfg: &Config{ + MaxRequest: 1024, + Uploads: &UploadsConfig{ + Dir: os.TempDir(), + Forbid: []string{}, + }, + }, + rr: roadrunner.NewServer(&roadrunner.ServerConfig{ + Command: "php ../../php-src/tests/http/client.php header pipes", + Relay: "pipes", + Pool: &roadrunner.Config{ + NumWorkers: 1, + AllocateTimeout: 10000000, + DestroyTimeout: 10000000, + }, + }), + } + + assert.NoError(t, st.rr.Start()) + defer st.rr.Stop() + + hs := &http.Server{Addr: ":8078", Handler: st,} + defer hs.Shutdown(context.Background()) + + go func() { hs.ListenAndServe() }() + + req, err := http.NewRequest("GET", "http://localhost:8078?hello=world", nil) + assert.NoError(t, err) + + req.Header.Add("input", "sample") + + r, err := http.DefaultClient.Do(req) + assert.NoError(t, err) + defer r.Body.Close() + + b, err := ioutil.ReadAll(r.Body) + assert.NoError(t, err) + + assert.NoError(t, err) + assert.Equal(t, 200, r.StatusCode) + assert.Equal(t, "world", r.Header.Get("Header")) + assert.Equal(t, "SAMPLE", string(b)) +} + +func TestServer_Cookies(t *testing.T) { + st := &Handler{ + cfg: &Config{ + MaxRequest: 1024, + Uploads: &UploadsConfig{ + Dir: os.TempDir(), + Forbid: []string{}, + }, + }, + rr: roadrunner.NewServer(&roadrunner.ServerConfig{ + Command: "php ../../php-src/tests/http/client.php cookie pipes", + Relay: "pipes", + Pool: &roadrunner.Config{ + NumWorkers: 1, + AllocateTimeout: 10000000, + DestroyTimeout: 10000000, + }, + }), + } + + assert.NoError(t, st.rr.Start()) + defer st.rr.Stop() + + hs := &http.Server{Addr: ":8079", Handler: st,} + defer hs.Shutdown(context.Background()) + + go func() { hs.ListenAndServe() }() + + req, err := http.NewRequest("GET", "http://localhost:8079", nil) + assert.NoError(t, err) + + req.AddCookie(&http.Cookie{Name: "input", Value: "input-value"}) + + r, err := http.DefaultClient.Do(req) + assert.NoError(t, err) + defer r.Body.Close() + + b, err := ioutil.ReadAll(r.Body) + assert.NoError(t, err) + + assert.NoError(t, err) + assert.Equal(t, 200, r.StatusCode) + assert.Equal(t, "INPUT-VALUE", string(b)) + + for _, c := range r.Cookies() { + assert.Equal(t, "output", c.Name) + assert.Equal(t, "cookie-output", c.Value) + } +} + +func TestServer_JsonPayload_POST(t *testing.T) { + st := &Handler{ + cfg: &Config{ + MaxRequest: 1024, + Uploads: &UploadsConfig{ + Dir: os.TempDir(), + Forbid: []string{}, + }, + }, + rr: roadrunner.NewServer(&roadrunner.ServerConfig{ + Command: "php ../../php-src/tests/http/client.php payload pipes", + Relay: "pipes", + Pool: &roadrunner.Config{ + NumWorkers: 1, + AllocateTimeout: 10000000, + DestroyTimeout: 10000000, + }, + }), + } + + assert.NoError(t, st.rr.Start()) + defer st.rr.Stop() + + hs := &http.Server{Addr: ":8090", Handler: st,} + defer hs.Shutdown(context.Background()) + + go func() { hs.ListenAndServe() }() + + req, err := http.NewRequest( + "POST", + "http://localhost"+hs.Addr, + bytes.NewBufferString(`{"key":"value"}`), + ) + assert.NoError(t, err) + + req.Header.Add("Content-Type", "application/json") + + r, err := http.DefaultClient.Do(req) + assert.NoError(t, err) + defer r.Body.Close() + + b, err := ioutil.ReadAll(r.Body) + assert.NoError(t, err) + + assert.NoError(t, err) + assert.Equal(t, 200, r.StatusCode) + assert.Equal(t, `{"value":"key"}`, string(b)) +} + +func TestServer_JsonPayload_PUT(t *testing.T) { + st := &Handler{ + cfg: &Config{ + MaxRequest: 1024, + Uploads: &UploadsConfig{ + Dir: os.TempDir(), + Forbid: []string{}, + }, + }, + rr: roadrunner.NewServer(&roadrunner.ServerConfig{ + Command: "php ../../php-src/tests/http/client.php payload pipes", + Relay: "pipes", + Pool: &roadrunner.Config{ + NumWorkers: 1, + AllocateTimeout: 10000000, + DestroyTimeout: 10000000, + }, + }), + } + + assert.NoError(t, st.rr.Start()) + defer st.rr.Stop() + + hs := &http.Server{Addr: ":8081", Handler: st,} + defer hs.Shutdown(context.Background()) + + go func() { hs.ListenAndServe() }() + + req, err := http.NewRequest( + "PUT", + "http://localhost"+hs.Addr, + bytes.NewBufferString(`{"key":"value"}`), + ) + assert.NoError(t, err) + + req.Header.Add("Content-Type", "application/json") + + r, err := http.DefaultClient.Do(req) + assert.NoError(t, err) + defer r.Body.Close() + + b, err := ioutil.ReadAll(r.Body) + assert.NoError(t, err) + + assert.NoError(t, err) + assert.Equal(t, 200, r.StatusCode) + assert.Equal(t, `{"value":"key"}`, string(b)) +} + +func TestServer_JsonPayload_PATCH(t *testing.T) { + st := &Handler{ + cfg: &Config{ + MaxRequest: 1024, + Uploads: &UploadsConfig{ + Dir: os.TempDir(), + Forbid: []string{}, + }, + }, + rr: roadrunner.NewServer(&roadrunner.ServerConfig{ + Command: "php ../../php-src/tests/http/client.php payload pipes", + Relay: "pipes", + Pool: &roadrunner.Config{ + NumWorkers: 1, + AllocateTimeout: 10000000, + DestroyTimeout: 10000000, + }, + }), + } + + assert.NoError(t, st.rr.Start()) + defer st.rr.Stop() + + hs := &http.Server{Addr: ":8082", Handler: st,} + defer hs.Shutdown(context.Background()) + + go func() { hs.ListenAndServe() }() + + req, err := http.NewRequest( + "PATCH", + "http://localhost"+hs.Addr, + bytes.NewBufferString(`{"key":"value"}`), + ) + assert.NoError(t, err) + + req.Header.Add("Content-Type", "application/json") + + r, err := http.DefaultClient.Do(req) + assert.NoError(t, err) + defer r.Body.Close() + + b, err := ioutil.ReadAll(r.Body) + assert.NoError(t, err) + + assert.NoError(t, err) + assert.Equal(t, 200, r.StatusCode) + assert.Equal(t, `{"value":"key"}`, string(b)) +} + +func TestServer_FormData_POST(t *testing.T) { + st := &Handler{ + cfg: &Config{ + MaxRequest: 1024, + Uploads: &UploadsConfig{ + Dir: os.TempDir(), + Forbid: []string{}, + }, + }, + rr: roadrunner.NewServer(&roadrunner.ServerConfig{ + Command: "php ../../php-src/tests/http/client.php data pipes", + Relay: "pipes", + Pool: &roadrunner.Config{ + NumWorkers: 1, + AllocateTimeout: 10000000, + DestroyTimeout: 10000000, + }, + }), + } + + assert.NoError(t, st.rr.Start()) + defer st.rr.Stop() + + hs := &http.Server{Addr: ":8083", Handler: st,} + defer hs.Shutdown(context.Background()) + + go func() { hs.ListenAndServe() }() + + form := url.Values{} + + form.Add("key", "value") + form.Add("name[]", "name1") + form.Add("name[]", "name2") + form.Add("name[]", "name3") + form.Add("arr[x][y][z]", "y") + form.Add("arr[x][y][e]", "f") + form.Add("arr[c]p", "l") + form.Add("arr[c]z", "") + + req, err := http.NewRequest( + "POST", + "http://localhost"+hs.Addr, + strings.NewReader(form.Encode()), + ) + assert.NoError(t, err) + + req.Header.Add("Content-Type", "application/x-www-form-urlencoded") + + r, err := http.DefaultClient.Do(req) + assert.NoError(t, err) + defer r.Body.Close() + + b, err := ioutil.ReadAll(r.Body) + assert.NoError(t, err) + + assert.NoError(t, err) + assert.Equal(t, 200, r.StatusCode) + + assert.Equal(t, `{"arr":{"c":{"p":"l","z":""},"x":{"y":{"e":"f","z":"y"}}},"key":"value","name":["name1","name2","name3"]}`, string(b)) +} + +func TestServer_FormData_PUT(t *testing.T) { + st := &Handler{ + cfg: &Config{ + MaxRequest: 1024, + Uploads: &UploadsConfig{ + Dir: os.TempDir(), + Forbid: []string{}, + }, + }, + rr: roadrunner.NewServer(&roadrunner.ServerConfig{ + Command: "php ../../php-src/tests/http/client.php data pipes", + Relay: "pipes", + Pool: &roadrunner.Config{ + NumWorkers: 1, + AllocateTimeout: 10000000, + DestroyTimeout: 10000000, + }, + }), + } + + assert.NoError(t, st.rr.Start()) + defer st.rr.Stop() + + hs := &http.Server{Addr: ":8084", Handler: st,} + defer hs.Shutdown(context.Background()) + + go func() { hs.ListenAndServe() }() + + form := url.Values{} + + form.Add("key", "value") + form.Add("name[]", "name1") + form.Add("name[]", "name2") + form.Add("name[]", "name3") + form.Add("arr[x][y][z]", "y") + form.Add("arr[x][y][e]", "f") + form.Add("arr[c]p", "l") + form.Add("arr[c]z", "") + + req, err := http.NewRequest( + "PUT", + "http://localhost"+hs.Addr, + strings.NewReader(form.Encode()), + ) + assert.NoError(t, err) + + req.Header.Add("Content-Type", "application/x-www-form-urlencoded") + + r, err := http.DefaultClient.Do(req) + assert.NoError(t, err) + defer r.Body.Close() + + b, err := ioutil.ReadAll(r.Body) + assert.NoError(t, err) + + assert.NoError(t, err) + assert.Equal(t, 200, r.StatusCode) + + assert.Equal(t, `{"arr":{"c":{"p":"l","z":""},"x":{"y":{"e":"f","z":"y"}}},"key":"value","name":["name1","name2","name3"]}`, string(b)) +} + +func TestServer_FormData_PATCH(t *testing.T) { + st := &Handler{ + cfg: &Config{ + MaxRequest: 1024, + Uploads: &UploadsConfig{ + Dir: os.TempDir(), + Forbid: []string{}, + }, + }, + rr: roadrunner.NewServer(&roadrunner.ServerConfig{ + Command: "php ../../php-src/tests/http/client.php data pipes", + Relay: "pipes", + Pool: &roadrunner.Config{ + NumWorkers: 1, + AllocateTimeout: 10000000, + DestroyTimeout: 10000000, + }, + }), + } + + assert.NoError(t, st.rr.Start()) + defer st.rr.Stop() + + hs := &http.Server{Addr: ":8085", Handler: st,} + defer hs.Shutdown(context.Background()) + + go func() { hs.ListenAndServe() }() + + form := url.Values{} + + form.Add("key", "value") + form.Add("name[]", "name1") + form.Add("name[]", "name2") + form.Add("name[]", "name3") + form.Add("arr[x][y][z]", "y") + form.Add("arr[x][y][e]", "f") + form.Add("arr[c]p", "l") + form.Add("arr[c]z", "") + + req, err := http.NewRequest( + "PATCH", + "http://localhost"+hs.Addr, + strings.NewReader(form.Encode()), + ) + assert.NoError(t, err) + + req.Header.Add("Content-Type", "application/x-www-form-urlencoded") + + r, err := http.DefaultClient.Do(req) + assert.NoError(t, err) + defer r.Body.Close() + + b, err := ioutil.ReadAll(r.Body) + assert.NoError(t, err) + + assert.NoError(t, err) + assert.Equal(t, 200, r.StatusCode) + + assert.Equal(t, `{"arr":{"c":{"p":"l","z":""},"x":{"y":{"e":"f","z":"y"}}},"key":"value","name":["name1","name2","name3"]}`, string(b)) +} 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/server.go b/service/http/server.go deleted file mode 100644 index cad2e46d..00000000 --- a/service/http/server.go +++ /dev/null @@ -1,113 +0,0 @@ -package http - -import ( - "net/http" - "strconv" - "github.com/spiral/roadrunner" - "github.com/pkg/errors" -) - -const ( - // EventResponse thrown after the request been processed. See Event as payload. - EventResponse = iota + 500 - - // EventError thrown on any non job error provided by road runner server. - EventError -) - -// Event represents singular http response event. -type Event struct { - // Method of the request. - Method string - - // Uri requested by the client. - Uri string - - // Status is response status. - Status int - - // Associated error, if any. - Error error -} - -// Server 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 { - 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 -} - -// 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) { - // validating request size - if s.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) - return - } else if size > s.cfg.MaxRequest*1024*1024 { - s.handleError(w, r, errors.New("request body max size is exceeded")) - return - } - } - } - - req, err := NewRequest(r, s.cfg.Uploads) - if err != nil { - s.handleError(w, r, err) - return - } - - if err = req.Open(); err != nil { - s.handleError(w, r, err) - return - } - defer req.Close() - - p, err := req.Payload() - if err != nil { - s.handleError(w, r, err) - return - } - - rsp, err := s.rr.Exec(p) - if err != nil { - s.handleError(w, r, err) - return - } - - resp, err := NewResponse(rsp) - if err != nil { - s.handleError(w, r, err) - return - } - - s.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}) -} - -// 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}) - - 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) - } -} diff --git a/service/http/server_test.go b/service/http/server_test.go deleted file mode 100644 index 35ee940c..00000000 --- a/service/http/server_test.go +++ /dev/null @@ -1,486 +0,0 @@ -package http - -import ( - "net/http" - "io/ioutil" - "github.com/spiral/roadrunner" - "testing" - "os" - "github.com/stretchr/testify/assert" - "net/url" - "strings" - "context" - "bytes" -) - -// get request and return body -func get(url string) (string, *http.Response, error) { - r, err := http.Get(url) - if err != nil { - return "", nil, err - } - defer r.Body.Close() - - b, err := ioutil.ReadAll(r.Body) - return string(b), r, err -} - -func TestServer_Echo(t *testing.T) { - st := &Server{ - cfg: &Config{ - MaxRequest: 1024, - Uploads: &UploadsConfig{ - Dir: os.TempDir(), - Forbid: []string{}, - }, - }, - rr: roadrunner.NewServer(&roadrunner.ServerConfig{ - Command: "php ../../php-src/tests/http/client.php echo pipes", - Relay: "pipes", - Pool: &roadrunner.Config{ - NumWorkers: 1, - AllocateTimeout: 10000000, - DestroyTimeout: 10000000, - }, - }), - } - - assert.NoError(t, st.rr.Start()) - defer st.rr.Stop() - - hs := &http.Server{Addr: ":8077", Handler: st} - defer hs.Shutdown(context.Background()) - - go func() { hs.ListenAndServe() }() - - body, r, err := get("http://localhost:8077/?hello=world") - assert.NoError(t, err) - assert.Equal(t, 200, r.StatusCode) - assert.Equal(t, "WORLD", body) -} - -func TestServer_Headers(t *testing.T) { - st := &Server{ - cfg: &Config{ - MaxRequest: 1024, - Uploads: &UploadsConfig{ - Dir: os.TempDir(), - Forbid: []string{}, - }, - }, - rr: roadrunner.NewServer(&roadrunner.ServerConfig{ - Command: "php ../../php-src/tests/http/client.php header pipes", - Relay: "pipes", - Pool: &roadrunner.Config{ - NumWorkers: 1, - AllocateTimeout: 10000000, - DestroyTimeout: 10000000, - }, - }), - } - - assert.NoError(t, st.rr.Start()) - defer st.rr.Stop() - - hs := &http.Server{Addr: ":8078", Handler: st,} - defer hs.Shutdown(context.Background()) - - go func() { hs.ListenAndServe() }() - - req, err := http.NewRequest("GET", "http://localhost:8078?hello=world", nil) - assert.NoError(t, err) - - req.Header.Add("input", "sample") - - r, err := http.DefaultClient.Do(req) - assert.NoError(t, err) - defer r.Body.Close() - - b, err := ioutil.ReadAll(r.Body) - assert.NoError(t, err) - - assert.NoError(t, err) - assert.Equal(t, 200, r.StatusCode) - assert.Equal(t, "world", r.Header.Get("Header")) - assert.Equal(t, "SAMPLE", string(b)) -} - -func TestServer_Cookies(t *testing.T) { - st := &Server{ - cfg: &Config{ - MaxRequest: 1024, - Uploads: &UploadsConfig{ - Dir: os.TempDir(), - Forbid: []string{}, - }, - }, - rr: roadrunner.NewServer(&roadrunner.ServerConfig{ - Command: "php ../../php-src/tests/http/client.php cookie pipes", - Relay: "pipes", - Pool: &roadrunner.Config{ - NumWorkers: 1, - AllocateTimeout: 10000000, - DestroyTimeout: 10000000, - }, - }), - } - - assert.NoError(t, st.rr.Start()) - defer st.rr.Stop() - - hs := &http.Server{Addr: ":8079", Handler: st,} - defer hs.Shutdown(context.Background()) - - go func() { hs.ListenAndServe() }() - - req, err := http.NewRequest("GET", "http://localhost:8079", nil) - assert.NoError(t, err) - - req.AddCookie(&http.Cookie{Name: "input", Value: "input-value"}) - - r, err := http.DefaultClient.Do(req) - assert.NoError(t, err) - defer r.Body.Close() - - b, err := ioutil.ReadAll(r.Body) - assert.NoError(t, err) - - assert.NoError(t, err) - assert.Equal(t, 200, r.StatusCode) - assert.Equal(t, "INPUT-VALUE", string(b)) - - for _, c := range r.Cookies() { - assert.Equal(t, "output", c.Name) - assert.Equal(t, "cookie-output", c.Value) - } -} - -func TestServer_JsonPayload_POST(t *testing.T) { - st := &Server{ - cfg: &Config{ - MaxRequest: 1024, - Uploads: &UploadsConfig{ - Dir: os.TempDir(), - Forbid: []string{}, - }, - }, - rr: roadrunner.NewServer(&roadrunner.ServerConfig{ - Command: "php ../../php-src/tests/http/client.php payload pipes", - Relay: "pipes", - Pool: &roadrunner.Config{ - NumWorkers: 1, - AllocateTimeout: 10000000, - DestroyTimeout: 10000000, - }, - }), - } - - assert.NoError(t, st.rr.Start()) - defer st.rr.Stop() - - hs := &http.Server{Addr: ":8090", Handler: st,} - defer hs.Shutdown(context.Background()) - - go func() { hs.ListenAndServe() }() - - req, err := http.NewRequest( - "POST", - "http://localhost"+hs.Addr, - bytes.NewBufferString(`{"key":"value"}`), - ) - assert.NoError(t, err) - - req.Header.Add("Content-Type", "application/json") - - r, err := http.DefaultClient.Do(req) - assert.NoError(t, err) - defer r.Body.Close() - - b, err := ioutil.ReadAll(r.Body) - assert.NoError(t, err) - - assert.NoError(t, err) - assert.Equal(t, 200, r.StatusCode) - assert.Equal(t, `{"value":"key"}`, string(b)) -} - -func TestServer_JsonPayload_PUT(t *testing.T) { - st := &Server{ - cfg: &Config{ - MaxRequest: 1024, - Uploads: &UploadsConfig{ - Dir: os.TempDir(), - Forbid: []string{}, - }, - }, - rr: roadrunner.NewServer(&roadrunner.ServerConfig{ - Command: "php ../../php-src/tests/http/client.php payload pipes", - Relay: "pipes", - Pool: &roadrunner.Config{ - NumWorkers: 1, - AllocateTimeout: 10000000, - DestroyTimeout: 10000000, - }, - }), - } - - assert.NoError(t, st.rr.Start()) - defer st.rr.Stop() - - hs := &http.Server{Addr: ":8081", Handler: st,} - defer hs.Shutdown(context.Background()) - - go func() { hs.ListenAndServe() }() - - req, err := http.NewRequest( - "PUT", - "http://localhost"+hs.Addr, - bytes.NewBufferString(`{"key":"value"}`), - ) - assert.NoError(t, err) - - req.Header.Add("Content-Type", "application/json") - - r, err := http.DefaultClient.Do(req) - assert.NoError(t, err) - defer r.Body.Close() - - b, err := ioutil.ReadAll(r.Body) - assert.NoError(t, err) - - assert.NoError(t, err) - assert.Equal(t, 200, r.StatusCode) - assert.Equal(t, `{"value":"key"}`, string(b)) -} - -func TestServer_JsonPayload_PATCH(t *testing.T) { - st := &Server{ - cfg: &Config{ - MaxRequest: 1024, - Uploads: &UploadsConfig{ - Dir: os.TempDir(), - Forbid: []string{}, - }, - }, - rr: roadrunner.NewServer(&roadrunner.ServerConfig{ - Command: "php ../../php-src/tests/http/client.php payload pipes", - Relay: "pipes", - Pool: &roadrunner.Config{ - NumWorkers: 1, - AllocateTimeout: 10000000, - DestroyTimeout: 10000000, - }, - }), - } - - assert.NoError(t, st.rr.Start()) - defer st.rr.Stop() - - hs := &http.Server{Addr: ":8082", Handler: st,} - defer hs.Shutdown(context.Background()) - - go func() { hs.ListenAndServe() }() - - req, err := http.NewRequest( - "PATCH", - "http://localhost"+hs.Addr, - bytes.NewBufferString(`{"key":"value"}`), - ) - assert.NoError(t, err) - - req.Header.Add("Content-Type", "application/json") - - r, err := http.DefaultClient.Do(req) - assert.NoError(t, err) - defer r.Body.Close() - - b, err := ioutil.ReadAll(r.Body) - assert.NoError(t, err) - - assert.NoError(t, err) - assert.Equal(t, 200, r.StatusCode) - assert.Equal(t, `{"value":"key"}`, string(b)) -} - -func TestServer_FormData_POST(t *testing.T) { - st := &Server{ - cfg: &Config{ - MaxRequest: 1024, - Uploads: &UploadsConfig{ - Dir: os.TempDir(), - Forbid: []string{}, - }, - }, - rr: roadrunner.NewServer(&roadrunner.ServerConfig{ - Command: "php ../../php-src/tests/http/client.php data pipes", - Relay: "pipes", - Pool: &roadrunner.Config{ - NumWorkers: 1, - AllocateTimeout: 10000000, - DestroyTimeout: 10000000, - }, - }), - } - - assert.NoError(t, st.rr.Start()) - defer st.rr.Stop() - - hs := &http.Server{Addr: ":8083", Handler: st,} - defer hs.Shutdown(context.Background()) - - go func() { hs.ListenAndServe() }() - - form := url.Values{} - - form.Add("key", "value") - form.Add("name[]", "name1") - form.Add("name[]", "name2") - form.Add("name[]", "name3") - form.Add("arr[x][y][z]", "y") - form.Add("arr[x][y][e]", "f") - form.Add("arr[c]p", "l") - form.Add("arr[c]z", "") - - req, err := http.NewRequest( - "POST", - "http://localhost"+hs.Addr, - strings.NewReader(form.Encode()), - ) - assert.NoError(t, err) - - req.Header.Add("Content-Type", "application/x-www-form-urlencoded") - - r, err := http.DefaultClient.Do(req) - assert.NoError(t, err) - defer r.Body.Close() - - b, err := ioutil.ReadAll(r.Body) - assert.NoError(t, err) - - assert.NoError(t, err) - assert.Equal(t, 200, r.StatusCode) - - assert.Equal(t, `{"arr":{"c":{"p":"l","z":""},"x":{"y":{"e":"f","z":"y"}}},"key":"value","name":["name1","name2","name3"]}`, string(b)) -} - -func TestServer_FormData_PUT(t *testing.T) { - st := &Server{ - cfg: &Config{ - MaxRequest: 1024, - Uploads: &UploadsConfig{ - Dir: os.TempDir(), - Forbid: []string{}, - }, - }, - rr: roadrunner.NewServer(&roadrunner.ServerConfig{ - Command: "php ../../php-src/tests/http/client.php data pipes", - Relay: "pipes", - Pool: &roadrunner.Config{ - NumWorkers: 1, - AllocateTimeout: 10000000, - DestroyTimeout: 10000000, - }, - }), - } - - assert.NoError(t, st.rr.Start()) - defer st.rr.Stop() - - hs := &http.Server{Addr: ":8084", Handler: st,} - defer hs.Shutdown(context.Background()) - - go func() { hs.ListenAndServe() }() - - form := url.Values{} - - form.Add("key", "value") - form.Add("name[]", "name1") - form.Add("name[]", "name2") - form.Add("name[]", "name3") - form.Add("arr[x][y][z]", "y") - form.Add("arr[x][y][e]", "f") - form.Add("arr[c]p", "l") - form.Add("arr[c]z", "") - - req, err := http.NewRequest( - "PUT", - "http://localhost"+hs.Addr, - strings.NewReader(form.Encode()), - ) - assert.NoError(t, err) - - req.Header.Add("Content-Type", "application/x-www-form-urlencoded") - - r, err := http.DefaultClient.Do(req) - assert.NoError(t, err) - defer r.Body.Close() - - b, err := ioutil.ReadAll(r.Body) - assert.NoError(t, err) - - assert.NoError(t, err) - assert.Equal(t, 200, r.StatusCode) - - assert.Equal(t, `{"arr":{"c":{"p":"l","z":""},"x":{"y":{"e":"f","z":"y"}}},"key":"value","name":["name1","name2","name3"]}`, string(b)) -} - -func TestServer_FormData_PATCH(t *testing.T) { - st := &Server{ - cfg: &Config{ - MaxRequest: 1024, - Uploads: &UploadsConfig{ - Dir: os.TempDir(), - Forbid: []string{}, - }, - }, - rr: roadrunner.NewServer(&roadrunner.ServerConfig{ - Command: "php ../../php-src/tests/http/client.php data pipes", - Relay: "pipes", - Pool: &roadrunner.Config{ - NumWorkers: 1, - AllocateTimeout: 10000000, - DestroyTimeout: 10000000, - }, - }), - } - - assert.NoError(t, st.rr.Start()) - defer st.rr.Stop() - - hs := &http.Server{Addr: ":8085", Handler: st,} - defer hs.Shutdown(context.Background()) - - go func() { hs.ListenAndServe() }() - - form := url.Values{} - - form.Add("key", "value") - form.Add("name[]", "name1") - form.Add("name[]", "name2") - form.Add("name[]", "name3") - form.Add("arr[x][y][z]", "y") - form.Add("arr[x][y][e]", "f") - form.Add("arr[c]p", "l") - form.Add("arr[c]z", "") - - req, err := http.NewRequest( - "PATCH", - "http://localhost"+hs.Addr, - strings.NewReader(form.Encode()), - ) - assert.NoError(t, err) - - req.Header.Add("Content-Type", "application/x-www-form-urlencoded") - - r, err := http.DefaultClient.Do(req) - assert.NoError(t, err) - defer r.Body.Close() - - b, err := ioutil.ReadAll(r.Body) - assert.NoError(t, err) - - assert.NoError(t, err) - assert.Equal(t, 200, r.StatusCode) - - assert.Equal(t, `{"arr":{"c":{"p":"l","z":""},"x":{"y":{"e":"f","z":"y"}}},"key":"value","name":["name1","name2","name3"]}`, string(b)) -} 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 -- cgit v1.2.3