@@ -44,6 +44,10 @@ type selectQueryContext struct {
4444 requestChannels []chan * qryResults
4545 errorChannel chan error
4646 wg sync.WaitGroup
47+ createDFLock sync.Mutex
48+ stopChan chan bool
49+ queryWG sync.WaitGroup
50+ finalErrorChan chan error
4751}
4852
4953func (queryCtx * selectQueryContext ) start (parts []* partmgr.DBPartition , params * SelectParams ) (* frameIterator , error ) {
@@ -77,18 +81,21 @@ func (queryCtx *selectQueryContext) start(parts []*partmgr.DBPartition, params *
7781 }
7882 }
7983
84+ queryCtx .stopChan = make (chan bool , 1 )
85+ queryCtx .finalErrorChan = make (chan error , 1 )
86+ queryCtx .errorChannel = make (chan error , queryCtx .workers + len (queries ))
87+
8088 err = queryCtx .startCollectors ()
8189 if err != nil {
8290 return nil , err
8391 }
8492
8593 for _ , query := range queries {
86- err = queryCtx .processQueryResults (query )
87- if err != nil {
88- return nil , err
89- }
94+ queryCtx .queryWG .Add (1 )
95+ go processQueryResults (queryCtx , query )
9096 }
9197
98+ queryCtx .queryWG .Wait ()
9299 for i := 0 ; i < queryCtx .workers ; i ++ {
93100 close (queryCtx .requestChannels [i ])
94101 }
@@ -98,7 +105,7 @@ func (queryCtx *selectQueryContext) start(parts []*partmgr.DBPartition, params *
98105 close (queryCtx .errorChannel )
99106
100107 // return first error
101- err = <- queryCtx .errorChannel
108+ err = <- queryCtx .finalErrorChan
102109 if err != nil {
103110 return nil , err
104111 }
@@ -260,7 +267,6 @@ func (queryCtx *selectQueryContext) parsePreAggregateLabels(partition *partmgr.D
260267func (queryCtx * selectQueryContext ) startCollectors () error {
261268
262269 queryCtx .requestChannels = make ([]chan * qryResults , queryCtx .workers )
263- queryCtx .errorChannel = make (chan error , queryCtx .workers )
264270
265271 // Increment the WaitGroup counter.
266272 queryCtx .wg .Add (queryCtx .workers )
@@ -274,27 +280,45 @@ func (queryCtx *selectQueryContext) startCollectors() error {
274280 }(i )
275281 }
276282
283+ // Watch error channel, and signal all go routines to stop in case of an error
284+ go func () {
285+ // Signal all goroutines to stop when error received
286+ err , ok := <- queryCtx .errorChannel
287+ if ok && err != nil {
288+ close (queryCtx .stopChan )
289+ queryCtx .finalErrorChan <- err
290+ }
291+
292+ close (queryCtx .finalErrorChan )
293+ return
294+ }()
295+
277296 return nil
278297}
279298
280- func (queryCtx * selectQueryContext ) processQueryResults (query * partQuery ) error {
299+ func processQueryResults (queryCtx * selectQueryContext , query * partQuery ) {
300+ defer queryCtx .queryWG .Done ()
301+
281302 for query .Next () {
282303
283304 // read metric name
284305 name , ok := query .GetField (config .MetricNameAttrName ).(string )
285306 if ! ok {
286- return fmt .Errorf ("could not find metric name attribute in response, res:%v" , query .GetFields ())
307+ queryCtx .errorChannel <- fmt .Errorf ("could not find metric name attribute in response, res:%v" , query .GetFields ())
308+ return
287309 }
288310
289311 // read label set
290312 lsetAttr , lok := query .GetField (config .LabelSetAttrName ).(string )
291313 if ! lok {
292- return fmt .Errorf ("could not find label set attribute in response, res:%v" , query .GetFields ())
314+ queryCtx .errorChannel <- fmt .Errorf ("could not find label set attribute in response, res:%v" , query .GetFields ())
315+ return
293316 }
294317
295318 lset , err := utils .LabelsFromString (lsetAttr )
296319 if err != nil {
297- return err
320+ queryCtx .errorChannel <- err
321+ return
298322 }
299323
300324 // read chunk encoding type
@@ -306,7 +330,8 @@ func (queryCtx *selectQueryContext) processQueryResults(query *partQuery) error
306330 } else {
307331 intEncoding , err := strconv .Atoi (encodingStr )
308332 if err != nil {
309- return fmt .Errorf ("error parsing encoding type of chunk, got: %v, error: %v" , encodingStr , err )
333+ queryCtx .errorChannel <- fmt .Errorf ("error parsing encoding type of chunk, got: %v, error: %v" , encodingStr , err )
334+ return
310335 }
311336 encoding = chunkenc .Encoding (intEncoding )
312337 }
@@ -324,7 +349,8 @@ func (queryCtx *selectQueryContext) processQueryResults(query *partQuery) error
324349 if labelValue != "" {
325350 newLset [i ] = utils.Label {Name : trimmed , Value : labelValue }
326351 } else {
327- return fmt .Errorf ("no label named %v found to group by" , trimmed )
352+ queryCtx .errorChannel <- fmt .Errorf ("no label named %v found to group by" , trimmed )
353+ return
328354 }
329355 }
330356 lset = newLset
@@ -336,6 +362,7 @@ func (queryCtx *selectQueryContext) processQueryResults(query *partQuery) error
336362 hash = lset .Hash ()
337363 }
338364
365+ queryCtx .createDFLock .Lock ()
339366 // find or create data frame
340367 frame , ok := queryCtx .dataFrames [hash ]
341368 if ! ok {
@@ -349,19 +376,30 @@ func (queryCtx *selectQueryContext) processQueryResults(query *partQuery) error
349376 results .IsServerAggregates (),
350377 queryCtx .showAggregateLabel )
351378 if err != nil {
352- return err
379+ queryCtx .errorChannel <- err
380+ queryCtx .createDFLock .Unlock ()
381+ return
353382 }
354383 queryCtx .dataFrames [hash ] = frame
355384 queryCtx .frameList = append (queryCtx .frameList , frame )
356385 }
386+ queryCtx .createDFLock .Unlock ()
357387
358388 results .frame = frame
359-
360389 workerNum := hash & uint64 (queryCtx .workers - 1 )
361- queryCtx .requestChannels [workerNum ] <- & results
390+
391+ // In case termination signal was received exit, Otherwise send query result to worker
392+ select {
393+ case _ = <- queryCtx .stopChan :
394+ return
395+ case queryCtx .requestChannels [workerNum ] <- & results :
396+ }
397+
362398 }
363399
364- return query .Err ()
400+ if query .Err () != nil {
401+ queryCtx .errorChannel <- query .Err ()
402+ }
365403}
366404
367405func (queryCtx * selectQueryContext ) createColumnSpecs () ([]columnMeta , map [string ][]columnMeta , error ) {
0 commit comments