From 4a2dbe7ea9dd0f1466153b6045e2615de2d0ca4d Mon Sep 17 00:00:00 2001 From: Gensuke Toshiro Date: Mon, 10 Aug 2020 11:51:44 +0900 Subject: [PATCH 1/2] Make concurrent requests whenever possible --- proxy/merging.go | 169 +++++++++++++++++++++++++++++++---------------- 1 file changed, 112 insertions(+), 57 deletions(-) diff --git a/proxy/merging.go b/proxy/merging.go index 8949495a2..dc1ef3192 100644 --- a/proxy/merging.go +++ b/proxy/merging.go @@ -82,6 +82,25 @@ func parallelMerge(timeout time.Duration, rc ResponseCombiner, next ...Proxy) Pr var reMergeKey = regexp.MustCompile(`\{\{\.Resp(\d+)_([\d\w-_\.]+)\}\}`) func sequentialMerge(patterns []string, timeout time.Duration, rc ResponseCombiner, next ...Proxy) Proxy { + deps := make([][]int, len(patterns)) + matches := make([][][]string, len(patterns)) + for i, pattern := range patterns { + matches[i] = reMergeKey.FindAllStringSubmatch(pattern, -1) + deps[i] = make([]int, 0, len(matches)) + for _, match := range matches[i] { + if rNum, err := strconv.Atoi(match[1]); err == nil { + var found bool + for _, j := range deps[i] { + if j == rNum { + found = true + } + } + if !found { + deps[i] = append(deps[i], rNum) + } + } + } + } return func(ctx context.Context, request *Request) (*Response, error) { localCtx, cancel := context.WithTimeout(ctx, timeout) @@ -89,75 +108,111 @@ func sequentialMerge(patterns []string, timeout time.Duration, rc ResponseCombin out := make(chan *Response, 1) errCh := make(chan error, 1) + type partResult struct { i int; err error } + ch := make(chan partResult, len(next)) + done := make([]chan struct {}, len(next)) + for i := range next { + done[i] = make(chan struct {}, 1) + } + acc := newIncrementalMergeAccumulator(len(next), rc) - TxLoop: - for i, n := range next { - if i > 0 { - for _, match := range reMergeKey.FindAllStringSubmatch(patterns[i], -1) { - if len(match) > 1 { - rNum, err := strconv.Atoi(match[1]) - if err != nil || rNum >= i || parts[rNum] == nil { - continue - } - key := "Resp" + match[1] + "_" + match[2] - - var v interface{} - var ok bool - - data := parts[rNum].Data - keys := strings.Split(match[2], ".") - if len(keys) > 1 { - for _, k := range keys[:len(keys)-1] { - v, ok = data[k] - if !ok { - break - } - switch clean := v.(type) { - case map[string]interface{}: - data = clean - default: - break + + for i := range next { + go func(i int) { + n := next[i] + for _, j := range deps[i] { + select{ + case <-done[j]: + case <-localCtx.Done(): + return + } + } + var req *Request = request + if i > 0 { + req = CloneRequest(request) + for _, match := range matches[i] { + if len(match) > 1 { + rNum, err := strconv.Atoi(match[1]) + if err != nil || rNum >= i || parts[rNum] == nil { + continue + } + key := "Resp" + match[1] + "_" + match[2] + + var v interface{} + var ok bool + + data := parts[rNum].Data + keys := strings.Split(match[2], ".") + if len(keys) > 1 { + for _, k := range keys[:len(keys)-1] { + v, ok = data[k] + if !ok { + break + } + switch clean := v.(type) { + case map[string]interface{}: + data = clean + default: + break + } } } - } - v, ok = data[keys[len(keys)-1]] - if !ok { - continue - } - switch clean := v.(type) { - case string: - request.Params[key] = clean - case int: - request.Params[key] = strconv.Itoa(clean) - case float64: - request.Params[key] = strconv.FormatFloat(clean, 'E', -1, 32) - case bool: - request.Params[key] = strconv.FormatBool(clean) - default: - request.Params[key] = fmt.Sprintf("%v", v) + v, ok = data[keys[len(keys)-1]] + if !ok { + continue + } + switch clean := v.(type) { + case string: + req.Params[key] = clean + case int: + req.Params[key] = strconv.Itoa(clean) + case float64: + req.Params[key] = strconv.FormatFloat(clean, 'E', -1, 32) + case bool: + req.Params[key] = strconv.FormatBool(clean) + default: + req.Params[key] = fmt.Sprintf("%v", v) + } } } } - } - requestPart(localCtx, n, request, out, errCh) - select { - case err := <-errCh: - if i == 0 { - cancel() - return nil, err + requestPart(localCtx, n, req, out, errCh) + select { + case err := <-errCh: + ch <- partResult{i, err} + case response := <-out: + if !response.IsComplete { + cancel() + return + } + parts[i] = response + close(done[i]) + ch <- partResult{i, nil} } - acc.Merge(nil, err) - break TxLoop - case response := <-out: - acc.Merge(response, nil) - if !response.IsComplete { - break TxLoop + }(i) + } + + for _ = range next { + select { + case res := <-ch: + if i, err := res.i, res.err; err != nil { + if i == 0 { + cancel() + return nil, err + } + acc.Merge(nil, err) } - parts[i] = response + case <-localCtx.Done(): + acc.Merge(nil, localCtx.Err()) + break } } + for _, part := range parts { + acc.Merge(part, nil) + } + result, err := acc.Result() cancel() return result, err From 06bd8e2453dbcf049839560586a70230f40b8928 Mon Sep 17 00:00:00 2001 From: Gensuke Toshiro Date: Tue, 11 Aug 2020 18:01:01 +0900 Subject: [PATCH 2/2] Halt when parent request failed --- proxy/merging.go | 37 +++++++++++++++++++++++++++++-------- 1 file changed, 29 insertions(+), 8 deletions(-) diff --git a/proxy/merging.go b/proxy/merging.go index dc1ef3192..7f415173b 100644 --- a/proxy/merging.go +++ b/proxy/merging.go @@ -81,6 +81,17 @@ func parallelMerge(timeout time.Duration, rc ResponseCombiner, next ...Proxy) Pr var reMergeKey = regexp.MustCompile(`\{\{\.Resp(\d+)_([\d\w-_\.]+)\}\}`) +func isBlocking(i int, deps [][]int) bool { + for _, dep := range deps { + for _, j := range dep { + if i == j { + return true + } + } + } + return false +} + func sequentialMerge(patterns []string, timeout time.Duration, rc ResponseCombiner, next ...Proxy) Proxy { deps := make([][]int, len(patterns)) matches := make([][][]string, len(patterns)) @@ -182,12 +193,12 @@ func sequentialMerge(patterns []string, timeout time.Duration, rc ResponseCombin case err := <-errCh: ch <- partResult{i, err} case response := <-out: + parts[i] = response + close(done[i]) if !response.IsComplete { cancel() return } - parts[i] = response - close(done[i]) ch <- partResult{i, nil} } }(i) @@ -197,20 +208,21 @@ func sequentialMerge(patterns []string, timeout time.Duration, rc ResponseCombin select { case res := <-ch: if i, err := res.i, res.err; err != nil { - if i == 0 { + acc.Merge(nil, err) + if isBlocking(i, deps) { cancel() - return nil, err + break } - acc.Merge(nil, err) } case <-localCtx.Done(): - acc.Merge(nil, localCtx.Err()) break } } for _, part := range parts { - acc.Merge(part, nil) + if part != nil { + acc.Merge(part, nil) + } } result, err := acc.Result() @@ -256,7 +268,14 @@ func (i *incrementalMergeAccumulator) Merge(res *Response, err error) { func (i *incrementalMergeAccumulator) Result() (*Response, error) { if i.data == nil { - return &Response{Data: make(map[string]interface{}, 0), IsComplete: false}, newMergeError(i.errs) + err := newMergeError(i.errs) + + // none succeeded + if len(i.errs) == 1 { + return nil, err + } + + return &Response{Data: make(map[string]interface{}, 0), IsComplete: false}, err } if i.pending != 0 || len(i.errs) != 0 { @@ -290,6 +309,8 @@ func requestPart(ctx context.Context, next Proxy, request *Request, out chan<- * func newMergeError(errs []error) error { if len(errs) == 0 { return nil + } else if len(errs) == 1 { + return errs[0] } return mergeError{errs} }