Revert "lib/streamaggr: do not flush dedup shards in parallel"

This reverts commit eb40395a1c.

Reason for revert: it has been appeared that the performance gain on multiple CPU cores
wasn't visible because the benchmark was generating incorrect pushSample.key.

See a207e0bf687d65f5198207477248d70c69284296

lib/streamaggr/dedup.go

@@ -113,14 +113,22 @@ func (ctx *dedupFlushCtx) reset() {
 }
 
 func (da *dedupAggr) flush(f func(samples []pushSample)) {
-	// Do not flush shards in parallel, since this significantly increases CPU usage
-	// on systems with many CPU cores, while doesn't improve flush latency too much.
-	ctx := getDedupFlushCtx()
+	var wg sync.WaitGroup
 	for i := range da.shards {
-		ctx.reset()
-		da.shards[i].flush(ctx, f)
+		flushConcurrencyCh <- struct{}{}
+		wg.Add(1)
+		go func(shard *dedupAggrShard) {
+			defer func() {
+				<-flushConcurrencyCh
+				wg.Done()
+			}()
+
+			ctx := getDedupFlushCtx()
+			shard.flush(ctx, f)
+			putDedupFlushCtx(ctx)
+		}(&da.shards[i])
 	}
-	putDedupFlushCtx(ctx)
+	wg.Wait()
 }
 
 type perShardSamples struct {

lib/streamaggr/streamaggr.go

@@ -670,9 +670,7 @@ func (a *aggregator) dedupFlush(dedupInterval time.Duration) {
 
 	startTime := time.Now()
 
-	flushConcurrencyCh <- struct{}{}
 	a.da.flush(a.pushSamples)
-	<-flushConcurrencyCh
 
 	d := time.Since(startTime)
 	a.dedupFlushDuration.Update(d.Seconds())