github-mirrors/VictoriaMetrics
1
0
Fork 0
mirror of https://github.com/VictoriaMetrics/VictoriaMetrics.git synced 2025-02-09 15:27:11 +00:00
Code Issues Projects Releases Packages Wiki Activity

lib/protoparser: stop reading the input stream as soon as the callback provided by the caller returns error

Browse source 
This is a follow-up for af90c3c43b
This commit is contained in:
Aliaksandr Valialkin 2021-06-14 15:18:46 +03:00
parent 2ea187e801
commit ebaf68bcb0
6 changed files with 47 additions and 12 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

9
lib/protoparser/csvimport/streamparser.go
View file

@ -69,7 +69,7 @@ func ParseStream(req *http.Request, callback func(rows []Row) error) error {
func (ctx *streamContext) Read() bool { func (ctx *streamContext) Read() bool {
readCalls.Inc() readCalls.Inc()
if ctx.err != nil { if ctx.err != nil || ctx.hasCallbackError() {
return false return false
} }
ctx.reqBuf, ctx.tailBuf, ctx.err = common.ReadLinesBlock(ctx.br, ctx.reqBuf, ctx.tailBuf) ctx.reqBuf, ctx.tailBuf, ctx.err = common.ReadLinesBlock(ctx.br, ctx.reqBuf, ctx.tailBuf)
@ -107,6 +107,13 @@ func (ctx *streamContext) Error() error {
return ctx.err return ctx.err
} }
func (ctx *streamContext) hasCallbackError() bool {
ctx.callbackErrLock.Lock()
ok := ctx.callbackErr != nil
ctx.callbackErrLock.Unlock()
return ok
}
func (ctx *streamContext) reset() { func (ctx *streamContext) reset() {
ctx.br.Reset(nil) ctx.br.Reset(nil)
ctx.reqBuf = ctx.reqBuf[:0] ctx.reqBuf = ctx.reqBuf[:0]

9
lib/protoparser/graphite/streamparser.go
View file

@ -54,7 +54,7 @@ func ParseStream(r io.Reader, callback func(rows []Row) error) error {
func (ctx *streamContext) Read() bool { func (ctx *streamContext) Read() bool {
readCalls.Inc() readCalls.Inc()
if ctx.err != nil { if ctx.err != nil || ctx.hasCallbackError() {
return false return false
} }
ctx.reqBuf, ctx.tailBuf, ctx.err = common.ReadLinesBlock(ctx.br, ctx.reqBuf, ctx.tailBuf) ctx.reqBuf, ctx.tailBuf, ctx.err = common.ReadLinesBlock(ctx.br, ctx.reqBuf, ctx.tailBuf)
@ -86,6 +86,13 @@ func (ctx *streamContext) Error() error {
return ctx.err return ctx.err
} }
func (ctx *streamContext) hasCallbackError() bool {
ctx.callbackErrLock.Lock()
ok := ctx.callbackErr != nil
ctx.callbackErrLock.Unlock()
return ok
}
func (ctx *streamContext) reset() { func (ctx *streamContext) reset() {
ctx.br.Reset(nil) ctx.br.Reset(nil)
ctx.reqBuf = ctx.reqBuf[:0] ctx.reqBuf = ctx.reqBuf[:0]

9
lib/protoparser/influx/streamparser.go
View file

@ -82,7 +82,7 @@ func ParseStream(r io.Reader, isGzipped bool, precision, db string, callback fun
func (ctx *streamContext) Read() bool { func (ctx *streamContext) Read() bool {
readCalls.Inc() readCalls.Inc()
if ctx.err != nil { if ctx.err != nil || ctx.hasCallbackError() {
return false return false
} }
ctx.reqBuf, ctx.tailBuf, ctx.err = common.ReadLinesBlockExt(ctx.br, ctx.reqBuf, ctx.tailBuf, maxLineSize.N) ctx.reqBuf, ctx.tailBuf, ctx.err = common.ReadLinesBlockExt(ctx.br, ctx.reqBuf, ctx.tailBuf, maxLineSize.N)
@ -120,6 +120,13 @@ func (ctx *streamContext) Error() error {
return ctx.err return ctx.err
} }
func (ctx *streamContext) hasCallbackError() bool {
ctx.callbackErrLock.Lock()
ok := ctx.callbackErr != nil
ctx.callbackErrLock.Unlock()
return ok
}
func (ctx *streamContext) reset() { func (ctx *streamContext) reset() {
ctx.br.Reset(nil) ctx.br.Reset(nil)
ctx.reqBuf = ctx.reqBuf[:0] ctx.reqBuf = ctx.reqBuf[:0]

9
lib/protoparser/opentsdb/streamparser.go
View file

@ -53,7 +53,7 @@ func ParseStream(r io.Reader, callback func(rows []Row) error) error {
func (ctx *streamContext) Read() bool { func (ctx *streamContext) Read() bool {
readCalls.Inc() readCalls.Inc()
if ctx.err != nil { if ctx.err != nil || ctx.hasCallbackError() {
return false return false
} }
ctx.reqBuf, ctx.tailBuf, ctx.err = common.ReadLinesBlock(ctx.br, ctx.reqBuf, ctx.tailBuf) ctx.reqBuf, ctx.tailBuf, ctx.err = common.ReadLinesBlock(ctx.br, ctx.reqBuf, ctx.tailBuf)
@ -85,6 +85,13 @@ func (ctx *streamContext) Error() error {
return ctx.err return ctx.err
} }
func (ctx *streamContext) hasCallbackError() bool {
ctx.callbackErrLock.Lock()
ok := ctx.callbackErr != nil
ctx.callbackErrLock.Unlock()
return ok
}
func (ctx *streamContext) reset() { func (ctx *streamContext) reset() {
ctx.br.Reset(nil) ctx.br.Reset(nil)
ctx.reqBuf = ctx.reqBuf[:0] ctx.reqBuf = ctx.reqBuf[:0]

14
lib/protoparser/prometheus/streamparser.go
View file

@ -56,7 +56,7 @@ func ParseStream(r io.Reader, defaultTimestamp int64, isGzipped bool, callback f
func (ctx *streamContext) Read() bool { func (ctx *streamContext) Read() bool {
readCalls.Inc() readCalls.Inc()
if ctx.err != nil || ctx.CallbackError() != nil { if ctx.err != nil || ctx.hasCallbackError() {
return false return false
} }
ctx.reqBuf, ctx.tailBuf, ctx.err = common.ReadLinesBlock(ctx.br, ctx.reqBuf, ctx.tailBuf) ctx.reqBuf, ctx.tailBuf, ctx.err = common.ReadLinesBlock(ctx.br, ctx.reqBuf, ctx.tailBuf)
@ -77,7 +77,7 @@ type streamContext struct {
err error err error
wg sync.WaitGroup wg sync.WaitGroup
callbackErrLock sync.RWMutex callbackErrLock sync.Mutex
callbackErr error callbackErr error
} }
@ -88,11 +88,11 @@ func (ctx *streamContext) Error() error {
return ctx.err return ctx.err
} }
func (ctx *streamContext) CallbackError() error { func (ctx *streamContext) hasCallbackError() bool {
ctx.callbackErrLock.RLock() ctx.callbackErrLock.Lock()
callbackErr := ctx.callbackErr ok := ctx.callbackErr != nil
ctx.callbackErrLock.RUnlock() ctx.callbackErrLock.Unlock()
return callbackErr return ok
} }
func (ctx *streamContext) reset() { func (ctx *streamContext) reset() {

9
lib/protoparser/vmimport/streamparser.go
View file

@ -59,7 +59,7 @@ func ParseStream(req *http.Request, callback func(rows []Row) error) error {
func (ctx *streamContext) Read() bool { func (ctx *streamContext) Read() bool {
readCalls.Inc() readCalls.Inc()
if ctx.err != nil { if ctx.err != nil || ctx.hasCallbackError() {
return false return false
} }
ctx.reqBuf, ctx.tailBuf, ctx.err = common.ReadLinesBlockExt(ctx.br, ctx.reqBuf, ctx.tailBuf, maxLineLen.N) ctx.reqBuf, ctx.tailBuf, ctx.err = common.ReadLinesBlockExt(ctx.br, ctx.reqBuf, ctx.tailBuf, maxLineLen.N)
@ -97,6 +97,13 @@ func (ctx *streamContext) Error() error {
return ctx.err return ctx.err
} }
func (ctx *streamContext) hasCallbackError() bool {
ctx.callbackErrLock.Lock()
ok := ctx.callbackErr != nil
ctx.callbackErrLock.Unlock()
return ok
}
func (ctx *streamContext) reset() { func (ctx *streamContext) reset() {
ctx.br.Reset(nil) ctx.br.Reset(nil)
ctx.reqBuf = ctx.reqBuf[:0] ctx.reqBuf = ctx.reqBuf[:0]