storage/rangefeed: per registration buffer #33557

mrtracy · 2019-01-07T23:43:31Z

Modifies the rangefeed package so that each individual registration
maintains an output buffer which is processed by a dedicated goroutine.
This eliminates the possibility of a slow consumer blocking the
processor from making progress.

Previously, writing events to registrations was synchronous, meaning
that a single slow consumer could cause the publishing process to block.
If publishing was blocked for too long, the processor along with all
registrations were immediately torn down. With this change, a slow
consumer will no longer block its siblings; if a single consumer blocks
for too long, it will enter an "overflow" state and will be torn down as
soon as its existing buffer is fully processed.

The registry/registration interface has been modified fairly
significantly to accomodate this; the processor itself has seen modest
modification (mostly moving functionality to the registry level and
removing no-longer-needed support for the previous functionality). Some
of the processor tests have been modified to account for the now
asynchronous nature of publishing events to individual streams.

Fixes #32945

Release note: None

cockroach-teamcity · 2019-01-07T23:43:37Z

This change is

nvanbenschoten

Aside from a few small questions, this looks really good! Have you tested it with the roachtests yet?

Reviewed 7 of 7 files at r1.
Reviewable status: complete! 0 of 0 LGTMs obtained

pkg/storage/rangefeed/processor.go, line 372 at r1 (raw file):

	case p.regC <- r:
	case <-p.stoppedC:
		if catchUpIter != nil {

Why did this change? Will this cause us to leak iterators if the Processor is stopped?

pkg/storage/rangefeed/registry.go, line 71 at r1 (raw file):

	mu struct {
		*syncutil.Mutex

sync.Locker

Or pass a pointer to a registration everywhere and don't make this pointer to a mutex.

pkg/storage/rangefeed/registry.go, line 150 at r1 (raw file):

// 1. If a catch-up scan is indicated, run one before beginning the proper
// output loop.
// 2. After catch-up is complete, Begin reading from the registration buffer

s/Begin/begin/

pkg/storage/rangefeed/registry.go, line 187 at r1 (raw file):

		case <-ctx.Done():
			return ctx.Err()
		case <-r.stream.Context().Done():

👍 I'm happy that we can do this instead of polling all stream contexts.

pkg/storage/rangefeed/registry.go, line 193 at r1 (raw file):

}

func (r *registration) runOutputLoop(ctx context.Context, doneCb func(r *registration)) {

If this is executing synchronously then why take a callback instead of just running the disconnect code after this returns?

pkg/storage/rangefeed/registry.go, line 206 at r1 (raw file):

// runCatchupScan starts a catchup scan which will output entries for all
// recorded changes in the replica that are newer than the catchupTimeStamp.
// This creates a single engine iterator immediately when this method

Is this sentence true?

pkg/storage/rangefeed/registry.go, line 274 at r1 (raw file):

}

func (r *registration) waitForCaughtUp() error {

This is just used in tests, right? Can we move it to a _test.go file so that it isn't part of the registration type's method set outside of tests?

pkg/storage/rangefeed/registry.go, line 432 at r1 (raw file):

}

func (reg *registry) waitForCaughtUp() error {

Same question about moving this to a _test.go file.

mrtracy

TFTR! Have not run these in roachtests yet, but we've only recently started running roachtests with rangefeed enabled at all.

Reviewable status: complete! 0 of 0 LGTMs obtained

pkg/storage/rangefeed/processor.go, line 372 at r1 (raw file):