worker: add ability to cancel task when queue is full

This commit is contained in:
criyle 2021-11-19 14:34:00 -08:00
parent 6643a592ba
commit dbcfc79614
7 changed files with 67 additions and 28 deletions

View File

@ -5,7 +5,3 @@ cmd = "go build -o ./tmp/executorserver ./cmd/executorserver"
full_bin = "tmp/executorserver -enable-grpc -enable-debug"
include_ext = ["go"]
delay = 1000
send_interrupt = true
kill_delay = 3000

View File

@ -56,7 +56,8 @@ func (e *execServer) Exec(ctx context.Context, req *pb.Request) (*pb.Response, e
return nil, fmt.Errorf("stream in / out are not available for exec request")
}
e.logger.Sugar().Debugf("request: %+v", r)
rt := <-e.worker.Submit(ctx, r)
rtCh, _ := e.worker.Submit(ctx, r)
rt := <-rtCh
e.logger.Sugar().Debugf("response: %+v", rt)
if rt.Error != nil {
return nil, rt.Error

View File

@ -65,7 +65,8 @@ func (h *handle) handleRun(c *gin.Context) {
return
}
h.logger.Sugar().Debugf("request: %+v", r)
rt := <-h.worker.Submit(c.Request.Context(), r)
rtCh, _ := h.worker.Submit(c.Request.Context(), r)
rt := <-rtCh
h.logger.Sugar().Debugf("response: %+v", rt)
if rt.Error != nil {
c.Error(rt.Error)

View File

@ -78,9 +78,12 @@ func (h *wsHandle) handleWS(c *gin.Context) {
ctx, cancel := context.WithCancel(baseCtx)
if err := cm.Add(r.RequestID, cancel); err != nil {
resultCh <- model.Response{
select {
case <-baseCtx.Done():
case resultCh <- model.Response{
RequestID: req.RequestID,
ErrorMsg: err.Error(),
}:
}
cancel()
h.logger.Sugar().Debugf("ws request error: %v", err)
@ -89,18 +92,38 @@ func (h *wsHandle) handleWS(c *gin.Context) {
go func() {
defer cm.Remove(r.RequestID)
h.logger.Sugar().Debugf("ws request: %+v", r)
ret := <-h.worker.Submit(ctx, r)
retCh, started := h.worker.Submit(ctx, r)
var ret worker.Response
select {
case <-baseCtx.Done(): // if connection lost
return
case <-ctx.Done(): // if context cancelled by cancelling request
select {
case <-started: // if started, wait for result
ret = <-retCh
default: // not started
ret = worker.Response{
RequestID: r.RequestID,
Error: fmt.Errorf("request cancelled before execute"),
}
}
case ret = <-retCh:
}
h.logger.Sugar().Debugf("ws response: %+v", ret)
resp, err := model.ConvertResponse(ret, false)
if err != nil {
resultCh <- model.Response{
resp = model.Response{
RequestID: r.RequestID,
ErrorMsg: resp.ErrorMsg,
}
return
}
resultCh <- resp
select {
case <-baseCtx.Done():
case resultCh <- resp:
}
}()
return nil
}

View File

@ -116,7 +116,8 @@ func Exec(e *C.char) *C.char {
if err != nil {
return nil
}
rt := <-work.Submit(context.TODO(), r)
rtCh, _ := work.Submit(context.TODO(), r)
rt := <-rtCh
ret, err := model.ConvertResponse(rt, true)
if err != nil {
return nil

View File

@ -61,7 +61,7 @@ func copyOutAndCollect(m Environment, c *Cmd, ptc []pipeCollector, newStoreFile
// check regular file
if stat.Mode()&os.ModeType != 0 {
t = ErrCopyOutNotRegularFile
return fmt.Errorf("%s: not a regular file %d", n.Name, stat.Mode()&os.ModeType)
return fmt.Errorf("%s: not a regular file: %v", n.Name, stat.Mode())
}
// check size limit
s := stat.Size()

View File

@ -37,7 +37,7 @@ type Config struct {
// Worker defines interface for executor
type Worker interface {
Start()
Submit(context.Context, *Request) <-chan Response
Submit(context.Context, *Request) (<-chan Response, <-chan struct{})
Execute(context.Context, *Request) <-chan Response
Shutdown()
}
@ -67,6 +67,7 @@ type worker struct {
type workRequest struct {
*Request
context.Context
started chan<- struct{}
resultCh chan<- Response
}
@ -99,14 +100,24 @@ func (w *worker) Start() {
}
// Submit submits a single request
func (w *worker) Submit(ctx context.Context, req *Request) <-chan Response {
func (w *worker) Submit(ctx context.Context, req *Request) (<-chan Response, <-chan struct{}) {
ch := make(chan Response, 1)
w.workCh <- workRequest{
started := make(chan struct{})
select {
case w.workCh <- workRequest{
Request: req,
Context: ctx,
started: started,
resultCh: ch,
}:
default:
close(started)
ch <- Response{
RequestID: req.RequestID,
Error: fmt.Errorf("worker queue is full"),
}
}
return ch
return ch, started
}
// Execute will execute the request in new goroutine (bypass the parallelism limit)
@ -115,12 +126,7 @@ func (w *worker) Execute(ctx context.Context, req *Request) <-chan Response {
w.wg.Add(1)
go func() {
defer w.wg.Done()
wq := workRequest{
Request: req,
Context: ctx,
resultCh: ch,
}
w.workDoCmd(wq)
ch <- w.workDoCmd(ctx, req)
}()
return ch
}
@ -141,25 +147,36 @@ func (w *worker) loop() {
if !ok {
return
}
w.workDoCmd(req)
close(req.started)
select {
case <-req.Context.Done():
req.resultCh <- Response{
RequestID: req.RequestID,
Error: fmt.Errorf("cancelled before execute"),
}
default:
req.resultCh <- w.workDoCmd(req.Context, req.Request)
}
case <-w.done:
return
}
}
}
func (w *worker) workDoCmd(req workRequest) {
func (w *worker) workDoCmd(ctx context.Context, req *Request) Response {
var rt Response
if len(req.Cmd) == 1 {
rt = w.workDoSingle(req.Context, req.Cmd[0])
rt = w.workDoSingle(ctx, req.Cmd[0])
} else {
rt = w.workDoGroup(req.Context, req.Cmd, req.PipeMapping)
rt = w.workDoGroup(ctx, req.Cmd, req.PipeMapping)
}
rt.RequestID = req.RequestID
if w.execObserver != nil {
w.execObserver(rt)
}
req.resultCh <- rt
return rt
}
func (w *worker) workDoSingle(ctx context.Context, rc Cmd) (rt Response) {