Throttler: How does the target-side and source-side components of OnlineDDL request the services of the throttler?

[newborn22]

Throttler: How does the target-side and source-side components of OnlineDDL request the services of the throttler?

[newborn22]

source:

rowStreamer: in streamQuery function

for {
	...
	// check throttler.
	if !rs.vse.throttlerClient.ThrottleCheckOKOrWait(rs.ctx) {
		rs.throttleResponseRateLimiter.Do(func() error {
			return safeSend(&binlogdatapb.VStreamRowsResponse{Throttled: true})
		})
		continue
	}
	...
}

vstreamer: in parseEvents function

for {
	...
	// check throttler.
	if !vs.vse.throttlerClient.ThrottleCheckOKOrWait(ctx) {
		// make sure to leave if context is cancelled
		select {
		case <-ctx.Done():
			return
		default:
			// do nothing special
			throttledHeartbeatsRateLimiter.Do(func() error {
				return injectHeartbeat(true)
		})
		// we won't process events, until we're no longer throttling
		continue
	}
	...
}

target:

vcopier: in copyTable function

for{
	...
	// verify throttler is happy, otherwise keep looping
	if vc.vr.vre.throttlerClient.ThrottleCheckOKOrWaitAppName(ctx, vc.throttlerAppName) {
		break // out of 'for' loop
	} else { // we're throttled
		_ = vc.vr.updateTimeThrottled(VCopierComponentName)
	}
	...
}

vplayer: in applyEvents function

for {
	...
	if ctx.Err() != nil {
		return ctx.Err()
	}
	// check throttler.
	if !vp.vr.vre.throttlerClient.ThrottleCheckOKOrWaitAppName(ctx, vp.throttlerAppName) {
		_ = vp.vr.updateTimeThrottled(VPlayerComponentName)
		continue
	}
	...
}

As you can see, both the target and source side call the trottler in a loop and only jump out of the loop when it returns ok. In addition, the source side will send a heartbeat packet when the throttler returns non-ok, which is supposed to inform the other side that the connection is not broken, but just can't send data temporarily due to throttling. The relationship among the functions about throttler request mentioned above is as follows, you can find that they finally call the function CheckByType.

ThrottleCheckOKOrWait(ctx)->ThrottleCheckOKOrWaitAppName(ctx,"")
ThrottleCheckOKOrWaitAppName->((c *Client))ThrottleCheckOK->(throttler *Throttler)CheckByType