Implement KZG proof batch verification (option 2) - uses worker pool

[terencechain]

This is an alt to #15609, it changes KZG proof batch verification from timer-based batching to a responsive worker pool model that processes messages immediately when idle and buffers when busy

[nalepae]

I did not run the code itself.

[satushh]

Should this be wrapped in a select statement with ctx.Done()

[kasey]

I think so too, and actually I would suggest using context.WithTimeout here to limit how long this potentially blocks on batch verification. This would simplify the other side of the channel because it would no longer need to write ctx.Err() back to all the channels (which racy anyway and can leave these goroutines wedged).

[nalepae]

Why not using

defer dv.recordResult(RequireSidecarKzgProofVerified, &err)

directly in s.validateWithKzgBatchVerifier as done for all other verification functions?

[kasey]

I think you could simplify the pair of methods kzgVerifierRoutine + pullKzgChan into a single method like the following. Also note that I have dropped this kzg.resChan <- s.ctx.Err() loop - I think this is not necessary if the callers use a select and a context.WithTimeout (which they should do anyway for safe concurrency).

// A routine that runs in the background to perform batch
// KZG verifications by draining the channel and processing all pending requests.
func (s *Service) kzgVerifierRoutine() {
    kzgBatch := make([]*kzgVerifier, 0, 1)
    for {
		select {
			case <-s.ctx.Done():
				return
			case kzg := <-s.kzgChan:
				kzgBatch = append(kzgBatch, kzg)
				continue
			default:
				verifyKzgBatch(kzgBatch)
				kzgBatch = make([]*kzgVerifier, 0, 1)
		}
	}
}

(edit: fixed a bug in the suggestion)

[kasey]

You don't really need to close channels, they get cleaned up by the garbage collector. You should only close a channel on the sending side, to unblock any receivers who could be waiting on more values from a sender which isn't going to produce any more. Receiving from a closed channel is fine (instantly unblocks any goroutine trying to receive), but sending to one /causes a panic/. This feedback will be more critical if you take the other piece of feedback to check ctx.Done() above, because in that instance you won't have read from resChan, so the other side could easily send to it after it is closed here and panic.

[kasey]

nitpick: Why a new variable and not just err = errors.Wrapf(err, "Could not verify")?

[kasey]

Should we have more than 1 goroutine doing this?