Add support for concurrent index block queries #1195

robskillington · 2018-11-22T20:49:47Z

No description provided.

codecov · 2018-11-22T20:55:48Z

Codecov Report

Merging #1195 into master will decrease coverage by <.1%.
The diff coverage is 80.9%.

@@           Coverage Diff            @@
##           master   #1195     +/-   ##
========================================
- Coverage    71.1%     71%   -0.1%     
========================================
  Files         739     739             
  Lines       62065   62206    +141     
========================================
+ Hits        44132   44197     +65     
- Misses      15083   15143     +60     
- Partials     2850    2866     +16

Flag	Coverage Δ
#aggregator	`81.6% <ø> (ø)`	⬆️
#cluster	`85.6% <ø> (-0.1%)`	⬇️
#collector	`78.1% <ø> (ø)`	⬆️
#dbnode	`80.7% <80.9%> (-0.2%)`	⬇️
#m3em	`73.2% <ø> (ø)`	⬆️
#m3ninx	`75.3% <ø> (ø)`	⬆️
#m3nsch	`51.1% <ø> (ø)`	⬆️
#metrics	`18.3% <ø> (ø)`	⬆️
#msg	`74.9% <ø> (ø)`	⬆️
#query	`61.4% <ø> (ø)`	⬆️
#x	`74.4% <ø> (ø)`	⬆️

Continue to review full report at Codecov.

Legend - Click here to learn more
Δ = absolute <relative> (impact), ø = not affected, ? = missing data
Powered by Codecov. Last update cc6f8fb...308b766. Read the comment docs.

… blocks

robskillington · 2018-11-26T19:40:27Z

Ready to review now

richardartoul · 2018-11-26T19:47:26Z

src/dbnode/runtime/runtime_options.go

@@ -38,6 +38,10 @@ const (
 	// DefaultBootstrapConsistencyLevel is the default bootstrap consistency level
 	DefaultBootstrapConsistencyLevel = topology.ReadConsistencyLevelMajority

+	// DefaultIndexDefaultQueryTimeout is the hard timeout value to use if none is
+	// specified for a specific query, zero specifies no timeout.
+	DefaultIndexDefaultQueryTimeout = time.Minute


What value are you seeing in prod? I would have thought 15->30 would be a more reasonable default

Yeah, it could be, I just don't want to necessarily break some users existing queries. I could be convinced using 30s instead though for sure.

Could you file an issue for this to be settable from config?

since we don't have good support for setting runtime values in O.S.S

Yup, I opened here:
#1220

richardartoul

Haven't finished reviewing everything yet but I need to run right now, will finish review later

richardartoul · 2018-11-26T19:50:07Z

src/dbnode/storage/index.go

 		return index.QueryResults{}, errDbIndexUnableToQueryClosed
 	}

-	// override query response limit if needed.
+	// Track this as an inflight query that needs to finish


Super nit: period at the end of all these comments, I've been trying to fix these as we go

richardartoul · 2018-11-26T19:51:03Z

src/dbnode/storage/index.go

+	}, nil
+}
+
+func (i *nsIndex) timeoutForQueryWithLock(


withRLock for consistency with our naming conventions elsewhere

richardartoul · 2018-11-26T19:51:07Z

src/dbnode/storage/index.go

+	return i.state.runtimeOpts.defaultQueryTimeout
+}
+
+func (i *nsIndex) overriddenOptsForQueryWithLock(


withRLock for consistency with our naming conventions elsewhere

richardartoul · 2018-11-26T19:54:12Z

src/dbnode/storage/index.go

-	results.Reset(i.nsMetadata.ID())
-	ctx.RegisterFinalizer(results)
-
+func (i *nsIndex) blocksForQueryWithLock(queryRange xtime.Ranges) ([]index.Block, error) {


withRLock for consistency with our naming conventions elsewhere

richardartoul · 2018-11-26T19:55:52Z

src/dbnode/storage/index.go

+	}
+
+	var (
+		start      = i.nowFn()


this doesn't depend on the lock right? Can probably move this above the RLock and then set the deadline down here still

richardartoul · 2018-11-26T20:27:27Z

src/dbnode/storage/index.go

 		}

-		// ensure the block has data requested by the query
+		if queryRange.IsEmpty() {


nit: This might be a little less confusing if it was the first part of the loop. As is, I was confused why you would need to check this after doing the block lookup, but its not related to that at all

richardartoul · 2018-11-26T20:29:01Z

src/dbnode/storage/index.go

-		}
-	}
+func (i *nsIndex) newConcurrentResults(ctx context.Context) *index.ConcurrentResults {
+	results := i.opts.IndexOptions().ResultsPool().Get()


want to just throw this pool onto the struct itself like we do with the worker pool so you don't have to do the triple function call?

richardartoul · 2018-11-26T20:33:42Z

src/dbnode/storage/index.go

 	i.state.closed = true

 	var multiErr xerrors.MultiError
 	multiErr = multiErr.Add(i.state.insertQueue.Stop())

+	// Wait for inflight queries to finish before closing blocks
+	i.queriesWg.Wait()


Might be safer (or less likely to deadlock when changes are made) if you do this wait outside the lock. So:

i.state.Lock() defer i.state.Unlock() if !i.isOpenWithRLock() { return errDbIndexAlreadyClosed } // Signal our intent to close so that we stop accepting reads / writes. i.state.closed = true i.state.Unlock() // Wait for in-flight queries to finish before continuing. i.queriesWg.Wait() // Reacquire the lock so we can complete the shutdown. // Not even sure you need to re-acquire the lock at this point if you're willing // to remove the defer at the top since technically no one else can call Close() // after you mark is closed. i.state.Lock() var multiErr xerror.MultiError multiErr = multiErr.Add(i.state.insertQueue.Stop()) ...

You understand the lifecycle better than I do, so take it or leave it

richardartoul · 2018-11-26T20:37:35Z

src/dbnode/storage/index/block.go

-		size       = results.Size()
-		brokeEarly = false
-	)
+	size := results.Size()


wanna add the var back? I think it was cleaner. I assume you removed it unintentionally while refactoring

richardartoul · 2018-11-26T20:46:55Z

src/dbnode/storage/index/block.go

+	// we only retrieve the results lock when we add a batch of documents
+	// to the results set.
+	batch := b.docsPool.Get()
+	// Use documentArrayPoolCapacity to as max batch to avoid growing outside


"Use the maximum capacity of the array pool as the max batch size to avoid growing outside the allowed pool capacity."

richardartoul · 2018-11-27T14:51:53Z

src/dbnode/storage/index/block.go

 	}()

+	// NB(r): This query method only called once per block so is relatively
+	// cheap to declared as a lambda.
+	flushBatch := func() error {


could you eliminate the allocation entirely if you had a named function / method with the signature:

func flushBatch(batch []Document, results *ConcurrentResults) ([]Document, error)

and then you could use it like:

batch = append(batch, d) if len(batch) < maxBatch { continue } batch, err = flushBatch(batch, results) if err != nil { ...

Maybe overkill, but the anonymous func doesn't seem like its closing over many vars

richardartoul · 2018-11-27T14:52:53Z

src/dbnode/storage/index/block.go

-		brokeEarly = false
-	)
+	size := results.Size()
+	limitedResults := false


can you just call this exhaustive and return it at the end instead of doing the limitedResults -> exhaustive translation

Reads worse, it was actually like this before.

richardartoul · 2018-11-27T14:54:06Z

src/dbnode/storage/index/options.go

+	// documentArrayPool size in general: 256*256*sizeof(doc.Document)
+	// = 256 * 256 * 16
+	// = 1mb (but with Go's heap probably 2mb)
+	// TODO(r): Make this configurable in a followup change.


We're probably gonna want this tunable really soon so we don't have to ask people to recompile when they're testing our changes, especially if their read workloads are very different from ours

This is now only going to be used by the compactor, so it should be tiny (we never will have 256 concurrent compactions, and compactor just uses a single pool).

richardartoul · 2018-11-27T14:55:41Z

src/dbnode/storage/index/results.go

-			Name:  r.copyBytes(f.Name),
-			Value: r.copyBytes(f.Value),
-		})
+		tags.Append(r.idPool.CloneTag(ident.Tag{


is this just a cleanup?

Yup, this wasn't being pooled before, now it is (it always could have been).

richardartoul · 2018-11-28T19:28:40Z

src/dbnode/storage/index.go

+		workers  = i.opts.QueryIDsWorkerPool()
+		wg       sync.WaitGroup
+
+		// results contains all concurrent mutalbe state below


…ced timeout tests

richardartoul · 2018-11-29T17:40:17Z

src/dbnode/runtime/runtime_options.go

@@ -38,6 +38,10 @@ const (
 	// DefaultBootstrapConsistencyLevel is the default bootstrap consistency level
 	DefaultBootstrapConsistencyLevel = topology.ReadConsistencyLevelMajority

+	// DefaultIndexDefaultQueryTimeout is the hard timeout value to use if none is
+	// specified for a specific query, zero specifies no timeout.
+	DefaultIndexDefaultQueryTimeout = time.Minute


Could you file an issue for this to be settable from config?

richardartoul · 2018-11-29T17:40:46Z

src/dbnode/runtime/runtime_options.go

@@ -38,6 +38,10 @@ const (
 	// DefaultBootstrapConsistencyLevel is the default bootstrap consistency level
 	DefaultBootstrapConsistencyLevel = topology.ReadConsistencyLevelMajority

+	// DefaultIndexDefaultQueryTimeout is the hard timeout value to use if none is
+	// specified for a specific query, zero specifies no timeout.
+	DefaultIndexDefaultQueryTimeout = time.Minute


since we don't have good support for setting runtime values in O.S.S

richardartoul · 2018-11-29T17:42:16Z

src/dbnode/storage/index.go

+
+	// queriesWg tracks outstanding queries to ensure
+	// we wait for all queries to complete before actually closing
+	// blocks and other cleanup tasks on index close


super nit: Period.

richardartoul · 2018-11-29T17:42:38Z

src/dbnode/storage/index.go

+	i.queriesWg.Add(1)
+	defer i.queriesWg.Done()
+
+	// Enact overrides for query options


super nit: period

richardartoul · 2018-11-29T18:40:22Z

src/dbnode/storage/index.go

+		}
+
+		var (
+			ticker  = time.NewTicker(timeLeft)


I know Prateek has talked about time tickers not getting scheduled properly in certain situations and preferring to use loops with time.Sleep which apparently has some special hooks into the runtime, although then you'd have to implement polling so ehhhh. Probably fine as is

richardartoul · 2018-11-29T18:45:53Z

src/dbnode/storage/index.go

@@ -905,18 +1138,19 @@ func (i *nsIndex) CleanupExpiredFileSets(t time.Time) error {

 func (i *nsIndex) Close() error {


is there not a "Tick()" loop? How does the ticking get stopped

There is a tick loop yeah, see Tick() on nsIndex.

Add support for concurrent index block queries

d8deb99

Rob Skillington added 3 commits November 22, 2018 15:55

Document the max batch behavior

25a11f7

Fix tests

8a02656

Fix codegen

bf7ea21

robskillington changed the title ~~[WIP] Add support for concurrent index block queries~~ Add support for concurrent index block queries Nov 23, 2018

Rob Skillington added 5 commits November 23, 2018 19:27

Fix race in block.Query

96dfeab

Fix notes

d3f2667

Refactor to not hold lock during the execution of a query

3d9c5e0

Track outstanding queries and wait to finish executing before closing…

afea517

… blocks

Regenerate mocks

f447a2a

richardartoul self-requested a review November 26, 2018 19:46

richardartoul reviewed Nov 26, 2018

View reviewed changes

Fix metalint errors

38a3202

richardartoul reviewed Nov 27, 2018

View reviewed changes

Address feedback, use merged results sets

20d2b0f

richardartoul reviewed Nov 28, 2018

View reviewed changes

Rob Skillington added 8 commits November 28, 2018 14:33

Address feedback

388c92f

Refactor and address further feedback

18e36f3

Wait for inflight queries to finish outside of lock in index close

8199a73

Fix build

af435bc

Add concurrent test

76dd8fe

Add early abort when waiting for all goroutines to finish and add for…

a566863

…ced timeout tests

Cleanup early abort more

62dcc65

Attempt to remove XML test output from tests causing test failures

658adb6

Rob Skillington added 4 commits November 29, 2018 10:41

Update ci-scripts to print out failing packages

9d52c4f

Update ci-scripts

546ee7e

Fix concurrent test OOMing on CI

527329d

Move concurrent test to its own file and pull in master ci-scripts

308b766

richardartoul approved these changes Nov 29, 2018

View reviewed changes

robskillington merged commit 8229522 into master Nov 29, 2018

justinjc deleted the r/concurrent-index-block-queries branch January 7, 2019 19:30

		@@ -905,18 +1138,19 @@ func (i *nsIndex) CleanupExpiredFileSets(t time.Time) error {

		func (i *nsIndex) Close() error {

Add support for concurrent index block queries #1195

Add support for concurrent index block queries #1195

Conversation

robskillington commented Nov 22, 2018 • edited Loading

codecov bot commented Nov 22, 2018 • edited Loading

Codecov Report

robskillington commented Nov 26, 2018

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

richardartoul 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

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

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

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

robskillington commented Nov 22, 2018 •

edited

Loading

codecov bot commented Nov 22, 2018 •

edited

Loading