Skip to content

Commit

Permalink
Revert PIP-322 changes in DispatchRateLimiter
Browse files Browse the repository at this point in the history
  • Loading branch information
@lhotari
lhotari committed Feb 24, 2025
1 parent 0719aaa commit ecee9b6
Show file tree
Hide file tree
Showing 4 changed files with 616 additions and 38 deletions.
92 changes: 54 additions & 38 deletions ...broker/src/main/java/org/apache/pulsar/broker/service/persistent/DispatchRateLimiter.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -21,14 +21,14 @@
import java.util.Optional;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.TimeUnit;
import java.util.function.Supplier;
import org.apache.pulsar.broker.ServiceConfiguration;
import org.apache.pulsar.broker.qos.AsyncTokenBucket;
import org.apache.pulsar.broker.qos.AsyncTokenBucketBuilder;
import org.apache.pulsar.broker.service.BrokerService;
import org.apache.pulsar.common.naming.NamespaceName;
import org.apache.pulsar.common.naming.TopicName;
import org.apache.pulsar.common.policies.data.DispatchRate;
import org.apache.pulsar.common.policies.data.Policies;
import org.apache.pulsar.common.util.RateLimiter;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

Expand All @@ -46,8 +46,8 @@ public enum Type {
private final Type type;

private final BrokerService brokerService;
private volatile AsyncTokenBucket dispatchRateLimiterOnMessage;
private volatile AsyncTokenBucket dispatchRateLimiterOnByte;
private volatile RateLimiter dispatchRateLimiterOnMessage;
private volatile RateLimiter dispatchRateLimiterOnByte;

public DispatchRateLimiter(PersistentTopic topic, Type type) {
this(topic, null, type);
Expand Down Expand Up @@ -77,9 +77,9 @@ public DispatchRateLimiter(BrokerService brokerService) {
* @return
*/
public long getAvailableDispatchRateLimitOnMsg() {
AsyncTokenBucket localDispatchRateLimiterOnMessage = dispatchRateLimiterOnMessage;
RateLimiter localDispatchRateLimiterOnMessage = dispatchRateLimiterOnMessage;
return localDispatchRateLimiterOnMessage == null ? -1 :
Math.max(localDispatchRateLimiterOnMessage.getTokens(), 0);
Math.max(localDispatchRateLimiterOnMessage.getAvailablePermits(), 0);
}

/**
Expand All @@ -88,8 +88,9 @@ public long getAvailableDispatchRateLimitOnMsg() {
* @return
*/
public long getAvailableDispatchRateLimitOnByte() {
AsyncTokenBucket localDispatchRateLimiterOnByte = dispatchRateLimiterOnByte;
return localDispatchRateLimiterOnByte == null ? -1 : Math.max(localDispatchRateLimiterOnByte.getTokens(), 0);
RateLimiter localDispatchRateLimiterOnByte = dispatchRateLimiterOnByte;
return localDispatchRateLimiterOnByte == null ? -1 :
Math.max(localDispatchRateLimiterOnByte.getAvailablePermits(), 0);
}

/**
Expand All @@ -99,13 +100,13 @@ public long getAvailableDispatchRateLimitOnByte() {
* @param byteSize
*/
public void consumeDispatchQuota(long numberOfMessages, long byteSize) {
AsyncTokenBucket localDispatchRateLimiterOnMessage = dispatchRateLimiterOnMessage;
RateLimiter localDispatchRateLimiterOnMessage = dispatchRateLimiterOnMessage;
if (numberOfMessages > 0 && localDispatchRateLimiterOnMessage != null) {
localDispatchRateLimiterOnMessage.consumeTokens(numberOfMessages);
localDispatchRateLimiterOnMessage.tryAcquire(numberOfMessages);
}
AsyncTokenBucket localDispatchRateLimiterOnByte = dispatchRateLimiterOnByte;
RateLimiter localDispatchRateLimiterOnByte = dispatchRateLimiterOnByte;
if (byteSize > 0 && localDispatchRateLimiterOnByte != null) {
localDispatchRateLimiterOnByte.consumeTokens(byteSize);
localDispatchRateLimiterOnByte.tryAcquire(byteSize);
}
}

Expand Down Expand Up @@ -221,50 +222,63 @@ public synchronized void updateDispatchRate(DispatchRate dispatchRate) {

long msgRate = dispatchRate.getDispatchThrottlingRateInMsg();
long byteRate = dispatchRate.getDispatchThrottlingRateInByte();
long ratePeriodNanos = TimeUnit.SECONDS.toNanos(Math.max(dispatchRate.getRatePeriodInSecond(), 1));
long ratePeriod = dispatchRate.getRatePeriodInSecond();

Supplier<Long> permitUpdaterMsg = dispatchRate.isRelativeToPublishRate()
? () -> getRelativeDispatchRateInMsg(dispatchRate)
: null;
// update msg-rateLimiter
if (msgRate > 0) {
if (dispatchRate.isRelativeToPublishRate()) {
if (this.dispatchRateLimiterOnMessage == null) {
this.dispatchRateLimiterOnMessage =
configureAsyncTokenBucket(AsyncTokenBucket.builderForDynamicRate())
.rateFunction(() -> getRelativeDispatchRateInMsg(dispatchRate))
.ratePeriodNanosFunction(() -> ratePeriodNanos)
RateLimiter.builder()
.scheduledExecutorService(brokerService.pulsar().getExecutor())
.permits(msgRate)
.rateTime(ratePeriod)
.timeUnit(TimeUnit.SECONDS)
.permitUpdater(permitUpdaterMsg)
.isDispatchOrPrecisePublishRateLimiter(true)
.build();
} else {
this.dispatchRateLimiterOnMessage =
configureAsyncTokenBucket(AsyncTokenBucket.builder())
.rate(msgRate).ratePeriodNanos(ratePeriodNanos)
.build();
this.dispatchRateLimiterOnMessage.setRate(msgRate, dispatchRate.getRatePeriodInSecond(),
TimeUnit.SECONDS, permitUpdaterMsg);
}
} else {
this.dispatchRateLimiterOnMessage = null;
// message-rate should be disable and close
if (this.dispatchRateLimiterOnMessage != null) {
this.dispatchRateLimiterOnMessage.close();
this.dispatchRateLimiterOnMessage = null;
}
}

Supplier<Long> permitUpdaterByte = dispatchRate.isRelativeToPublishRate()
? () -> getRelativeDispatchRateInByte(dispatchRate)
: null;
// update byte-rateLimiter
if (byteRate > 0) {
if (dispatchRate.isRelativeToPublishRate()) {
if (this.dispatchRateLimiterOnByte == null) {
this.dispatchRateLimiterOnByte =
configureAsyncTokenBucket(AsyncTokenBucket.builderForDynamicRate())
.rateFunction(() -> getRelativeDispatchRateInByte(dispatchRate))
.ratePeriodNanosFunction(() -> ratePeriodNanos)
RateLimiter.builder()
.scheduledExecutorService(brokerService.pulsar().getExecutor())
.permits(byteRate)
.rateTime(ratePeriod)
.timeUnit(TimeUnit.SECONDS)
.permitUpdater(permitUpdaterByte)
.isDispatchOrPrecisePublishRateLimiter(true)
.build();
} else {
this.dispatchRateLimiterOnByte =
configureAsyncTokenBucket(AsyncTokenBucket.builder())
.rate(byteRate).ratePeriodNanos(ratePeriodNanos)
.build();
this.dispatchRateLimiterOnByte.setRate(byteRate, dispatchRate.getRatePeriodInSecond(),
TimeUnit.SECONDS, permitUpdaterByte);
}
} else {
this.dispatchRateLimiterOnByte = null;
// message-rate should be disable and close
if (this.dispatchRateLimiterOnByte != null) {
this.dispatchRateLimiterOnByte.close();
this.dispatchRateLimiterOnByte = null;
}
}
}

private <T extends AsyncTokenBucketBuilder<T>> T configureAsyncTokenBucket(T builder) {
builder.clock(brokerService.getPulsar().getMonotonicSnapshotClock());
return builder;
}

private long getRelativeDispatchRateInMsg(DispatchRate dispatchRate) {
return (topic != null && dispatchRate != null)
? (long) topic.getLastUpdatedAvgPublishRateInMsg() + dispatchRate.getDispatchThrottlingRateInMsg()
Expand All @@ -283,7 +297,7 @@ private long getRelativeDispatchRateInByte(DispatchRate dispatchRate) {
* @return
*/
public long getDispatchRateOnMsg() {
AsyncTokenBucket localDispatchRateLimiterOnMessage = dispatchRateLimiterOnMessage;
RateLimiter localDispatchRateLimiterOnMessage = dispatchRateLimiterOnMessage;
return localDispatchRateLimiterOnMessage != null ? localDispatchRateLimiterOnMessage.getRate() : -1;
}

Expand All @@ -293,7 +307,7 @@ public long getDispatchRateOnMsg() {
* @return
*/
public long getDispatchRateOnByte() {
AsyncTokenBucket localDispatchRateLimiterOnByte = dispatchRateLimiterOnByte;
RateLimiter localDispatchRateLimiterOnByte = dispatchRateLimiterOnByte;
return localDispatchRateLimiterOnByte != null ? localDispatchRateLimiterOnByte.getRate() : -1;
}

Expand All @@ -306,9 +320,11 @@ public static boolean isDispatchRateEnabled(DispatchRate dispatchRate) {
public void close() {
// close rate-limiter
if (dispatchRateLimiterOnMessage != null) {
dispatchRateLimiterOnMessage.close();
dispatchRateLimiterOnMessage = null;
}
if (dispatchRateLimiterOnByte != null) {
dispatchRateLimiterOnByte.close();
dispatchRateLimiterOnByte = null;
}
}
Expand Down
26 changes: 26 additions & 0 deletions pulsar-common/src/main/java/org/apache/pulsar/common/util/RateLimitFunction.java
View file
Original file line number Diff line number Diff line change
@@ -0,0 +1,26 @@
/*
* Licensed to the Apache Software Foundation (ASF) under one
* or more contributor license agreements. See the NOTICE file
* distributed with this work for additional information
* regarding copyright ownership. The ASF licenses this file
* to you under the Apache License, Version 2.0 (the
* "License"); you may not use this file except in compliance
* with the License. You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing,
* software distributed under the License is distributed on an
* "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
* KIND, either express or implied. See the License for the
* specific language governing permissions and limitations
* under the License.
*/
package org.apache.pulsar.common.util;

/**
* Function use when rate limiter renew permit.
* */
public interface RateLimitFunction {
void apply();
}
Loading

0 comments on commit ecee9b6

Please sign in to comment.