compact: add concurrency to group compact #898

earthdiaosi · 2019-03-07T08:04:30Z

Changes

Allow for compacting of group in parallel.

Verification

Tested by running thanos compact to see that compacting much quicker when concurrency is added.

earthdiaosi · 2019-03-11T01:40:46Z

@bwplotka please review

GiedriusS

NVM my comments that I've deleted - I missed some sneaky braces. LGTM from a purely mechanical side.

earthdiaosi · 2019-03-14T11:49:43Z

@GiedriusS i have fixed the conflicts, please review

bwplotka

Unfortunately we need something more complex than this.

We need to ensure directory is different (that might be the case already)
We have to limit the number of concurrent go routines. Essentially the main problem with compactor is not that it's too slow for typical cases. It's that it uses too much memory. We need to optimize the compaction process for it at some point, but essentially even single run for bigger project can take 30-50GB of RAM or sometimes more if people have milions of series.

This change will hit those users as now instead of "just" 60GB they will need 60GB * number of individual groups they have. For example for us it 30 different ones. So 60 * 30 GB suddenly with this PR merged. It's quite memory intensive part.

So I would suggest having something like this but have configurable number of workers to do it concurrently if user wish so and he has enough memory for their blocks. What do you think? (:

pkg/compact/compact.go

…group

earthdiaosi · 2019-03-21T04:20:42Z

@bwplotka @GiedriusS Please review again

bwplotka

Some question.

Also what do you think about moving to errgroup.Group instead?

bwplotka · 2019-03-22T17:12:09Z

pkg/compact/compact.go

+		errChan := make(chan error, len(groups))
+		groupChan := make(chan struct{}, groupCompactConcurrency)
+		defer close(groupChan)
+		for i := 0; i < groupCompactConcurrency; i++ {


I am reviewing on mobile, so might miss something, but I don't get this loop. What is the purpose of it?

Ok, I think I get it, but seem my comment on line 917

bwplotka

Thanks for update! but I think there is still some work to be done.. @earthdiaosi let me know what do you think!

bwplotka · 2019-04-03T15:32:12Z

pkg/compact/compact.go

@@ -886,7 +886,7 @@ func NewBucketCompactor(logger log.Logger, sy *Syncer, comp tsdb.Compactor, comp
 }

 // Compact runs compaction over bucket.
-func (c *BucketCompactor) Compact(ctx context.Context) error {
+func (c *BucketCompactor) Compact(ctx context.Context, groupCompactConcurrency int) error {


I think just concurrency would be enough.

Suggested change

func (c *BucketCompactor) Compact(ctx context.Context, groupCompactConcurrency int) error {

func (c *BucketCompactor) Compact(ctx context.Context, concurrency int) error {

bwplotka · 2019-04-03T15:32:42Z

pkg/compact/compact.go

 				}
-			}
-			return errors.Wrap(err, "compaction")
+				errChan <- errors.Wrap(err, "compaction")


Can we use errgroup.Group instead?

bwplotka · 2019-04-03T15:33:05Z

cmd/thanos/compact.go

@@ -95,6 +95,9 @@ func registerCompact(m map[string]setupFunc, app *kingpin.Application, name stri
 	blockSyncConcurrency := cmd.Flag("block-sync-concurrency", "Number of goroutines to use when syncing block metadata from object storage.").
 		Default("20").Int()

+	groupCompactConcurrency := cmd.Flag("group-compact-concurrency", "Number of goroutines to use when compacting group.").


Suggested change

groupCompactConcurrency := cmd.Flag("group-compact-concurrency", "Number of goroutines to use when compacting group.").

groupCompactConcurrency := cmd.Flag("compact.concurrency", "Number of goroutines to use when compacting group.").

bwplotka · 2019-04-03T15:33:19Z

cmd/thanos/compact.go

@@ -206,7 +211,7 @@ func runCompact(

 	ctx, cancel := context.WithCancel(context.Background())
 	f := func() error {
-		if err := compactor.Compact(ctx); err != nil {
+		if err := compactor.Compact(ctx, groupCompactConcurrency); err != nil {


Suggested change

if err := compactor.Compact(ctx, groupCompactConcurrency); err != nil {

if err := compactor.Compact(ctx, concurrency); err != nil {

bwplotka · 2019-04-03T15:55:28Z

pkg/compact/compact.go

+		errChan := make(chan error, len(groups))
+		groupChan := make(chan struct{}, groupCompactConcurrency)
+		defer close(groupChan)
+		for i := 0; i < groupCompactConcurrency; i++ {


Ok, I think I get it, but seem my comment on line 917

bwplotka · 2019-04-03T16:14:36Z

pkg/compact/compact.go

+				shouldRerunGroup, _, err := g.Compact(ctx, c.compactDir, c.comp)
+				if err == nil {
+					if shouldRerunGroup {
+						finishedAllGroups = false


Blocker: Non-thread-safe memory accessed by mulitple go routines.

bwplotka · 2019-04-03T16:19:40Z

pkg/compact/compact.go

@@ -913,22 +913,42 @@ func (c *BucketCompactor) Compact(ctx context.Context) error {
 			return errors.Wrap(err, "build compaction groups")
 		}
 		finishedAllGroups := true
+		var wg sync.WaitGroup
+		errChan := make(chan error, len(groups))


I think we need something more similiar to this https://github.com/improbable-eng/thanos/pull/887/files still.
This is for consistency and readability reasons. so:

Run concurrency number of go routines BEFORE for{ loop (891 line)

Do the whole sync Iterate over groups

for _, g := range groups { ````. 3. Send `g` via some channel with 0 queue. 4. In go routines do the g (`g.Compact`) and mark `finishedAllGroups` to false if there are still some group to run. Make sure to make it `atomic.Value` or use `sync.Mutex`. 5. After ` for _, g := range groups {` close channel and wait for all groups to finish. If `finishedAllGroups == true` return otherwise continue

mjd95 · 2019-04-04T14:39:07Z

@earthdiaosi thanks for raising this PR! I think it's a good optional addition, and can be a very helpful optimisation in some cases.

We are working through a large backlog of data and this PR is very helpful for us. Since we need to use it as soon as possible, I've addressed @bwplotka's comments and raised another PR (#1010) based on top of yours addressing them. I hope that's okay!

bwplotka · 2019-04-04T16:01:45Z

Closing to avoid confusion, hope it's ok with you @earthdiaosi ! Let us know if not, good work with starting with this!

earthdiaosi · 2019-04-08T07:00:22Z

@mjd95 @bwplotka it's ok with me.

GiedriusS approved these changes Mar 13, 2019

View reviewed changes

earthdiaosi force-pushed the master branch 2 times, most recently from afca3ac to de76399 Compare March 14, 2019 10:18

compact: add concurrency to group compact

609af48

earthdiaosi force-pushed the master branch from de76399 to 609af48 Compare March 14, 2019 11:34

bwplotka requested changes Mar 14, 2019

View reviewed changes

pkg/compact/compact.go Show resolved Hide resolved

wangyaqiang1 added 3 commits March 20, 2019 18:33

add flag to controll the number of goroutines to use when compacting …

ae90bf0

…group

update compact.md for group-compact-concurrency

d3aa040

fixed: miss wg.Add()

4d802ce

bwplotka reviewed Mar 22, 2019

View reviewed changes

bwplotka requested changes Apr 3, 2019

View reviewed changes

mjd95 mentioned this pull request Apr 4, 2019

Compact: group concurrency #1010

Merged

bwplotka closed this Apr 4, 2019

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

compact: add concurrency to group compact #898

compact: add concurrency to group compact #898

earthdiaosi commented Mar 7, 2019

earthdiaosi commented Mar 11, 2019

GiedriusS left a comment •

edited

Loading

earthdiaosi commented Mar 14, 2019

bwplotka left a comment

earthdiaosi commented Mar 21, 2019

bwplotka left a comment

bwplotka Mar 22, 2019

bwplotka Apr 3, 2019

bwplotka left a comment

bwplotka Apr 3, 2019

bwplotka Apr 3, 2019

bwplotka Apr 3, 2019

bwplotka Apr 3, 2019

bwplotka Apr 3, 2019

bwplotka Apr 3, 2019

bwplotka Apr 3, 2019

mjd95 commented Apr 4, 2019

bwplotka commented Apr 4, 2019

earthdiaosi commented Apr 8, 2019

	func (c *BucketCompactor) Compact(ctx context.Context, groupCompactConcurrency int) error {
	func (c *BucketCompactor) Compact(ctx context.Context, concurrency int) error {

	groupCompactConcurrency := cmd.Flag("group-compact-concurrency", "Number of goroutines to use when compacting group.").
	groupCompactConcurrency := cmd.Flag("compact.concurrency", "Number of goroutines to use when compacting group.").

	if err := compactor.Compact(ctx, groupCompactConcurrency); err != nil {
	if err := compactor.Compact(ctx, concurrency); err != nil {

compact: add concurrency to group compact #898

compact: add concurrency to group compact #898

Conversation

earthdiaosi commented Mar 7, 2019

Changes

Verification

earthdiaosi commented Mar 11, 2019

GiedriusS left a comment • edited Loading

Choose a reason for hiding this comment

earthdiaosi commented Mar 14, 2019

bwplotka left a comment

Choose a reason for hiding this comment

earthdiaosi commented Mar 21, 2019

bwplotka left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

bwplotka left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

mjd95 commented Apr 4, 2019

bwplotka commented Apr 4, 2019

earthdiaosi commented Apr 8, 2019

GiedriusS left a comment •

edited

Loading