diff options
| -rw-r--r-- | service/http/errors.go | 25 | ||||
| -rw-r--r-- | service/http/errors_windows.go | 27 | ||||
| -rw-r--r-- | service/http/handler.go | 27 | ||||
| -rw-r--r-- | service/http/response.go | 6 | ||||
| -rw-r--r-- | static_pool.go | 12 |
5 files changed, 80 insertions, 17 deletions
diff --git a/service/http/errors.go b/service/http/errors.go new file mode 100644 index 00000000..fb8762ef --- /dev/null +++ b/service/http/errors.go @@ -0,0 +1,25 @@ +// +build !windows + +package http + +import ( + "errors" + "net" + "os" + "syscall" +) + +// Broken pipe +var errEPIPE = errors.New("EPIPE(32) -> connection reset by peer") + +// handleWriteError just check if error was caused by aborted connection on linux +func handleWriteError(err error) error { + if netErr, ok2 := err.(*net.OpError); ok2 { + if syscallErr, ok3 := netErr.Err.(*os.SyscallError); ok3 { + if syscallErr.Err == syscall.EPIPE { + return errEPIPE + } + } + } + return err +} diff --git a/service/http/errors_windows.go b/service/http/errors_windows.go new file mode 100644 index 00000000..3d0ba04c --- /dev/null +++ b/service/http/errors_windows.go @@ -0,0 +1,27 @@ +// +build windows + +package http + +import ( + "errors" + "net" + "os" + "syscall" +) + +//Software caused connection abort. +//An established connection was aborted by the software in your host computer, +//possibly due to a data transmission time-out or protocol error. +var errEPIPE = errors.New("WSAECONNABORTED (10053) -> an established connection was aborted by peer") + +// handleWriteError just check if error was caused by aborted connection on windows +func handleWriteError(err error) error { + if netErr, ok2 := err.(*net.OpError); ok2 { + if syscallErr, ok3 := netErr.Err.(*os.SyscallError); ok3 { + if syscallErr.Err == syscall.WSAECONNABORTED { + return errEPIPE + } + } + } + return err +} diff --git a/service/http/handler.go b/service/http/handler.go index ab7c382b..eca05483 100644 --- a/service/http/handler.go +++ b/service/http/handler.go @@ -1,15 +1,17 @@ package http import ( - "github.com/pkg/errors" - "github.com/sirupsen/logrus" - "github.com/spiral/roadrunner" + "fmt" "net" "net/http" "strconv" "strings" "sync" "time" + + "github.com/pkg/errors" + "github.com/sirupsen/logrus" + "github.com/spiral/roadrunner" ) const ( @@ -130,13 +132,22 @@ func (h *Handler) ServeHTTP(w http.ResponseWriter, r *http.Request) { // handleError sends error. func (h *Handler) handleError(w http.ResponseWriter, r *http.Request, err error, start time.Time) { - h.throw(EventError, &ErrorEvent{Request: r, Error: err, start: start, elapsed: time.Since(start)}) - - w.WriteHeader(500) - _, err = w.Write([]byte(err.Error())) - if err != nil { + // if pipe is broken, there is no sense to write the header + // in this case we just report about error + if err == errEPIPE { h.throw(EventError, &ErrorEvent{Request: r, Error: err, start: start, elapsed: time.Since(start)}) + return } + // ResponseWriter is ok, write the error code + w.WriteHeader(500) + _, err2 := w.Write([]byte(err.Error())) + // error during the writing to the ResponseWriter + if err2 != nil { + // concat original error with ResponseWriter error + h.throw(EventError, &ErrorEvent{Request: r, Error: errors.New(fmt.Sprintf("error: %v, during handle this error, ResponseWriter error occurred: %v", err, err2)), start: start, elapsed: time.Since(start)}) + return + } + h.throw(EventError, &ErrorEvent{Request: r, Error: err, start: start, elapsed: time.Since(start)}) } // handleResponse triggers response event. diff --git a/service/http/response.go b/service/http/response.go index 0942b3d2..f34754be 100644 --- a/service/http/response.go +++ b/service/http/response.go @@ -1,14 +1,16 @@ package http import ( - json "github.com/json-iterator/go" "io" "net/http" "strings" + json "github.com/json-iterator/go" + "github.com/spiral/roadrunner" ) + // Response handles PSR7 response logic. type Response struct { // Status contains response status. @@ -57,7 +59,7 @@ func (r *Response) Write(w http.ResponseWriter) error { if data, ok := r.body.([]byte); ok { _, err := w.Write(data) if err != nil { - return err + return handleWriteError(err) } } diff --git a/static_pool.go b/static_pool.go index ac9c2529..c7cc6517 100644 --- a/static_pool.go +++ b/static_pool.go @@ -2,11 +2,12 @@ package roadrunner import ( "fmt" - "github.com/pkg/errors" "os/exec" "sync" "sync/atomic" "time" + + "github.com/pkg/errors" ) const ( @@ -26,7 +27,7 @@ type StaticPool struct { factory Factory // active task executions - tmu *sync.Mutex + tmu sync.Mutex tasks sync.WaitGroup // workers circular allocation buf @@ -36,13 +37,13 @@ type StaticPool struct { numDead int64 // protects state of worker list, does not affect allocation - muw *sync.RWMutex + muw sync.RWMutex // all registered workers workers []*Worker // invalid declares set of workers to be removed from the pool. - remove *sync.Map + remove sync.Map // pool is being destroyed inDestroy int32 @@ -66,9 +67,6 @@ func NewPool(cmd func() *exec.Cmd, factory Factory, cfg Config) (*StaticPool, er workers: make([]*Worker, 0, cfg.NumWorkers), free: make(chan *Worker, cfg.NumWorkers), destroy: make(chan interface{}), - tmu: &sync.Mutex{}, - remove: &sync.Map{}, - muw: &sync.RWMutex{}, } // constant number of workers simplify logic |