Throttle.Do does not panic after Finish

[ekalinin]

Fixes #1394

I'm not sure about an error at Do. Maybe it will be ok to just return nil.

---

This change is 

[jarifibrahim]

Thanks for fixing this @ekalinin . I've added some comments.

Reviewable status: 0 of 2 files reviewed, 4 unresolved discussions (waiting on @ashish-goswami, @ekalinin, @jarifibrahim, and @manishrjain)

---

batch_test.go, line 128 at r1 (raw file):

// This test ensures we don't panic during flush.
func TestFlushPanic(t *testing.T) {

This test can be moved to the y/y_test.go file. You could actually test the throttle type instead of testing the writeBatch.

In your test, you would do

th := NewThrottle(..)
th.Do -> Shouldn't return error
th.Finish -> shouldn't return error
th.Do -> Should return an error

---

batch_test.go, line 130 at r1 (raw file):

func TestFlushPanic(t *testing.T) {
	t.Run("after flush", func(t *testing.T) {
		runBadgerTest(t, nil, func(t *testing.T, db *DB) {

We're opening a new DB here and since we just want to test the throttle type, this would be an overkill :)

---

y/y.go, line 46 at r1 (raw file):

	// ErrDoAfterFinish indicates a pointless Do call after Finish
	ErrDoAfterFinish = errors.New("Do has no any point after Finish")

Please rephrase the comment and the error message it to

Throttle.Do called after Finish

---

y/y.go, line 270 at r1 (raw file):

// previously Done workers, it would return it.
func (t *Throttle) Do() error {
	if t.finished {

This would result in a race condition. Two threads can call throttle.Do and throttle.Finish() at the same time. We might need to use an atomic here and do a compare and swap operation.

Something like

Throttle.Finish does
compareAndSwap(finish, 1, 0) => mark as finished. If compareAndSwap returns false, it means 
someone called finish for the second time and we should return error (I think think this could 
happen because we have sync.Once in finish() function but it's good to add a check.)

Throttle.Do does
if atomic.Load(finish) != 0 { return error }

[ekalinin]

Thanks for comments @jarifibrahim
Fixed all of them. Please, take a look.

[jarifibrahim]

Some minor comments. Please address them and this PR is good to go 🎉

Reviewable status: 0 of 2 files reviewed, 2 unresolved discussions (waiting on @ashish-goswami, @ekalinin, and @manishrjain)

---

y/y.go, line 312 at r2 (raw file):

		close(t.ch)
		close(t.errCh)
		if !atomic.CompareAndSwapInt32(&t.finished, 0, 1) {

The atomic check would never return false here since it is wrapped inside sync.Do which means this code will be called only once. See this https://play.golang.org/p/OHHqqPClxjb

It should be enough to do a blind update. My previous comment was incorrect. Apologies.
So we can remove the ErrDoubleFinish too.

---

y/y_test.go, line 265 at r2 (raw file):

	require.NoError(t, th.Finish())
	require.NotPanics(t, func() {
		require.Error(t, th.Do())

You can remove the require.NotPanics.

A simple require.Error should be enough.

[ekalinin]

Some minor comments. Please address them and this PR is good to go

It should be enough to do a blind update. My previous comment was incorrect. Apologies.
So we can remove the ErrDoubleFinish too.

Done.

You can remove the require.NotPanics.

A simple require.Error should be enough.

Done.

[NamanJain8]

@ekalinin, consider a scenario where Do() and Finish() are called by different goroutines. In that case, there is a race condition.
Thread 1 passes this check while thread 2 closes the channel and then thread 1 push into closed channel ch.
Try running this scenario using --race flag.

func TestThrottleDoAfterFinish(t *testing.T) {
	th := NewThrottle(16)

	var wg sync.WaitGroup
	for i := 1; i <= 10; i++ {
		wg.Add(1)
		go func(i int) {
			defer wg.Done()
			err := th.Do()
			switch err {
			case nil:
				th.Done(nil)
			case ErrDoAfterFinish:
				// pass
			default:
				panic("Failed test")
			}
			if i == 5 {
				th.Finish()
			}
		}(i)
	}
	wg.Wait()
}

[NamanJain8]

@ekalinin, In the Finish(), you can have an atomic Store operation for finished. While in Do(), you can use a lock to ensure that Finish() does not update the finished while Do() is updating executing.

[ekalinin]

@ekalinin, In the Finish(), you can have an atomic Store operation for finished. While in Do(), you can use a lock to ensure that Finish() does not update the finished while Do() is updating executing.

Thanks for the hint, @NamanJain8!
Fixed.

[jarifibrahim]

Hey @ekalinin, we got this PR reviewed internally and there are a few changes needed. @NamanJain8 will make those change and push on this PR. Thank you for helping fix this 🎉

[ekalinin]

Hey @ekalinin, we got this PR reviewed internally and there are a few changes needed. @NamanJain8 will make those change and push on this PR. Thank you for helping fix this

Cool! Thanks for help as well :) !