213
// Updates the API-configured RPS based on the latest user data
214
// and applies the new rate limit if the effective RPS has changed.
216
>
r.mu.Lock()
217
>
defer r.mu.Unlock()
218
>
// Fetch the latest user data and update the API-configured RPS.
219
>
r.trySetRPSFromUserDataLocked()
220
>
r.computeAndApplyRateLimitLocked()
221
>
}
222
223
// trySetRPSFromUserDataLocked sets the apiConfigRPS from user data.
224
// Called exclusively in response to updates in user data.
226
>
userData, _, err := r.userDataManager.GetUserData()
227
>
if err != nil {
228
return
229
}
230
>
config := userData.GetData().GetPerType()[int32(r.taskQueueType)].GetConfig()
ratelimit_manager.go
231
>
// If rate limit is an empty message, it means rate limit could have been unset via API.
232
>
// In this case, the apiConfigRPS will need to be unset.
233
>
queueRateLimit := config.GetQueueRateLimit()
234
>
if queueRateLimit.GetRateLimit() == nil {
235
>
r.apiConfigRPS = nil
236
>
} else {
237
val := float64(queueRateLimit.GetRateLimit().GetRequestsPerSecond())
238
r.apiConfigRPS = &val
239
}
241
>
if fairnessKeyRateLimitDefault.GetRateLimit() == nil {
242
r.fairnessKeyRateLimitDefault = nil
244
// Maintain the fairnessKeyRateLimitDefault as per-partition rate, scaled by the same
245
// fraction applied to the whole-queue effectiveRPS.